'as' becomes a keyword when with statement activated
"""plan execution of rql queries on multiple sourcesthe best way to understand what are we trying to acheive here is to readthe unit-tests in unittest_querier_planner.pySplit and execution specifications~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~For a system source and a ldap user source (only EUser and its attributesis supported, no group or such)::EUser X:1. fetch EUser X from both sources and return concatenation of results:EUser X WHERE X in_group G, G name 'users':* catch 1 1. fetch EUser X from both sources, store concatenation of results into a temporary table 2. return the result of TMP X WHERE X in_group G, G name 'users' from the system source* catch 2 1. return the result of EUser X WHERE X in_group G, G name 'users' from system source, that's enough (optimization of the sql querier will avoid join on EUser, so we will directly get local eids):EUser X,L WHERE X in_group G, X login L, G name 'users':1. fetch Any X,L WHERE X is EUser, X login L from both sources, store concatenation of results into a temporary table2. return the result of Any X, L WHERE X is TMP, X login LX in_group G, G name 'users' from the system source:Any X WHERE X owned_by Y:* catch 1 1. fetch EUser X from both sources, store concatenation of results into a temporary table 2. return the result of Any X WHERE X owned_by Y, Y is TMP from the system source* catch 2 1. return the result of Any X WHERE X owned_by Y from system source, that's enough (optimization of the sql querier will avoid join on EUser, so we will directly get local eids):organization: Logilab:copyright: 2003-2008 LOGILAB S.A. (Paris, FRANCE), all rights reserved.:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr"""__docformat__="restructuredtext en"fromcopyimportdeepcopyfromitertoolsimportimap,ifilterfalsefromlogilab.common.compatimportanyfromlogilab.common.decoratorsimportcachedfromrql.stmtsimportUnion,Selectfromrql.nodesimportVariableRef,Comparison,Relation,Constant,Exists,Variablefromcubicwebimportserverfromcubicweb.common.utilsimportmake_uidfromcubicweb.server.utilsimportcleanup_solutionsfromcubicweb.server.ssplannerimportSSPlanner,OneFetchStep,add_types_restrictionfromcubicweb.server.msstepsimport*fromcubicweb.server.sourcesimportAbstractSourceVariable._ms_table_key=lambdax:x.nameRelation._ms_table_key=lambdax:x.r_type# str() Constant.value to ensure generated table name won't be unicodeConstant._ms_table_key=lambdax:str(x.value)AbstractSource.dont_cross_relations=()AbstractSource.cross_relations=()defallequals(solutions):"""return true if all solutions are identical"""sol=solutions.next()forsol_insolutions:ifsol_!=sol:returnFalsereturnTruedefneed_aggr_step(select,sources,stepdefs=None):"""return True if a temporary table is necessary to store some partial results to execute the given query """iflen(sources)==1:# can do everything at once with a single sourcereturnFalseifselect.orderbyorselect.groupbyorselect.has_aggregat:# if more than one source, we need a temp table to deal with sort /# groups / aggregat if :# * the rqlst won't be splitted (in the other case the last query# using partial temporary table can do sort/groups/aggregat without# the need for a later AggrStep)# * the rqlst is splitted in multiple steps and there are more than one# final stepifstepdefsisNone:returnTruehas_one_final=Falsefstepsolindices=set()forstepdefinstepdefs:ifstepdef[-1]:ifhas_one_finalorfrozenset(stepdef[2])!=fstepsolindices:returnTruehas_one_final=Trueelse:fstepsolindices.update(stepdef[2])returnFalsedefcopy_node(newroot,node,subparts=()):newnode=node.__class__(*node.initargs(newroot))forpartinsubparts:newnode.append(part)returnnewnodedefsame_scope(var):"""return true if the variable is always used in the same scope"""try:returnvar.stinfo['samescope']exceptKeyError:forrelinvar.stinfo['relations']:ifnotrel.scopeisvar.scope:var.stinfo['samescope']=FalsereturnFalsevar.stinfo['samescope']=TruereturnTruedefselect_group_sort(select):# XXX something similar done in rql2sql# add variables used in groups and sort terms to the selection# if necessaryifselect.groupby:forvrefinselect.groupby:ifnotvrefinselect.selection:select.append_selected(vref.copy(select))forsortterminselect.orderby:forvrefinsortterm.iget_nodes(VariableRef):ifnotvrefinselect.get_selected_variables():# we can't directly insert sortterm.term because it references# a variable of the select before the copy.# XXX if constant term are used to define sort, their value# may necessite a decayselect.append_selected(vref.copy(select))ifselect.groupbyandnotvrefinselect.groupby:select.add_group_var(vref.copy(select))classPartPlanInformation(object):"""regroups necessary information to execute some part of a "global" rql query ("global" means as received by the querier, which may result in several internal queries, e.g. parts, due to security insertions) it exposes as well some methods helping in executing this part on a multi-sources repository, modifying its internal structure during the process :attr solutions: a list of mappings (varname -> vartype) :attr sourcesvars: a dictionnary telling for each source which variable/solution are supported, of the form {source : {varname: [solution index, ]}} """def__init__(self,plan,rqlst,rqlhelper=None):self.needsplit=Falseself.temptable=Noneself.finaltable=Noneself.plan=planself.rqlst=rqlstself._session=plan.sessionself._solutions=rqlst.solutionsself._solindices=range(len(self._solutions))# source : {var: [solution index, ]}self.sourcesvars=self._sourcesvars={}# source : {relation: set(child variable and constant)}self._crossrelations={}# dictionnary of variables which are linked to each other using a non# final relation which is supported by multiple sourcesself._linkedvars={}self._crosslinkedvars={}# processingself._compute_sourcesvars()self._remove_invalid_sources()self._compute_needsplit()self.sourcesvars={}fork,vinself._sourcesvars.iteritems():self.sourcesvars[k]={}fork2,v2inv.iteritems():self.sourcesvars[k][k2]=v2.copy()self._inputmaps={}ifrqlhelperisnotNone:# else testself._insert_identity_variable=rqlhelper._annotator.rewrite_shared_optionaldefcopy_solutions(self,solindices):return[self._solutions[solidx].copy()forsolidxinsolindices]@property@cacheddefpart_sources(self):ifself._sourcesvars:returntuple(sorted(self._sourcesvars))return(self._session.repo.system_source,)@property@cacheddef_sys_source_set(self):returnfrozenset((self._session.repo.system_source,solindex)forsolindexinself._solindices)@cacheddef_norel_support_set(self,relation):"""return a set of (source, solindex) where source doesn't support the relation """returnfrozenset((source,solidx)forsourceinself._session.repo.sourcesforsolidxinself._solindicesifnot((source.support_relation(relation.r_type)andnotself.crossed_relation(source,relation))orrelation.r_typeinsource.dont_cross_relations))def_compute_sourcesvars(self):"""compute for each variable/solution in the rqlst which sources support them """repo=self._session.repoeschema=repo.schema.eschemasourcesvars=self._sourcesvars# find for each source which variable/solution are supportedforvarname,varobjinself.rqlst.defined_vars.items():# if variable has an eid specified, we can get its source directly# NOTE: use uidrels and not constnode to deal with "X eid IN(1,2,3,4)"ifvarobj.stinfo['uidrels']:vrels=varobj.stinfo['relations']-varobj.stinfo['uidrels']forrelinvarobj.stinfo['uidrels']:ifrel.neged(strict=True)orrel.operator()!='=':continueforconstinrel.children[1].get_nodes(Constant):eid=const.eval(self.plan.args)source=self._session.source_from_eid(eid)ifvrelsandnotany(source.support_relation(r.r_type)forrinvrels):self._set_source_for_var(repo.system_source,varobj)else:self._set_source_for_var(source,varobj)continuerels=varobj.stinfo['relations']ifnotrelsandnotvarobj.stinfo['typerels']:# (rare) case where the variable has no type specified nor# relation accessed ex. "Any MAX(X)"self._set_source_for_var(repo.system_source,varobj)continuefori,solinenumerate(self._solutions):vartype=sol[varname]# skip final variableifeschema(vartype).is_final():breakforsourceinrepo.sources:ifsource.support_entity(vartype):# the source support the entity type, though we will# actually have to fetch from it only if# * the variable isn't invariant# * at least one supported relation specifiedifnotvarobj._q_invariantor \any(imap(source.support_relation,(r.r_typeforrinrelsifr.r_type!='eid'))):sourcesvars.setdefault(source,{}).setdefault(varobj,set()).add(i)# if variable is not invariant and is used by a relation# not supported by this source, we'll have to split the# queryifnotvarobj._q_invariantandany(ifilterfalse(source.support_relation,(r.r_typeforrinrels))):self.needsplit=Truedef_handle_cross_relation(self,rel,relsources,vsources):crossvars=Noneforsourceinrelsources:ifrel.r_typeinsource.cross_relations:crossvars=set(x.variableforxinrel.get_nodes(VariableRef))crossvars.update(frozenset(xforxinrel.get_nodes(Constant)))assertlen(crossvars)==2ssource=self._session.repo.system_sourceneedsplit=Trueflag=0forvincrossvars:ifisinstance(v,Constant):self._sourcesvars[ssource][v]=set(self._solindices)iflen(vsources[v])==1:ifiter(vsources[v]).next()[0].uri=='system':flag=1forovincrossvars:ifovisnotvandov._q_invariant:ssset=frozenset((ssource,))self._remove_sources(ov,vsources[ov]-ssset)else:forovincrossvars:ifovisnotvandov._q_invariant:needsplit=Falsebreakelse:continueifnotrel.neged(strict=True):breakelse:self._crossrelations.setdefault(source,{})[rel]=crossvarsifnotflag:self._sourcesvars.setdefault(source,{})[rel]=set(self._solindices)self._sourcesvars.setdefault(ssource,{})[rel]=set(self._solindices)ifneedsplit:self.needsplit=TruereturncrossvarsisNonedef_remove_invalid_sources(self):"""removes invalid sources from `sourcesvars` member according to traversed relations and their properties (which sources support them, can they cross sources, etc...) """repo=self._session.reporschema=repo.schema.rschemavsources={}forrelinself.rqlst.iget_nodes(Relation):# process non final relations only# note: don't try to get schema for 'is' relation (not available# during bootstrap)ifnotrel.is_types_restriction()andnotrschema(rel.r_type).is_final():# nothing to do if relation is not supported by multiple sources# or if some source has it listed in its cross_relations# attribute## XXX code below don't deal if some source allow relation# crossing but not another onerelsources=repo.rel_type_sources(rel.r_type)crossvars=Noneiflen(relsources)<2:# filter out sources being there because they have this# relation in their dont_cross_relations attributerelsources=[sourceforsourceinrelsourcesifsource.support_relation(rel.r_type)]ifrelsources:# this means the relation is using a variable inlined as# a constant and another unsupported variable, in which# case we put the relation in sourcesvarsself._sourcesvars.setdefault(relsources[0],{})[rel]=set(self._solindices)continuelhs,rhs=rel.get_variable_parts()lhsv,rhsv=getattr(lhs,'variable',lhs),getattr(rhs,'variable',rhs)# update dictionnary of sources supporting lhs and rhs varsifnotlhsvinvsources:vsources[lhsv]=self._term_sources(lhs)ifnotrhsvinvsources:vsources[rhsv]=self._term_sources(rhs)ifself._handle_cross_relation(rel,relsources,vsources):self._linkedvars.setdefault(lhsv,set()).add((rhsv,rel))self._linkedvars.setdefault(rhsv,set()).add((lhsv,rel))else:self._crosslinkedvars.setdefault(lhsv,set()).add((rhsv,rel))self._crosslinkedvars.setdefault(rhsv,set()).add((lhsv,rel))forterminself._linkedvars:self._remove_sources_until_stable(term,vsources)iflen(self._sourcesvars)>1andhasattr(self.plan.rqlst,'main_relations'):# the querier doesn't annotate write queries, need to do it hereself.plan.annotate_rqlst()# insert/update/delete queries, we may get extra information from# the main relation (eg relations to the left of the WHEREifself.plan.rqlst.TYPE=='insert':inserted=dict((vref.variable,etype)foretype,vrefinself.plan.rqlst.main_variables)else:inserted={}forrelinself.plan.rqlst.main_relations:ifnotrschema(rel.r_type).is_final():# nothing to do if relation is not supported by multiple sourcesiflen(repo.rel_type_sources(rel.r_type))<2:continuelhs,rhs=rel.get_variable_parts()try:lhsv=self._extern_term(lhs,vsources,inserted)rhsv=self._extern_term(rhs,vsources,inserted)exceptKeyError,ex:continuenorelsup=self._norel_support_set(rel)self._remove_var_sources(lhsv,norelsup,rhsv,vsources)self._remove_var_sources(rhsv,norelsup,lhsv,vsources)# cleanup linked varforvar,linkedrelsinfoinself._linkedvars.iteritems():self._linkedvars[var]=frozenset(x[0]forxinlinkedrelsinfo)# if there are other sources than the system source, consider simplified# variables'sourceifself._sourcesvarsandself._sourcesvars.keys()!=[self._session.repo.system_source]:# add source for rewritten constants to sourcesvarsforvconstsinself.rqlst.stinfo['rewritten'].itervalues():const=vconsts[0]eid=const.eval(self.plan.args)source=self._session.source_from_eid(eid)ifsourceisself._session.repo.system_source:forconstinvconsts:self._set_source_for_var(source,const)elifsourceinself._sourcesvars:source_scopes=frozenset(v.scopeforvinself._sourcesvars[source])forconstinvconsts:ifconst.scopeinsource_scopes:self._set_source_for_var(source,const)def_extern_term(self,term,vsources,inserted):var=term.variableifvar.stinfo['constnode']:termv=var.stinfo['constnode']vsources[termv]=self._term_sources(termv)elifvarininserted:termv=varsource=self._session.repo.locate_etype_source(inserted[var])vsources[termv]=set((source,solindex)forsolindexinself._solindices)else:termv=self.rqlst.defined_vars[var.name]ifnottermvinvsources:vsources[termv]=self._term_sources(termv)returntermvdef_remove_sources_until_stable(self,var,vsources):sourcesvars=self._sourcesvarsforovar,relinself._linkedvars.get(var,()):ifnotvar.scopeisovar.scopeandrel.scope.neged(strict=True):# can't get information from relation inside a NOT exists# where variables don't belong to the same scopecontinueifnot(var.scopeisrel.scopeandovar.scopeisrel.scope)andrel.ored():continuerelsources=self._session.repo.rel_type_sources(rel.r_type)ifrel.neged(strict=True)and(len(relsources)<2ornotisinstance(ovar,Variable)orovar.valuable_references()!=1orany(sourcesvars[source][var]!=sourcesvars[source][ovar]forsourceinrelsourcesifvarinsourcesvars.get(source,())andovarinsourcesvars.get(source,()))):# neged relation doesn't allow to infer variable sources unless we're# on a multisource relation for a variable only used by this relation# (eg "Any X WHERE NOT X multisource_rel Y" and over is Y), iif continuenorelsup=self._norel_support_set(rel)# compute invalid sources for variables and remove themself._remove_var_sources(var,norelsup,ovar,vsources)self._remove_var_sources(ovar,norelsup,var,vsources)def_remove_var_sources(self,var,norelsup,ovar,vsources):"""remove invalid sources for var according to ovar's sources and the relation between those two variables. """varsources=vsources[var]invalid_sources=varsources-(vsources[ovar]|norelsup)ifinvalid_sources:self._remove_sources(var,invalid_sources)varsources-=invalid_sourcesself._remove_sources_until_stable(var,vsources)def_compute_needsplit(self):"""tell according to sourcesvars if the rqlst has to be splitted for execution among multiple sources the execution has to be split if * a source support an entity (non invariant) but doesn't support a relation on it * a source support an entity which is accessed by an optional relation * there is more than one source and either all sources'supported variable/solutions are not equivalent or multiple variables have to be fetched from some source """# NOTE: < 2 since may be 0 on queries such as Any X WHERE X eid 2iflen(self._sourcesvars)<2:self.needsplit=Falseelifnotself.needsplit:ifnotallequals(self._sourcesvars.itervalues()):self.needsplit=Trueelse:sample=self._sourcesvars.itervalues().next()iflen(sample)>1andany(vforvinsampleifnotvinself._linkedvarsandnotvinself._crosslinkedvars):self.needsplit=Truedef_set_source_for_var(self,source,var):self._sourcesvars.setdefault(source,{})[var]=set(self._solindices)def_term_sources(self,term):"""returns possible sources for terms `term`"""ifisinstance(term,Constant):source=self._session.source_from_eid(term.eval(self.plan.args))returnset((source,solindex)forsolindexinself._solindices)else:var=getattr(term,'variable',term)sources=[sourceforsource,varobjsinself.sourcesvars.iteritems()ifvarinvarobjs]returnset((source,solindex)forsourceinsourcesforsolindexinself.sourcesvars[source][var])def_remove_sources(self,var,sources):"""removes invalid sources (`sources`) from `sourcesvars` :param sources: the list of sources to remove :param var: the analyzed variable """sourcesvars=self._sourcesvarsforsource,solindexinsources:try:sourcesvars[source][var].remove(solindex)exceptKeyError:return# may occur with subquery column aliasifnotsourcesvars[source][var]:delsourcesvars[source][var]ifnotsourcesvars[source]:delsourcesvars[source]defcrossed_relation(self,source,relation):returnrelationinself._crossrelations.get(source,())defpart_steps(self):"""precompute necessary part steps before generating actual rql for each step. This is necessary to know if an aggregate step will be necessary or not. """steps=[]select=self.rqlstrschema=self.plan.schema.rschemaforsourceinself.part_sources:sourcevars=self._sourcesvars[source]whilesourcevars:# take a variable randomly, and all variables supporting the# same solutionsvar,solindices=self._choose_var(sourcevars)ifsource.uri=='system':# ensure all variables are available for the latest step# (missing one will be available from temporary tables# of previous steps)scope=selectvariables=scope.defined_vars.values()+scope.aliases.values()sourcevars.clear()else:scope=var.scopevariables=self._expand_vars(var,source,sourcevars,scope,solindices)ifnotsourcevars:delself._sourcesvars[source]# find which sources support the same variables/solutionssources=self._expand_sources(source,variables,solindices)# suppose this is a final step until the contrary is provenfinal=scopeisselect# set of variables which should be additionaly selected when# possibleneedsel=set()# add attribute variables and mark variables which should be# additionaly selected when possibleforvarinselect.defined_vars.itervalues():ifnotvarinvariables:stinfo=var.stinfoforovar,rtypeinstinfo['attrvars']:ifovarinvariables:needsel.add(var.name)variables.append(var)breakelse:needsel.add(var.name)final=Falseiffinalandsource.uri!='system':# check rewritten constantsforvconstsinselect.stinfo['rewritten'].itervalues():const=vconsts[0]eid=const.eval(self.plan.args)_source=self._session.source_from_eid(eid)iflen(sources)>1ornot_sourceinsources:# if there is some rewriten constant used by a# not neged relation while there are some source# not supporting the associated entity, this step# can't be final (unless the relation is explicitly# in `variables`, eg cross relations)forcinvconsts:rel=c.relation()ifrelisNoneornot(relinvariablesorrel.neged(strict=True)):#if rel is not None and rel.r_type == 'identity' and not rel.neged(strict=True):final=Falsebreakbreak# check where all relations are supported by the sourcesforrelinscope.iget_nodes(Relation):ifrel.is_types_restriction():continue# take care not overwriting the existing "source" identifierfor_sourceinsources:ifnot_source.support_relation(rel.r_type):forvrefinrel.iget_nodes(VariableRef):needsel.add(vref.name)final=Falsebreakelifself.crossed_relation(_source,rel)andnotrelinvariables:final=Falsebreakelse:ifnotscopeisselect:self._exists_relation(rel,variables,needsel)# if relation is supported by all sources and some of# its lhs/rhs variable isn't in "variables", and the# other end *is* in "variables", mark it have to be# selectedifsource.uri!='system'andnotrschema(rel.r_type).is_final():lhs,rhs=rel.get_variable_parts()try:lhsvar=lhs.variableexceptAttributeError:lhsvar=lhstry:rhsvar=rhs.variableexceptAttributeError:rhsvar=rhsiflhsvarinvariablesandnotrhsvarinvariables:needsel.add(lhsvar.name)elifrhsvarinvariablesandnotlhsvarinvariables:needsel.add(rhsvar.name)iffinal:self._cleanup_sourcesvars(sources,solindices)# XXX rename: variables may contain Relation and Constant nodes...steps.append((sources,variables,solindices,scope,needsel,final))returnstepsdef_exists_relation(self,rel,variables,needsel):rschema=self.plan.schema.rschema(rel.r_type)lhs,rhs=rel.get_variable_parts()try:lhsvar,rhsvar=lhs.variable,rhs.variableexceptAttributeError:passelse:# supported relation with at least one end supported, check the# other end is in as well. If not this usually means the# variable is refed by an outer scope and should be substituted# using an 'identity' relation (else we'll get a conflict of# temporary tables)ifrhsvarinvariablesandnotlhsvarinvariables:self._identity_substitute(rel,lhsvar,variables,needsel)eliflhsvarinvariablesandnotrhsvarinvariables:self._identity_substitute(rel,rhsvar,variables,needsel)def_identity_substitute(self,relation,var,variables,needsel):newvar=self._insert_identity_variable(relation.scope,var)ifnewvarisnotNone:# ensure relation is using '=' operator, else we rely on a# sqlgenerator side effect (it won't insert an inequality operator# in this case)relation.children[1].operator='='variables.append(newvar)needsel.add(newvar.name)def_choose_var(self,sourcevars):secondchoice=Noneiflen(self._sourcesvars)>1:# priority to variable from subscopesforvarinsourcevars:ifnotvar.scopeisself.rqlst:ifisinstance(var,Variable):returnvar,sourcevars.pop(var)secondchoice=var,sourcevars.pop(var)else:# priority to variable outer scopeforvarinsourcevars:ifvar.scopeisself.rqlst:ifisinstance(var,Variable):returnvar,sourcevars.pop(var)secondchoice=varifsecondchoiceisnotNone:returnsecondchoice,sourcevars.pop(secondchoice)# priority to variableforvarinsourcevars:ifisinstance(var,Variable):returnvar,sourcevars.pop(var)# whatevervar=iter(sourcevars).next()returnvar,sourcevars.pop(var)def_expand_vars(self,var,source,sourcevars,scope,solindices):variables=[var]nbunlinked=1linkedvars=self._linkedvars# variable has to belong to the same scope if there is more# than the system source remainingiflen(self._sourcesvars)>1andnotscopeisself.rqlst:candidates=(vforvinsourcevars.keys()ifscopeisv.scope)else:candidates=sourcevars#.iterkeys()# we only want one unlinked variable in each generated querycandidates=[vforvincandidatesifisinstance(v,Constant)or(solindices.issubset(sourcevars[v])andvinlinkedvars)]accept_var=lambdax:(isinstance(x,Constant)orany(vforvinvariablesifvinlinkedvars.get(x,())))source_cross_rels=self._crossrelations.get(source,())ifisinstance(var,Relation)andvarinsource_cross_rels:cross_vars=source_cross_rels.pop(var)base_accept_var=accept_varaccept_var=lambdax:(base_accept_var(x)orxincross_vars)forrefedincross_vars:ifnotrefedincandidates:candidates.append(refed)else:cross_vars=()# repeat until no variable can't be added, since addition of a new# variable may permit to another one to be addedmodified=Truewhilemodifiedandcandidates:modified=Falseforvarincandidates[:]:ifaccept_var(var):variables.append(var)try:# constant nodes should be systematically deletedifisinstance(var,Constant):delsourcevars[var]else:# variable nodes should be deleted once all possible# solutions indices have been consumedsourcevars[var]-=solindicesifnotsourcevars[var]:delsourcevars[var]exceptKeyError:assertvarincross_varscandidates.remove(var)modified=Truereturnvariablesdef_expand_sources(self,selected_source,vars,solindices):sources=[selected_source]sourcesvars=self._sourcesvarsforsourceinsourcesvars:ifsourceisselected_source:continueforvarinvars:ifnot(varinsourcesvars[source]andsolindices.issubset(sourcesvars[source][var])):breakelse:sources.append(source)ifsource.uri!='system':forvarinvars:varsolindices=sourcesvars[source][var]varsolindices-=solindicesifnotvarsolindices:delsourcesvars[source][var]returnsourcesdef_cleanup_sourcesvars(self,sources,solindices):"""on final parts, remove solutions so we know they are already processed"""forsourceinsources:try:sourcevars=self._sourcesvars[source]exceptKeyError:continueforvar,varsolindicesinsourcevars.items():ifisinstance(var,Relation)andself.crossed_relation(source,var):continuevarsolindices-=solindicesifnotvarsolindices:delsourcevars[var]defmerge_input_maps(self,allsolindices):"""inputmaps is a dictionary with tuple of solution indices as key with an associateed input map as value. This function compute for each solution its necessary input map and return them grouped ex: inputmaps = {(0, 1, 2): {'A': 't1.login1', 'U': 't1.C0', 'U.login': 't1.login1'}, (1,): {'X': 't2.C0', 'T': 't2.C1'}} return : [([1], {'A': 't1.login1', 'U': 't1.C0', 'U.login': 't1.login1', 'X': 't2.C0', 'T': 't2.C1'}), ([0,2], {'A': 't1.login1', 'U': 't1.C0', 'U.login': 't1.login1'})] """ifnotself._inputmaps:return[(allsolindices,None)]mapbysol={}# compute a single map for each solutionforsolindices,basemapinself._inputmaps.iteritems():forsolindexinsolindices:solmap=mapbysol.setdefault(solindex,{})solmap.update(basemap)try:allsolindices.remove(solindex)exceptKeyError:continue# already removed# group results by identical input mapresult=[]forsolindex,solmapinmapbysol.iteritems():forsolindices,commonmapinresult:ifcommonmap==solmap:solindices.append(solindex)breakelse:result.append(([solindex],solmap))ifallsolindices:result.append((list(allsolindices),None))returnresultdefbuild_final_part(self,select,solindices,inputmap,sources,insertedvars):plan=self.planrqlst=plan.finalize(select,[self._solutions[i]foriinsolindices],insertedvars)ifself.temptableisNoneandself.finaltableisNone:returnOneFetchStep(plan,rqlst,sources,inputmap=inputmap)table=self.temptableorself.finaltablereturnFetchStep(plan,rqlst,sources,table,True,inputmap)defbuild_non_final_part(self,select,solindices,sources,insertedvars,table):"""non final step, will have to store results in a temporary table"""plan=self.planrqlst=plan.finalize(select,[self._solutions[i]foriinsolindices],insertedvars)step=FetchStep(plan,rqlst,sources,table,False)# update input map for following steps, according to processed solutionsinputmapkey=tuple(sorted(solindices))inputmap=self._inputmaps.setdefault(inputmapkey,{})inputmap.update(step.outputmap)plan.add_step(step)classMSPlanner(SSPlanner):"""MultiSourcesPlanner: build execution plan for rql queries decompose the RQL query according to sources'schema """defbuild_select_plan(self,plan,rqlst):"""build execution plan for a SELECT RQL query the rqlst should not be tagged at this point """ifserver.DEBUG:print'-'*80print'PLANNING',rqlstforselectinrqlst.children:iflen(select.solutions)>1:hasmultiplesols=Truebreakelse:hasmultiplesols=False# preprocess deals with security insertion and returns a new syntax tree# which have to be executed to fulfill the query: according# to permissions for variable's type, different rql queries may have to# be executedplan.preprocess(rqlst)ppis=[PartPlanInformation(plan,select,self.rqlhelper)forselectinrqlst.children]steps=self._union_plan(plan,rqlst,ppis)ifserver.DEBUG:frompprintimportpprintforstepinplan.steps:pprint(step.test_repr())pprint(steps[0].test_repr())returnstepsdef_ppi_subqueries(self,ppi):# part plan info for subqueriesplan=ppi.planinputmap={}forsubqueryinppi.rqlst.with_[:]:sppis=[PartPlanInformation(plan,select)forselectinsubquery.query.children]forsppiinsppis:ifsppi.needsplitorsppi.part_sources!=ppi.part_sources:temptable='T%s'%make_uid(id(subquery))sstep=self._union_plan(plan,subquery.query,sppis,temptable)[0]breakelse:sstep=NoneifsstepisnotNone:ppi.rqlst.with_.remove(subquery)fori,colaliasinenumerate(subquery.aliases):inputmap[colalias.name]='%s.C%s'%(temptable,i)ppi.plan.add_step(sstep)returninputmapdef_union_plan(self,plan,union,ppis,temptable=None):tosplit,cango,allsources=[],{},set()forplaninfoinppis:ifplaninfo.needsplit:tosplit.append(planinfo)else:cango.setdefault(planinfo.part_sources,[]).append(planinfo)forsourceinplaninfo.part_sources:allsources.add(source)# first add steps for query parts which doesn't need to splittedsteps=[]forsources,cppisincango.iteritems():byinputmap={}forppiincppis:select=ppi.rqlstifsources!=(plan.session.repo.system_source,):add_types_restriction(self.schema,select)# part plan info for subqueriesinputmap=self._ppi_subqueries(ppi)aggrstep=need_aggr_step(select,sources)ifaggrstep:atemptable='T%s'%make_uid(id(select))sunion=Union()sunion.append(select)selected=select.selection[:]select_group_sort(select)step=AggrStep(plan,selected,select,atemptable,temptable)step.set_limit_offset(select.limit,select.offset)select.limit=Noneselect.offset=0fstep=FetchStep(plan,sunion,sources,atemptable,True,inputmap)step.children.append(fstep)steps.append(step)else:byinputmap.setdefault(tuple(inputmap.iteritems()),[]).append((select))forinputmap,queriesinbyinputmap.iteritems():inputmap=dict(inputmap)sunion=Union()forselectinqueries:sunion.append(select)iftemptable:steps.append(FetchStep(plan,sunion,sources,temptable,True,inputmap))else:steps.append(OneFetchStep(plan,sunion,sources,inputmap))# then add steps for splitted query partsforplaninfointosplit:steps.append(self.split_part(planinfo,temptable))iflen(steps)>1:iftemptable:step=UnionFetchStep(plan)else:step=UnionStep(plan)step.children=stepsreturn(step,)returnsteps# internal methods for multisources decomposition #########################defsplit_part(self,ppi,temptable):ppi.finaltable=temptableplan=ppi.planselect=ppi.rqlstsubinputmap=self._ppi_subqueries(ppi)stepdefs=ppi.part_steps()ifneed_aggr_step(select,ppi.part_sources,stepdefs):atemptable='T%s'%make_uid(id(select))selection=select.selection[:]select_group_sort(select)else:atemptable=Noneselection=select.selectionppi.temptable=atemptablevfilter=VariablesFiltererVisitor(self.schema,ppi)steps=[]forsources,variables,solindices,scope,needsel,finalinstepdefs:# extract an executable query using only the specified variablesifsources[0].uri=='system':# in this case we have to merge input maps before call to# filter so already processed restriction are correctly# removedsolsinputmaps=ppi.merge_input_maps(solindices)forsolindices,inputmapinsolsinputmaps:minrqlst,insertedvars=vfilter.filter(sources,variables,scope,set(solindices),needsel,final)ifinputmapisNone:inputmap=subinputmapelse:inputmap.update(subinputmap)steps.append(ppi.build_final_part(minrqlst,solindices,inputmap,sources,insertedvars))else:# this is a final part (i.e. retreiving results for the# original query part) if all variable / sources have been# treated or if this is the last shot for used solutionsminrqlst,insertedvars=vfilter.filter(sources,variables,scope,solindices,needsel,final)iffinal:solsinputmaps=ppi.merge_input_maps(solindices)forsolindices,inputmapinsolsinputmaps:ifinputmapisNone:inputmap=subinputmapelse:inputmap.update(subinputmap)steps.append(ppi.build_final_part(minrqlst,solindices,inputmap,sources,insertedvars))else:table='_T%s%s'%(''.join(sorted(v._ms_table_key()forvinvariables)),''.join(sorted(str(i)foriinsolindices)))ppi.build_non_final_part(minrqlst,solindices,sources,insertedvars,table)# finally: join parts, deal with aggregat/group/sorts if necessaryifatemptableisnotNone:step=AggrStep(plan,selection,select,atemptable,temptable)step.children=stepseliflen(steps)>1:ifselect.need_intersect:iftemptable:step=IntersectFetchStep(plan)else:step=IntersectStep(plan)else:iftemptable:step=UnionFetchStep(plan)else:step=UnionStep(plan)step.children=stepselse:step=steps[0]ifselect.limitisnotNoneorselect.offset:step.set_limit_offset(select.limit,select.offset)returnstepclassUnsupportedBranch(Exception):passclassVariablesFiltererVisitor(object):def__init__(self,schema,ppi):self.schema=schemaself.ppi=ppiself.skip={}self.hasaggrstep=self.ppi.temptableself.extneedsel=frozenset(vref.nameforsortterminppi.rqlst.orderbyforvrefinsortterm.iget_nodes(VariableRef))def_rqlst_accept(self,rqlst,node,newroot,variables,setfunc=None):try:newrestr,node_=node.accept(self,newroot,variables[:])exceptUnsupportedBranch:returnrqlstifsetfuncisnotNoneandnewrestrisnotNone:setfunc(newrestr)ifnotnode_isnode:rqlst=node.parentreturnrqlstdeffilter(self,sources,variables,rqlst,solindices,needsel,final):ifserver.DEBUG:print'filter',finaland'final'or'',sources,variables,rqlst,solindices,needselnewroot=Select()self.sources=sorted(sources)self.variables=variablesself.solindices=solindicesself.final=final# variables which appear in unsupported branchesneedsel|=self.extneedselself.needsel=needsel# variables which appear in supported branchesself.mayneedsel=set()# new inserted variablesself.insertedvars=[]# other structures (XXX document)self.mayneedvar,self.hasvar={},{}self.use_only_defined=Falseself.scopes={rqlst:newroot}ifrqlst.where:rqlst=self._rqlst_accept(rqlst,rqlst.where,newroot,variables,newroot.set_where)ifisinstance(rqlst,Select):self.use_only_defined=Trueifrqlst.groupby:groupby=[]fornodeinrqlst.groupby:rqlst=self._rqlst_accept(rqlst,node,newroot,variables,groupby.append)ifgroupby:newroot.set_groupby(groupby)ifrqlst.having:having=[]fornodeinrqlst.having:rqlst=self._rqlst_accept(rqlst,node,newroot,variables,having.append)ifhaving:newroot.set_having(having)iffinalandrqlst.orderbyandnotself.hasaggrstep:orderby=[]fornodeinrqlst.orderby:rqlst=self._rqlst_accept(rqlst,node,newroot,variables,orderby.append)iforderby:newroot.set_orderby(orderby)self.process_selection(newroot,variables,rqlst)elifnotnewroot.where:# no restrictions have been copied, just select variables and add# type restriction (done later by add_types_restriction)forvinvariables:ifnotisinstance(v,Variable):continuenewroot.append_selected(VariableRef(newroot.get_variable(v.name)))solutions=self.ppi.copy_solutions(solindices)cleanup_solutions(newroot,solutions)newroot.set_possible_types(solutions)iffinal:ifself.hasaggrstep:self.add_necessary_selection(newroot,self.mayneedsel&self.extneedsel)newroot.distinct=rqlst.distinctelse:self.add_necessary_selection(newroot,self.mayneedsel&self.needsel)# insert vars to fetch constant values when neededfor(varname,rschema),reldefsinself.mayneedvar.iteritems():forrel,oredinreldefs:ifnot(varname,rschema)inself.hasvar:self.hasvar[(varname,rschema)]=None# just to avoid further insertioncvar=newroot.make_variable()forsolinnewroot.solutions:sol[cvar.name]=rschema.objects(sol[varname])[0]# if the current restriction is not used in a OR branch,# we can keep it, else we have to drop the constant# restriction (or we may miss some results)ifnotored:rel=rel.copy(newroot)newroot.add_restriction(rel)# add a relation to link the variablenewroot.remove_node(rel.children[1])cmp=Comparison('=')rel.append(cmp)cmp.append(VariableRef(cvar))self.insertedvars.append((varname,rschema,cvar.name))newroot.append_selected(VariableRef(newroot.get_variable(cvar.name)))# NOTE: even if the restriction is done by this query, we have# to let it in the original rqlst so that it appears anyway in# the "final" query, else we may change the meaning of the query# if there are NOT somewhere :# 'NOT X relation Y, Y name "toto"' means X WHERE X isn't related# to Y whose name is toto while# 'NOT X relation Y' means X WHERE X has no 'relation' (whatever Y)elifored:newroot.remove_node(rel)add_types_restriction(self.schema,rqlst,newroot,solutions)ifserver.DEBUG:print'--->',newrootreturnnewroot,self.insertedvarsdefvisit_and(self,node,newroot,variables):subparts=[]foriinxrange(len(node.children)):child=node.children[i]try:newchild,child_=child.accept(self,newroot,variables)ifnotchild_ischild:node=child_.parentifnewchildisNone:continuesubparts.append(newchild)exceptUnsupportedBranch:continueifnotsubparts:returnNone,nodeiflen(subparts)==1:returnsubparts[0],nodereturncopy_node(newroot,node,subparts),nodevisit_or=visit_anddef_relation_supported(self,relation):rtype=relation.r_typeforsourceinself.sources:ifnotsource.support_relation(rtype) \or(rtypeinsource.cross_relationsandnotrelationinself.variables):#self.ppi.crossed_relation(source, relation):returnFalseifnotself.final:rschema=self.schema.rschema(relation.r_type)ifnotrschema.is_final():forterminrelation.get_nodes((VariableRef,Constant)):term=getattr(term,'variable',term)termsources=sorted(set(x[0]forxinself.ppi._term_sources(term)))iftermsourcesandtermsources!=self.sources:returnFalsereturnTruedefvisit_relation(self,node,newroot,variables):ifnotnode.is_types_restriction():ifnodeinself.skipandself.solindices.issubset(self.skip[node]):ifnotself.schema.rschema(node.r_type).is_final():# can't really skip the relation if one variable is selected and only# referenced by this relationforvrefinnode.iget_nodes(VariableRef):stinfo=vref.variable.stinfoifstinfo['selected']andlen(stinfo['relations'])==1:breakelse:returnNone,nodeelse:returnNone,nodeifnotself._relation_supported(node):raiseUnsupportedBranch()# don't copy type restriction unless this is the only relation for the# rhs variable, else they'll be reinserted later as needed (else we may# copy a type restriction while the variable is not actually used)elifnotany(self._relation_supported(rel)forrelinnode.children[0].variable.stinfo['relations']):rel,node=self.visit_default(node,newroot,variables)returnrel,nodeelse:raiseUnsupportedBranch()rschema=self.schema.rschema(node.r_type)res=self.visit_default(node,newroot,variables)[0]ored=node.ored()ifrschema.is_final()orrschema.inlined:vrefs=node.children[1].get_nodes(VariableRef)ifnotvrefs:ifnotored:self.skip.setdefault(node,set()).update(self.solindices)else:self.mayneedvar.setdefault((node.children[0].name,rschema),[]).append((res,ored))else:assertlen(vrefs)==1vref=vrefs[0]# XXX check operator ?self.hasvar[(node.children[0].name,rschema)]=vrefifself._may_skip_attr_rel(rschema,node,vref,ored,variables,res):self.skip.setdefault(node,set()).update(self.solindices)elifnotored:self.skip.setdefault(node,set()).update(self.solindices)returnres,nodedef_may_skip_attr_rel(self,rschema,rel,vref,ored,variables,res):var=vref.variableifored:returnFalseifvar.nameinself.extneedselorvar.stinfo['selected']:returnFalseifnotsame_scope(var):returnFalseifany(vforv,_invar.stinfo['attrvars']ifnotv.nameinvariables):returnFalsereturnTruedefvisit_exists(self,node,newroot,variables):newexists=node.__class__()self.scopes={node:newexists}subparts,node=self._visit_children(node,newroot,variables)ifnotsubparts:returnNone,nodenewexists.set_where(subparts[0])returnnewexists,nodedefvisit_not(self,node,newroot,variables):subparts,node=self._visit_children(node,newroot,variables)ifnotsubparts:returnNone,nodereturncopy_node(newroot,node,subparts),nodedefvisit_group(self,node,newroot,variables):ifnotself.final:returnNone,nodereturnself.visit_default(node,newroot,variables)defvisit_variableref(self,node,newroot,variables):ifself.use_only_defined:ifnotnode.variable.nameinnewroot.defined_vars:raiseUnsupportedBranch(node.name)elifnotnode.variableinvariables:raiseUnsupportedBranch(node.name)self.mayneedsel.add(node.name)# set scope so we can insert types restriction properlynewvar=newroot.get_variable(node.name)newvar.stinfo['scope']=self.scopes.get(node.variable.scope,newroot)returnVariableRef(newvar),nodedefvisit_constant(self,node,newroot,variables):returncopy_node(newroot,node),nodedefvisit_default(self,node,newroot,variables):subparts,node=self._visit_children(node,newroot,variables)returncopy_node(newroot,node,subparts),nodevisit_comparison=visit_mathexpression=visit_constant=visit_function=visit_defaultvisit_sort=visit_sortterm=visit_defaultdef_visit_children(self,node,newroot,variables):subparts=[]foriinxrange(len(node.children)):child=node.children[i]newchild,child_=child.accept(self,newroot,variables)ifnotchildischild_:node=child_.parentifnewchildisnotNone:subparts.append(newchild)returnsubparts,nodedefprocess_selection(self,newroot,variables,rqlst):ifself.final:forterminrqlst.selection:newroot.append_selected(term.copy(newroot))forvrefinterm.get_nodes(VariableRef):self.needsel.add(vref.name)returnforterminrqlst.selection:vrefs=term.get_nodes(VariableRef)ifvrefs:supportedvars=[]forvrefinvrefs:var=vref.variableifvarinvariables:supportedvars.append(vref)continueelse:self.needsel.add(vref.name)breakelse:forvrefinvrefs:newroot.append_selected(vref.copy(newroot))supportedvars=[]forvrefinsupportedvars:ifnotvrefinnewroot.get_selected_variables():newroot.append_selected(VariableRef(newroot.get_variable(vref.name)))defadd_necessary_selection(self,newroot,variables):selected=tuple(newroot.get_selected_variables())forvarnameinvariables:var=newroot.defined_vars[varname]forvrefinvar.references():rel=vref.relation()ifrelisNoneandvrefinselected:# already selectedbreakelse:selvref=VariableRef(var)newroot.append_selected(selvref)ifnewroot.groupby:newroot.add_group_var(VariableRef(selvref.variable,noautoref=1))