cubicweb: comparison server/msplanner.py

equal deleted inserted replaced

-:054bb575c013
+:91ae10ffb611
 """plan execution of rql queries on multiple sources
-the best way to understand what are we trying to acheive here is to read
+the best way to understand what are we trying to acheive here is to read the
-the unit-tests in unittest_querier_planner.py
+unit-tests in unittest_msplanner.py
+What you need to know
-Split and execution specifications
+~~~~~~~~~~~~~~~~~~~~~
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+1. The system source is expected  to support every entity and relation types
+2. Given "X relation Y":
+* if relation, X and Y types are supported by the external source, we suppose
+by default that X and Y should both come from the same source as the
+relation. You can specify otherwise by adding relation into the
+"cross_relations" set in the source's mapping file and it that case, we'll
+consider that we can also find in the system source some relation between
+X and Y coming from different sources.
+* if "relation" isn't supported by the external source but X or Y
+types (or both) are, we suppose by default that can find in the system
+source some relation where X and/or Y come from the external source. You
+can specify otherwise by adding relation into the "dont_cross_relations"
+set in the source's mapping file and it that case, we'll consider that we
+can only find in the system source some relation between X and Y coming
+the system source.
+Implementation
+~~~~~~~~~~~~~~
+XXX explain algorithm
+Exemples of multi-sources query execution
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 For a system source and a ldap user source (only EUser and its attributes
 is 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
+1. fetch EUser X from both sources, store concatenation of results into a
-into a temporary table
+temporary table
-2. return the result of TMP X WHERE X in_group G, G name 'users' from
+2. return the result of TMP X WHERE X in_group G, G name 'users' from the
-the system source
+system source
 * catch 2
-1. return the result of EUser X WHERE X in_group G, G name 'users'
+1. return the result of EUser X WHERE X in_group G, G name 'users' from system
-from system source, that's enough (optimization of the sql querier
+source, that's enough (optimization of the sql querier will avoid join on
-will avoid join on EUser, so we will directly get local eids)
+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 table
 2. 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
+1. fetch EUser X from both sources, store concatenation of results into a
-into a temporary table
+temporary table
-2. return the result of Any X WHERE X owned_by Y, Y is TMP from
+2. return the result of Any X WHERE X owned_by Y, Y is TMP from the system
-the system source
+source
 * catch 2
-1. return the result of Any X WHERE X owned_by Y
+1. return the result of Any X WHERE X owned_by Y from system source, that's
-from system source, that's enough (optimization of the sql querier
+enough (optimization of the sql querier will avoid join on EUser, so we
-will avoid join on EUser, so we will directly get local eids)
+will directly get local eids)
 :organization: Logilab
 :copyright: 2003-2009 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
 :contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr
 from rql.nodes import VariableRef, Comparison, Relation, Constant, Exists, Variable
 from cubicweb import server
 from cubicweb.common.utils import make_uid
 from cubicweb.server.utils import cleanup_solutions
-from cubicweb.server.ssplanner import SSPlanner, OneFetchStep, add_types_restriction
+from cubicweb.server.ssplanner import (SSPlanner, OneFetchStep,
+add_types_restriction)
 from cubicweb.server.mssteps import *
 from cubicweb.server.sources import AbstractSource
 Variable._ms_table_key = lambda x: x.name
 Relation._ms_table_key = lambda x: x.r_type
 # str() Constant.value to ensure generated table name won't be unicode
 Constant._ms_table_key = lambda x: str(x.value)
 AbstractSource.dont_cross_relations = ()
 AbstractSource.cross_relations = ()
-def allequals(solutions):
-"""return true if all solutions are identical"""
-sol = solutions.next()
-for sol_ in solutions:
-if sol_ != sol:
-return False
-return True
 def need_aggr_step(select, sources, stepdefs=None):
 """return True if a temporary table is necessary to store some partial
 results to execute the given query
 """
 has_one_final = True
 else:
 fstepsolindices.update(stepdef[2])
 return False
+def select_group_sort(select): # XXX something similar done in rql2sql
+# add variables used in groups and sort terms to the selection
+# if necessary
+if select.groupby:
+for vref in select.groupby:
+if not vref in select.selection:
+select.append_selected(vref.copy(select))
+for sortterm in select.orderby:
+for vref in sortterm.iget_nodes(VariableRef):
+if not vref in select.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 decay
+select.append_selected(vref.copy(select))
+if select.groupby and not vref in select.groupby:
+select.add_group_var(vref.copy(select))
+def allequals(solutions):
+"""return true if all solutions are identical"""
+sol = solutions.next()
+noconstsol = None
+for sol_ in solutions:
+if sol_ != sol:
+return False
+return True
+# XXX move functions below to rql ##############################################
+def is_ancestor(n1, n2):
+p = n1.parent
+while p is not None:
+if p is n2:
+return True
+p = p.parent
+return False
 def copy_node(newroot, node, subparts=()):
 newnode = node.__class__(*node.initargs(newroot))
 for part in subparts:
 newnode.append(part)
 return newnode
 if not rel.scope is var.scope:
 var.stinfo['samescope'] = False
 return False
 var.stinfo['samescope'] = True
 return True
-def select_group_sort(select): # XXX something similar done in rql2sql
+################################################################################
-# add variables used in groups and sort terms to the selection
-# if necessary
-if select.groupby:
-for vref in select.groupby:
-if not vref in select.selection:
-select.append_selected(vref.copy(select))
-for sortterm in select.orderby:
-for vref in sortterm.iget_nodes(VariableRef):
-if not vref in select.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 decay
-select.append_selected(vref.copy(select))
-if select.groupby and not vref in select.groupby:
-select.add_group_var(vref.copy(select))
-# XXX move to rql
-def is_ancestor(n1, n2):
-p = n1.parent
-while p is not None:
-if p is n2:
-return True
-p = p.parent
-return False
 class PartPlanInformation(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)
+several internal queries, e.g. parts, due to security insertions). Actually
+a PPI is created for each subquery and for each query in a union.
-it exposes as well some methods helping in executing this part on a
+It exposes as well some methods helping in executing this part on a
 multi-sources repository, modifying its internal structure during the
-process
+process.
-:attr solutions: a list of mappings (varname -> vartype)
+:attr plan:
-:attr sourcesvars:
+the execution plan
-a dictionnary telling for each source which variable/solution are
+:attr rqlst:
-supported, of the form {source : {varname: [solution index, ]}}
+the original rql syntax tree handled by this part
+:attr needsplit:
+bool telling if the query has to be split into multiple steps for
+execution or if it can be executed at once
+:attr temptable:
+a SQL temporary table name or None, if necessary to handle aggregate /
+sorting for this part of the query
+:attr finaltable:
+a SQL table name or None, if results for this part of the query should be
+written into a temporary table (usually shared by multiple PPI)
+:attr sourcesterms:
+a dictionary {source : {term: set([solution index, ])}} telling for each
+source which terms are supported for which solutions. A "term" may be
+either a rql Variable, Constant or Relation node.
 """
 def __init__(self, plan, rqlst, rqlhelper=None):
+self.plan = plan
+self.rqlst = rqlst
 self.needsplit = False
 self.temptable = None
 self.finaltable = None
-self.plan = plan
+self._schema = plan.schema
-self.rqlst = rqlst
 self._session = plan.session
+self._repo = self._session.repo
 self._solutions = rqlst.solutions
 self._solindices = range(len(self._solutions))
-# source : {var: [solution index, ]}
+# source : {term: [solution index, ]}
-self.sourcesvars = self._sourcesvars = {}
+self.sourcesterms = self._sourcesterms = {}
 # source : {relation: set(child variable and constant)}
 self._crossrelations = {}
-# dictionnary of variables which are linked to each other using a non
+# dictionary of variables and constants which are linked to each other
-# final relation which is supported by multiple sources
+# using a non final relation supported by multiple sources (crossed or
-self._linkedvars = {}
+# not).
-self._crosslinkedvars = {}
+self._linkedterms = {}
 # processing
-self._compute_sourcesvars()
+termssources = self._compute_sourcesterms()
-self._remove_invalid_sources()
+self._remove_invalid_sources(termssources)
 self._compute_needsplit()
-self.sourcesvars = {}
+# after initialisation, .sourcesterms contains the same thing as
-for k, v in self._sourcesvars.iteritems():
+# ._sourcesterms though during plan construction, ._sourcesterms will
-self.sourcesvars[k] = {}
+# be modified while .sourcesterms will be kept unmodified
+self.sourcesterms = {}
+for k, v in self._sourcesterms.iteritems():
+self.sourcesterms[k] = {}
 for k2, v2 in v.iteritems():
-self.sourcesvars[k][k2] = v2.copy()
+self.sourcesterms[k][k2] = v2.copy()
+# cleanup linked var
+for var, linkedrelsinfo in self._linkedterms.iteritems():
+self._linkedterms[var] = frozenset(x[0] for x in linkedrelsinfo)
+# map output of a step to input of a following step
 self._inputmaps = {}
+# record input map conflicts to resolve them on final step generation
+self._conflicts = []
 if rqlhelper is not None: # else test
 self._insert_identity_variable = rqlhelper._annotator.rewrite_shared_optional
+if server.DEBUG:
+print 'sourcesterms:'
+for source, terms in self.sourcesterms.items():
+print source, terms
 def copy_solutions(self, solindices):
 return [self._solutions[solidx].copy() for solidx in solindices]
 @property
 @cached
 def part_sources(self):
-if self._sourcesvars:
+if self._sourcesterms:
-return tuple(sorted(self._sourcesvars))
+return tuple(sorted(self._sourcesterms))
-return (self._session.repo.system_source,)
+return (self._repo.system_source,)
 @property
 @cached
 def _sys_source_set(self):
-return frozenset((self._session.repo.system_source, solindex)
+return frozenset((self._repo.system_source, solindex)
 for solindex in self._solindices)
 @cached
 def _norel_support_set(self, relation):
 """return a set of (source, solindex) where source doesn't support the
 relation
 """
-return frozenset((source, solidx) for source in self._session.repo.sources
+return frozenset((source, solidx) for source in self._repo.sources
 for solidx in self._solindices
-if not ((source.support_relation(relation.r_type) and
+if not ((source.support_relation(relation.r_type))
-not self.crossed_relation(source, relation))
 or relation.r_type in source.dont_cross_relations))
+def _compute_sourcesterms(self):
-def _compute_sourcesvars(self):
+"""compute for each term (variable, rewritten constant, relation) and
-"""compute for each variable/solution in the rqlst which sources support
+for each solution in the rqlst which sources support them
-them
 """
-repo = self._session.repo
+repo = self._repo
-eschema = repo.schema.eschema
+eschema = self._schema.eschema
-sourcesvars = self._sourcesvars
+sourcesterms = self._sourcesterms
 # find for each source which variable/solution are supported
 for varname, varobj in self.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)"
 if varobj.stinfo['uidrels']:
 vrels = varobj.stinfo['relations'] - varobj.stinfo['uidrels']
 for rel in varobj.stinfo['uidrels']:
-if  rel.neged(strict=True) or rel.operator() != '=':
+if rel.neged(strict=True) or rel.operator() != '=':
 continue
 for const in rel.children[1].get_nodes(Constant):
 eid = const.eval(self.plan.args)
 source = self._session.source_from_eid(eid)
 if vrels and not any(source.support_relation(r.r_type)
 for r in vrels):
-self._set_source_for_var(repo.system_source, varobj)
+self._set_source_for_term(repo.system_source, varobj)
 else:
-self._set_source_for_var(source, varobj)
+self._set_source_for_term(source, varobj)
 continue
 rels = varobj.stinfo['relations']
 if not rels and not varobj.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)
+self._set_source_for_term(repo.system_source, varobj)
 continue
 for i, sol in enumerate(self._solutions):
 vartype = sol[varname]
 # skip final variable
 if eschema(vartype).is_final():
 # * the variable isn't invariant
 # * at least one supported relation specified
 if not varobj._q_invariant or \
 any(imap(source.support_relation,
 (r.r_type for r in rels if r.r_type != 'eid'))):
-sourcesvars.setdefault(source, {}).setdefault(varobj, set()).add(i)
+sourcesterms.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
 # query
 if not varobj._q_invariant and any(ifilterfalse(
 source.support_relation, (r.r_type for r in rels))):
 self.needsplit = True
+# add source for rewritten constants to sourcesterms
-def _handle_cross_relation(self, rel, relsources, vsources):
+for vconsts in self.rqlst.stinfo['rewritten'].itervalues():
-crossvars = None
+const = vconsts[0]
-for source in relsources:
+source = self._session.source_from_eid(const.eval(self.plan.args))
-if rel.r_type in source.cross_relations:
+if source is self._repo.system_source:
-crossvars = set(x.variable for x in rel.get_nodes(VariableRef))
+for const in vconsts:
-crossvars.update(frozenset(x for x in rel.get_nodes(Constant)))
+self._set_source_for_term(source, const)
-assert len(crossvars) == 2
+elif source in self._sourcesterms:
-ssource = self._session.repo.system_source
+source_scopes = frozenset(t.scope for t in self._sourcesterms[source])
-needsplit = True
+for const in vconsts:
-flag = 0
+if const.scope in source_scopes:
-for v in crossvars:
+self._set_source_for_term(source, const)
-if isinstance(v, Constant):
+# add source for relations
-allsols = set(self._solindices)
+rschema = self._schema.rschema
-try:
+termssources = {}
-self._sourcesvars[ssource][v] = allsols
-except KeyError:
-self._sourcesvars[ssource] = {v: allsols}
-if len(vsources[v]) == 1:
-if iter(vsources[v]).next()[0].uri == 'system':
-flag = 1
-for ov in crossvars:
-if ov is not v and (isinstance(ov, Constant) or ov._q_invariant):
-ssset = frozenset((ssource,))
-self._remove_sources(ov, vsources[ov] - ssset)
-else:
-for ov in crossvars:
-if ov is not v and (isinstance(ov, Constant) or ov._q_invariant):
-needsplit = False
-break
-else:
-continue
-if not rel.neged(strict=True):
-break
-else:
-self._crossrelations.setdefault(source, {})[rel] = crossvars
-if not flag:
-self._sourcesvars.setdefault(source, {})[rel] = set(self._solindices)
-self._sourcesvars.setdefault(ssource, {})[rel] = set(self._solindices)
-if needsplit:
-self.needsplit = True
-return crossvars is None
-def _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.repo
-rschema = repo.schema.rschema
-vsources = {}
 for rel in self.rqlst.iget_nodes(Relation):
 # process non final relations only
 # note: don't try to get schema for 'is' relation (not available
 # during bootstrap)
 if not rel.is_types_restriction() and not rschema(rel.r_type).is_final():
 # attribute
 #
 # XXX code below don't deal if some source allow relation
 #     crossing but not another one
 relsources = repo.rel_type_sources(rel.r_type)
-crossvars = None
 if len(relsources) < 2:
 # filter out sources being there because they have this
 # relation in their dont_cross_relations attribute
 relsources = [source for source in relsources
 if source.support_relation(rel.r_type)]
 if relsources:
 # this means the relation is using a variable inlined as
 # a constant and another unsupported variable, in which
-# case we put the relation in sourcesvars
+# case we put the relation in sourcesterms
-self._sourcesvars.setdefault(relsources[0], {})[rel] = set(self._solindices)
+self._sourcesterms.setdefault(relsources[0], {})[rel] = set(self._solindices)
 continue
 lhs, rhs = rel.get_variable_parts()
 lhsv, rhsv = getattr(lhs, 'variable', lhs), getattr(rhs, 'variable', rhs)
-# update dictionnary of sources supporting lhs and rhs vars
+# update dictionary of sources supporting lhs and rhs vars
-if not lhsv in vsources:
+if not lhsv in termssources:
-vsources[lhsv] = self._term_sources(lhs)
+termssources[lhsv] = self._term_sources(lhs)
-if not rhsv in vsources:
+if not rhsv in termssources:
-vsources[rhsv] = self._term_sources(rhs)
+termssources[rhsv] = self._term_sources(rhs)
-if self._handle_cross_relation(rel, relsources, vsources):
+self._handle_cross_relation(rel, relsources, termssources)
-self._linkedvars.setdefault(lhsv, set()).add((rhsv, rel))
+self._linkedterms.setdefault(lhsv, set()).add((rhsv, rel))
-self._linkedvars.setdefault(rhsv, set()).add((lhsv, rel))
+self._linkedterms.setdefault(rhsv, set()).add((lhsv, rel))
-else:
+return termssources
-self._crosslinkedvars.setdefault(lhsv, set()).add((rhsv, rel))
-self._crosslinkedvars.setdefault(rhsv, set()).add((lhsv, rel))
+def _handle_cross_relation(self, rel, relsources, termssources):
-for term in self._linkedvars:
+cross_rel = False
-self._remove_sources_until_stable(term, vsources)
+for source in relsources:
-if len(self._sourcesvars) > 1 and hasattr(self.plan.rqlst, 'main_relations'):
+if rel.r_type in source.cross_relations:
+ssource = self._repo.system_source
+crossvars = set(x.variable for x in rel.get_nodes(VariableRef))
+for const in rel.get_nodes(Constant):
+if source.uri != 'system' and not const in self._sourcesterms.get(source, ()):
+continue
+crossvars.add(const)
+# XXX this is counter intuitive, though this is currently a
+# trick to add const to system source terms so we get a
+# chance that solutions will compare to equals when
+# computing need split
+allsols = set(self._solindices)
+try:
+self._sourcesterms[ssource][const] = allsols
+except KeyError:
+self._sourcesterms[ssource] = {const: allsols}
+self._crossrelations.setdefault(source, {})[rel] = crossvars
+if len(crossvars) < 2:
+# this means there is a constant in the relation which is
+# not supported by the source, so we can stop here
+continue
+self._sourcesterms.setdefault(ssource, {})[rel] = set(self._solindices)
+cross_rel = True
+needsplit = True
+flag = False
+for term in crossvars:
+if len(termssources[term]) == 1:
+if iter(termssources[term]).next()[0].uri == 'system':
+flag = True
+for ov in crossvars:
+if ov is not term and (isinstance(ov, Constant) or ov._q_invariant):
+ssset = frozenset((ssource,))
+self._remove_sources(ov, termssources[ov] - ssset)
+else:
+for ov in crossvars:
+if ov is not term and (isinstance(ov, Constant) or ov._q_invariant):
+needsplit = False
+break
+else:
+continue
+if not flag:
+self._sourcesterms.setdefault(source, {})[rel] = set(self._solindices)
+if needsplit:
+self.needsplit = True
+return cross_rel
+def _remove_invalid_sources(self, termssources):
+"""removes invalid sources from `sourcesterms` member according to
+traversed relations and their properties (which sources support them,
+can they cross sources, etc...)
+"""
+for term in self._linkedterms:
+self._remove_sources_until_stable(term, termssources)
+if len(self._sourcesterms) > 1 and hasattr(self.plan.rqlst, 'main_relations'):
 # the querier doesn't annotate write queries, need to do it here
 self.plan.annotate_rqlst()
 # insert/update/delete queries, we may get extra information from
 # the main relation (eg relations to the left of the WHERE
 if self.plan.rqlst.TYPE == 'insert':
 inserted = dict((vref.variable, etype)
 for etype, vref in self.plan.rqlst.main_variables)
 else:
 inserted = {}
+repo = self._repo
+rschema = self._schema.rschema
 for rel in self.plan.rqlst.main_relations:
 if not rschema(rel.r_type).is_final():
 # nothing to do if relation is not supported by multiple sources
 if len(repo.rel_type_sources(rel.r_type)) < 2:
 continue
 lhs, rhs = rel.get_variable_parts()
 try:
-lhsv = self._extern_term(lhs, vsources, inserted)
+lhsv = self._extern_term(lhs, termssources, inserted)
-rhsv = self._extern_term(rhs, vsources, inserted)
+rhsv = self._extern_term(rhs, termssources, inserted)
 except KeyError, ex:
 continue
-norelsup = self._norel_support_set(rel)
+self._remove_term_sources(lhsv, rel, rhsv, termssources)
-self._remove_var_sources(lhsv, norelsup, rhsv, vsources)
+self._remove_term_sources(rhsv, rel, lhsv, termssources)
-self._remove_var_sources(rhsv, norelsup, lhsv, vsources)
-# cleanup linked var
+def _extern_term(self, term, termssources, inserted):
-for var, linkedrelsinfo in self._linkedvars.iteritems():
-self._linkedvars[var] = frozenset(x[0] for x in linkedrelsinfo)
-# if there are other sources than the system source, consider simplified
-# variables'source
-if self._sourcesvars and self._sourcesvars.keys() != [self._session.repo.system_source]:
-# add source for rewritten constants to sourcesvars
-for vconsts in self.rqlst.stinfo['rewritten'].itervalues():
-const = vconsts[0]
-eid = const.eval(self.plan.args)
-source = self._session.source_from_eid(eid)
-if source is self._session.repo.system_source:
-for const in vconsts:
-self._set_source_for_var(source, const)
-elif source in self._sourcesvars:
-source_scopes = frozenset(v.scope for v in self._sourcesvars[source])
-for const in vconsts:
-if const.scope in source_scopes:
-self._set_source_for_var(source, const)
-def _extern_term(self, term, vsources, inserted):
 var = term.variable
 if var.stinfo['constnode']:
 termv = var.stinfo['constnode']
-vsources[termv] = self._term_sources(termv)
+termssources[termv] = self._term_sources(termv)
 elif var in inserted:
 termv = var
-source = self._session.repo.locate_etype_source(inserted[var])
+source = self._repo.locate_etype_source(inserted[var])
-vsources[termv] = set((source, solindex) for solindex in self._solindices)
+termssources[termv] = set((source, solindex)
+for solindex in self._solindices)
 else:
 termv = self.rqlst.defined_vars[var.name]
-if not termv in vsources:
+if not termv in termssources:
-vsources[termv] = self._term_sources(termv)
+termssources[termv] = self._term_sources(termv)
 return termv
-def _remove_sources_until_stable(self, var, vsources):
+def _remove_sources_until_stable(self, term, termssources):
-sourcesvars = self._sourcesvars
+sourcesterms = self._sourcesterms
-for ovar, rel in self._linkedvars.get(var, ()):
+for oterm, rel in self._linkedterms.get(term, ()):
-if not var.scope is ovar.scope and rel.scope.neged(strict=True):
+if not term.scope is oterm.scope and rel.scope.neged(strict=True):
 # can't get information from relation inside a NOT exists
-# where variables don't belong to the same scope
+# where terms don't belong to the same scope
 continue
 need_ancestor_scope = False
-if not (var.scope is rel.scope and ovar.scope is rel.scope):
+if not (term.scope is rel.scope and oterm.scope is rel.scope):
 if rel.ored():
 continue
 if rel.ored(traverse_scope=True):
 # if relation has some OR as parent, constraints should only
 # propagate from parent scope to child scope, nothing else
 need_ancestor_scope = True
-relsources = self._session.repo.rel_type_sources(rel.r_type)
+relsources = self._repo.rel_type_sources(rel.r_type)
 if rel.neged(strict=True) and (
 len(relsources) < 2
-or not isinstance(ovar, Variable)
+or not isinstance(oterm, Variable)
-or ovar.valuable_references() != 1
+or oterm.valuable_references() != 1
-or any(sourcesvars[source][var] != sourcesvars[source][ovar]
+or any(sourcesterms[source][term] != sourcesterms[source][oterm]
 for source in relsources
-if var in sourcesvars.get(source, ())
+if term in sourcesterms.get(source, ())
-and ovar in sourcesvars.get(source, ()))):
+and oterm in sourcesterms.get(source, ()))):
-# neged relation doesn't allow to infer variable sources unless we're
+# neged relation doesn't allow to infer term sources unless
-# on a multisource relation for a variable only used by this relation
+# we're on a multisource relation for a term only used by this
-# (eg "Any X WHERE NOT X multisource_rel Y" and over is Y), iif
+# relation (eg "Any X WHERE NOT X multisource_rel Y" and over is
+# Y)
 continue
-norelsup = self._norel_support_set(rel)
+# compute invalid sources for terms and remove them
-# compute invalid sources for variables and remove them
+if not need_ancestor_scope or is_ancestor(term.scope, oterm.scope):
-if not need_ancestor_scope or is_ancestor(var.scope, ovar.scope):
+self._remove_term_sources(term, rel, oterm, termssources)
-self._remove_var_sources(var, norelsup, ovar, vsources)
+if not need_ancestor_scope or is_ancestor(oterm.scope, term.scope):
-if not need_ancestor_scope or is_ancestor(ovar.scope, var.scope):
+self._remove_term_sources(oterm, rel, term, termssources)
-self._remove_var_sources(ovar, norelsup, var, vsources)
+def _remove_term_sources(self, term, rel, oterm, termssources):
-def _remove_var_sources(self, var, norelsup, ovar, vsources):
+"""remove invalid sources for term according to oterm's sources and the
-"""remove invalid sources for var according to ovar's sources and the
+relation between those two terms.
-relation between those two variables.
 """
-varsources = vsources[var]
+norelsup = self._norel_support_set(rel)
-invalid_sources = varsources - (vsources[ovar] | norelsup)
+termsources = termssources[term]
+invalid_sources = termsources - (termssources[oterm] | norelsup)
+if invalid_sources and self._repo.can_cross_relation(rel.r_type):
+invalid_sources -= self._sys_source_set
+if invalid_sources and isinstance(term, Variable) and self._need_ext_source_access(term, rel):
+# if the term is a not invariant variable, we should filter out
+# source where the relation is a cross relation from invalid
+# sources
+invalid_sources = frozenset([(s, solidx) for s, solidx in invalid_sources
+if not (s in self._crossrelations and
+rel in self._crossrelations[s])])
 if invalid_sources:
-self._remove_sources(var, invalid_sources)
+self._remove_sources(term, invalid_sources)
-varsources -= invalid_sources
+termsources -= invalid_sources
-self._remove_sources_until_stable(var, vsources)
+self._remove_sources_until_stable(term, termssources)
 def _compute_needsplit(self):
-"""tell according to sourcesvars if the rqlst has to be splitted for
+"""tell according to sourcesterms 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
 * 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 2
-if len(self._sourcesvars) < 2:
+if len(self._sourcesterms) < 2:
 self.needsplit = False
 elif not self.needsplit:
-if not allequals(self._sourcesvars.itervalues()):
+if not allequals(self._sourcesterms.itervalues()):
 self.needsplit = True
 else:
-sample = self._sourcesvars.itervalues().next()
+sample = self._sourcesterms.itervalues().next()
-if len(sample) > 1 and any(v for v in sample
+if len(sample) > 1:
-if not v in self._linkedvars
+for term in sample:
-and not v in self._crosslinkedvars):
+# need split if unlinked variable
-self.needsplit = True
+if isinstance(term, Variable) and not term in self._linkedterms:
+self.needsplit = True
-def _set_source_for_var(self, source, var):
+break
-self._sourcesvars.setdefault(source, {})[var] = set(self._solindices)
+else:
+# need split if there are some cross relation on non
+# invariant variable or if the variable is used in
+# multi-sources relation
+if self._crossrelations:
+for reldict in self._crossrelations.itervalues():
+for rel, terms in reldict.iteritems():
+for term in terms:
+if isinstance(term, Variable) and self._need_ext_source_access(term, rel):
+self.needsplit = True
+return
+@cached
+def _need_ext_source_access(self, var, rel):
+if not var._q_invariant:
+return True
+if  any(r for x, r in self._linkedterms[var]
+if not r is rel and self._repo.is_multi_sources_relation(r.r_type)):
+return True
+return False
+def _set_source_for_term(self, source, term):
+self._sourcesterms.setdefault(source, {})[term] = set(self._solindices)
 def _term_sources(self, term):
 """returns possible sources for terms `term`"""
 if isinstance(term, Constant):
 source = self._session.source_from_eid(term.eval(self.plan.args))
 return set((source, solindex) for solindex in self._solindices)
 else:
 var = getattr(term, 'variable', term)
-sources = [source for source, varobjs in self.sourcesvars.iteritems()
+sources = [source for source, varobjs in self.sourcesterms.iteritems()
 if var in varobjs]
 return set((source, solindex) for source in sources
-for solindex in self.sourcesvars[source][var])
+for solindex in self.sourcesterms[source][var])
-def _remove_sources(self, var, sources):
+def _remove_sources(self, term, sources):
-"""removes invalid sources (`sources`) from `sourcesvars`
+"""removes invalid sources (`sources`) from `sourcesterms`
 :param sources: the list of sources to remove
-:param var: the analyzed variable
+:param term: the analyzed term
 """
-sourcesvars = self._sourcesvars
+sourcesterms = self._sourcesterms
 for source, solindex in sources:
 try:
-sourcesvars[source][var].remove(solindex)
+sourcesterms[source][term].remove(solindex)
 except KeyError:
 return # may occur with subquery column alias
-if not sourcesvars[source][var]:
+if not sourcesterms[source][term]:
-del sourcesvars[source][var]
+del sourcesterms[source][term]
-if not sourcesvars[source]:
+if not sourcesterms[source]:
-del sourcesvars[source]
+del sourcesterms[source]
 def crossed_relation(self, source, relation):
 return relation in self._crossrelations.get(source, ())
 def part_steps(self):
 each step. This is necessary to know if an aggregate step will be
 necessary or not.
 """
 steps = []
 select = self.rqlst
-rschema = self.plan.schema.rschema
+rschema = self._schema.rschema
 for source in self.part_sources:
-sourcevars = self._sourcesvars[source]
+sourceterms = self._sourcesterms[source]
-while sourcevars:
+while sourceterms:
-# take a variable randomly, and all variables supporting the
+# take a term randomly, and all terms supporting the
 # same solutions
-var, solindices = self._choose_var(sourcevars)
+term, solindices = self._choose_term(sourceterms)
 if source.uri == 'system':
 # ensure all variables are available for the latest step
 # (missing one will be available from temporary tables
 # of previous steps)
 scope = select
-variables = scope.defined_vars.values() + scope.aliases.values()
+terms = scope.defined_vars.values() + scope.aliases.values()
-sourcevars.clear()
+sourceterms.clear()
+sources = [source]
 else:
-scope = var.scope
+scope = term.scope
-variables = self._expand_vars(var, source, sourcevars, scope, solindices)
+# find which sources support the same term and solutions
-if not sourcevars:
+sources = self._expand_sources(source, term, solindices)
-del self._sourcesvars[source]
+# no try to get as much terms as possible
-# find which sources support the same variables/solutions
+terms = self._expand_terms(term, sources, sourceterms,
-sources = self._expand_sources(source, variables, solindices)
+scope, solindices)
+if len(terms) == 1 and isinstance(terms[0], Constant):
+# we can't generate anything interesting with a single
+# constant term (will generate an empty "Any" query),
+# go to the next iteration directly!
+continue
+if not sourceterms:
+del self._sourcesterms[source]
 # suppose this is a final step until the contrary is proven
 final = scope is select
-# set of variables which should be additionaly selected when
+# set of terms which should be additionaly selected when
 # possible
 needsel = set()
 # add attribute variables and mark variables which should be
 # additionaly selected when possible
 for var in select.defined_vars.itervalues():
-if not var in variables:
+if not var in terms:
 stinfo = var.stinfo
 for ovar, rtype in stinfo['attrvars']:
-if ovar in variables:
+if ovar in terms:
 needsel.add(var.name)
-variables.append(var)
+terms.append(var)
 break
 else:
 needsel.add(var.name)
 final = False
 if final and source.uri != 'system':
 for vconsts in select.stinfo['rewritten'].itervalues():
 const = vconsts[0]
 eid = const.eval(self.plan.args)
 _source = self._session.source_from_eid(eid)
 if len(sources) > 1 or not _source in sources:
-# if there is some rewriten constant used by a
+# if there is some rewriten constant used by a not
-# not neged relation while there are some source
+# neged relation while there are some source not
-# not supporting the associated entity, this step
+# supporting the associated entity, this step can't
-# can't be final (unless the relation is explicitly
+# be final (unless the relation is explicitly in
-# in `variables`, eg cross relations)
+# `terms`, eg cross relations)
 for c in vconsts:
 rel = c.relation()
-if rel is None or not (rel in variables or rel.neged(strict=True)):
+if rel is None or not (rel in terms or rel.neged(strict=True)):
-#if rel is not None and rel.r_type == 'identity' and not rel.neged(strict=True):
 final = False
 break
 break
 # check where all relations are supported by the sources
 for rel in scope.iget_nodes(Relation):
 if not _source.support_relation(rel.r_type):
 for vref in rel.iget_nodes(VariableRef):
 needsel.add(vref.name)
 final = False
 break
-elif self.crossed_relation(_source, rel) and not rel in variables:
+elif self.crossed_relation(_source, rel) and not rel in terms:
 final = False
 break
 else:
 if not scope is select:
-self._exists_relation(rel, variables, needsel)
+self._exists_relation(rel, terms, needsel)
 # if relation is supported by all sources and some of
-# its lhs/rhs variable isn't in "variables", and the
+# its lhs/rhs variable isn't in "terms", and the
-# other end *is* in "variables", mark it have to be
+# other end *is* in "terms", mark it have to be
 # selected
 if source.uri != 'system' and not rschema(rel.r_type).is_final():
 lhs, rhs = rel.get_variable_parts()
 try:
 lhsvar = lhs.variable
 lhsvar = lhs
 try:
 rhsvar = rhs.variable
 except AttributeError:
 rhsvar = rhs
-if lhsvar in variables and not rhsvar in variables:
+if lhsvar in terms and not rhsvar in terms:
 needsel.add(lhsvar.name)
-elif rhsvar in variables and not lhsvar in variables:
+elif rhsvar in terms and not lhsvar in terms:
 needsel.add(rhsvar.name)
 if final:
-self._cleanup_sourcesvars(sources, solindices)
+self._cleanup_sourcesterms(sources, solindices)
-# XXX rename: variables may contain Relation and Constant nodes...
+steps.append((sources, terms, solindices, scope, needsel, final)
-steps.append( (sources, variables, solindices, scope, needsel,
+)
-final) )
 return steps
-def _exists_relation(self, rel, variables, needsel):
+def _exists_relation(self, rel, terms, needsel):
-rschema = self.plan.schema.rschema(rel.r_type)
+rschema = self._schema.rschema(rel.r_type)
 lhs, rhs = rel.get_variable_parts()
 try:
 lhsvar, rhsvar = lhs.variable, rhs.variable
 except AttributeError:
 pass
 # 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)
-if rhsvar in variables and not lhsvar in variables:
+if rhsvar in terms and not lhsvar in terms:
-self._identity_substitute(rel, lhsvar, variables, needsel)
+self._identity_substitute(rel, lhsvar, terms, needsel)
-elif lhsvar in variables and not rhsvar in variables:
+elif lhsvar in terms and not rhsvar in terms:
-self._identity_substitute(rel, rhsvar, variables, needsel)
+self._identity_substitute(rel, rhsvar, terms, needsel)
-def _identity_substitute(self, relation, var, variables, needsel):
+def _identity_substitute(self, relation, var, terms, needsel):
 newvar = self._insert_identity_variable(relation.scope, var)
 if newvar is not None:
 # 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)
+terms.append(newvar)
 needsel.add(newvar.name)
-def _choose_var(self, sourcevars):
+def _choose_term(self, sourceterms):
+"""pick one term among terms supported by a source, which will be used
+as a base to generate an execution step
+"""
 secondchoice = None
-if len(self._sourcesvars) > 1:
+if len(self._sourcesterms) > 1:
 # priority to variable from subscopes
-for var in sourcevars:
+for var in sourceterms:
 if not var.scope is self.rqlst:
 if isinstance(var, Variable):
-return var, sourcevars.pop(var)
+return var, sourceterms.pop(var)
 secondchoice = var
 else:
 # priority to variable outer scope
-for var in sourcevars:
+for var in sourceterms:
 if var.scope is self.rqlst:
 if isinstance(var, Variable):
-return var, sourcevars.pop(var)
+return var, sourceterms.pop(var)
 secondchoice = var
 if secondchoice is not None:
-return secondchoice, sourcevars.pop(secondchoice)
+return secondchoice, sourceterms.pop(secondchoice)
 # priority to variable
-for var in sourcevars:
+for var in sourceterms:
 if isinstance(var, Variable):
-return var, sourcevars.pop(var)
+return var, sourceterms.pop(var)
 # whatever
-var = iter(sourcevars).next()
+var = iter(sourceterms).next()
-return var, sourcevars.pop(var)
+return var, sourceterms.pop(var)
+def _expand_sources(self, selected_source, term, solindices):
+"""return all sources supporting given term / solindices"""
+sources = [selected_source]
+sourcesterms = self._sourcesterms
+for source in sourcesterms:
+if source is selected_source:
+continue
+if not (term in sourcesterms[source] and
+solindices.issubset(sourcesterms[source][term])):
+continue
+sources.append(source)
+if source.uri != 'system':
+termsolindices = sourcesterms[source][term]
+termsolindices -= solindices
+if not termsolindices:
+del sourcesterms[source][term]
+return sources
-def _expand_vars(self, var, source, sourcevars, scope, solindices):
+def _expand_terms(self, term, sources, sourceterms, scope, solindices):
-variables = [var]
+terms = [term]
+sources = sorted(sources)
 nbunlinked = 1
-linkedvars = self._linkedvars
+linkedterms = self._linkedterms
-# variable has to belong to the same scope if there is more
+# term has to belong to the same scope if there is more
 # than the system source remaining
-if len(self._sourcesvars) > 1 and not scope is self.rqlst:
+if len(self._sourcesterms) > 1 and not scope is self.rqlst:
-candidates = (v for v in sourcevars.keys() if scope is v.scope)
+candidates = (t for t in sourceterms.keys() if scope is t.scope)
 else:
-candidates = sourcevars #.iterkeys()
+candidates = sourceterms #.iterkeys()
-# we only want one unlinked variable in each generated query
+# we only want one unlinked term in each generated query
-candidates = [v for v in candidates
+candidates = [t for t in candidates
-if isinstance(v, Constant) or
+if isinstance(t, Constant) or
-(solindices.issubset(sourcevars[v]) and v in linkedvars)]
+(solindices.issubset(sourceterms[t]) and t in linkedterms)]
-accept_var = lambda x: (isinstance(x, Constant) or any(v for v in variables if v in linkedvars.get(x, ())))
+accept_term = lambda x: (not any(s for s in sources if not x in self._sourcesterms[s])
-source_cross_rels = self._crossrelations.get(source, ())
+and any(t for t in terms if t in linkedterms.get(x, ())))
-if isinstance(var, Relation) and var in source_cross_rels:
+source_cross_rels = {}
-cross_vars = source_cross_rels.pop(var)
+for source in sources:
-base_accept_var = accept_var
+source_cross_rels.update(self._crossrelations.get(source, {}))
-accept_var = lambda x: (base_accept_var(x) or x in cross_vars)
+if isinstance(term, Relation) and term in source_cross_rels:
-for refed in cross_vars:
+cross_terms = source_cross_rels.pop(term)
+base_accept_term = accept_term
+accept_term = lambda x: (base_accept_term(x) or x in cross_terms)
+for refed in cross_terms:
 if not refed in candidates:
 candidates.append(refed)
 else:
-cross_vars = ()
+cross_terms = ()
-# repeat until no variable can't be added, since addition of a new
+# repeat until no term can't be added, since addition of a new
-# variable may permit to another one to be added
+# term may permit to another one to be added
 modified = True
 while modified and candidates:
 modified = False
-for var in candidates[:]:
+for term in candidates[:]:
-if accept_var(var):
+if isinstance(term, Constant):
-variables.append(var)
+relation = term.relation()
-try:
+if sorted(set(x[0] for x in self._term_sources(term))) != sources:
-# constant nodes should be systematically deleted
+continue
-if isinstance(var, Constant):
+terms.append(term)
-del sourcevars[var]
+candidates.remove(term)
-else:
-# variable nodes should be deleted once all possible
-# solutions indices have been consumed
-sourcevars[var] -= solindices
-if not sourcevars[var]:
-del sourcevars[var]
-except KeyError:
-assert var in cross_vars
-candidates.remove(var)
 modified = True
-return variables
+del sourceterms[term]
+elif accept_term(term):
-def _expand_sources(self, selected_source, vars, solindices):
+terms.append(term)
-sources = [selected_source]
+candidates.remove(term)
-sourcesvars = self._sourcesvars
+modified = True
-for source in sourcesvars:
+for source in sources:
-if source is selected_source:
+sourceterms = self._sourcesterms[source]
-continue
+# terms should be deleted once all possible solutions
-for var in vars:
+# indices have been consumed
-if not (var in sourcesvars[source] and
+try:
-solindices.issubset(sourcesvars[source][var])):
+sourceterms[term] -= solindices
-break
+if not sourceterms[term]:
-else:
+del sourceterms[term]
-sources.append(source)
+except KeyError:
-if source.uri != 'system':
+assert term in cross_terms
-for var in vars:
+return terms
-varsolindices = sourcesvars[source][var]
-varsolindices -= solindices
+def _cleanup_sourcesterms(self, sources, solindices):
-if not varsolindices:
-del sourcesvars[source][var]
-return sources
-def _cleanup_sourcesvars(self, sources, solindices):
 """on final parts, remove solutions so we know they are already processed"""
 for source in sources:
 try:
-sourcevars = self._sourcesvars[source]
+sourceterms = self._sourcesterms[source]
 except KeyError:
 continue
-for var, varsolindices in sourcevars.items():
+for term, termsolindices in sourceterms.items():
-if isinstance(var, Relation) and self.crossed_relation(source, var):
+if isinstance(term, Relation) and self.crossed_relation(source, term):
 continue
-varsolindices -= solindices
+termsolindices -= solindices
-if not varsolindices:
+if not termsolindices:
-del sourcevars[var]
+del sourceterms[term]
 def merge_input_maps(self, allsolindices):
-"""inputmaps is a dictionary with tuple of solution indices as key with an
+"""inputmaps is a dictionary with tuple of solution indices as key with
-associateed input map as value. This function compute for each solution
+an associated input map as value. This function compute for each
-its necessary input map and return them grouped
+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',
 result.append( (list(allsolindices), None) )
 return result
 def build_final_part(self, select, solindices, inputmap,  sources,
 insertedvars):
-plan = self.plan
+solutions = [self._solutions[i] for i in solindices]
-rqlst = plan.finalize(select, [self._solutions[i] for i in solindices],
+if self._conflicts:
-insertedvars)
+for varname, mappedto in self._conflicts:
+var = select.defined_vars[varname]
+newvar = select.make_variable()
+# XXX should use var.scope but scope hasn't been computed yet
+select.add_relation(var, 'identity', newvar)
+for sol in solutions:
+sol[newvar.name] = sol[varname]
+inputmap[newvar.name] = mappedto
+rqlst = self.plan.finalize(select, solutions, insertedvars)
 if self.temptable is None and self.finaltable is None:
-return OneFetchStep(plan, rqlst, sources, inputmap=inputmap)
+return OneFetchStep(self.plan, rqlst, sources, inputmap=inputmap)
 table = self.temptable or self.finaltable
-return FetchStep(plan, rqlst, sources, table, True, inputmap)
+return FetchStep(self.plan, rqlst, sources, table, True, inputmap)
 def build_non_final_part(self, select, solindices, sources, insertedvars,
 table):
 """non final step, will have to store results in a temporary table"""
-plan = self.plan
+solutions = [self._solutions[i] for i in solindices]
-rqlst = plan.finalize(select, [self._solutions[i] for i in solindices],
+rqlst = self.plan.finalize(select, solutions, insertedvars)
-insertedvars)
+step = FetchStep(self.plan, rqlst, sources, table, False)
-step = FetchStep(plan, rqlst, sources, table, False)
 # update input map for following steps, according to processed solutions
 inputmapkey = tuple(sorted(solindices))
 inputmap = self._inputmaps.setdefault(inputmapkey, {})
+for varname, mapping in step.outputmap.iteritems():
+if varname in inputmap and \
+not (mapping == inputmap[varname] or
+self._schema.eschema(solutions[0][varname]).is_final()):
+self._conflicts.append((varname, inputmap[varname]))
 inputmap.update(step.outputmap)
-plan.add_step(step)
+self.plan.add_step(step)
 class MSPlanner(SSPlanner):
 """MultiSourcesPlanner: build execution plan for rql queries
 select_group_sort(select)
 else:
 atemptable = None
 selection = select.selection
 ppi.temptable = atemptable
-vfilter = VariablesFiltererVisitor(self.schema, ppi)
+vfilter = TermsFiltererVisitor(self.schema, ppi)
 steps = []
-for sources, variables, solindices, scope, needsel, final in stepdefs:
+for sources, terms, solindices, scope, needsel, final in stepdefs:
-# extract an executable query using only the specified variables
+# extract an executable query using only the specified terms
 if sources[0].uri == 'system':
 # in this case we have to merge input maps before call to
 # filter so already processed restriction are correctly
 # removed
 solsinputmaps = ppi.merge_input_maps(solindices)
 for solindices, inputmap in solsinputmaps:
 minrqlst, insertedvars = vfilter.filter(
-sources, variables, scope, set(solindices), needsel, final)
+sources, terms, scope, set(solindices), needsel, final)
 if inputmap is None:
 inputmap = subinputmap
 else:
 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
+# original query part) if all term / sources have been
 # treated or if this is the last shot for used solutions
 minrqlst, insertedvars = vfilter.filter(
-sources, variables, scope, solindices, needsel, final)
+sources, terms, scope, solindices, needsel, final)
 if final:
 solsinputmaps = ppi.merge_input_maps(solindices)
 for solindices, inputmap in solsinputmaps:
 if inputmap is None:
 inputmap = subinputmap
 else:
 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() for v in variables)),
+table = '_T%s%s' % (''.join(sorted(v._ms_table_key() for v in terms)),
 ''.join(sorted(str(i) for i in solindices)))
 ppi.build_non_final_part(minrqlst, solindices, sources,
 insertedvars, table)
 # finally: join parts, deal with aggregat/group/sorts if necessary
 if atemptable is not None:
 class UnsupportedBranch(Exception):
 pass
-class VariablesFiltererVisitor(object):
+class TermsFiltererVisitor(object):
 def __init__(self, schema, ppi):
 self.schema = schema
 self.ppi = ppi
 self.skip = {}
 self.hasaggrstep = self.ppi.temptable
 self.extneedsel = frozenset(vref.name for sortterm in ppi.rqlst.orderby
 for vref in sortterm.iget_nodes(VariableRef))
-def _rqlst_accept(self, rqlst, node, newroot, variables, setfunc=None):
+def _rqlst_accept(self, rqlst, node, newroot, terms, setfunc=None):
 try:
-newrestr, node_ = node.accept(self, newroot, variables[:])
+newrestr, node_ = node.accept(self, newroot, terms[:])
 except UnsupportedBranch:
 return rqlst
 if setfunc is not None and newrestr is not None:
 setfunc(newrestr)
 if not node_ is node:
 rqlst = node.parent
 return rqlst
-def filter(self, sources, variables, rqlst, solindices, needsel, final):
+def filter(self, sources, terms, rqlst, solindices, needsel, final):
 if server.DEBUG:
-print 'filter', final and 'final' or '', sources, variables, rqlst, solindices, needsel
+print 'filter', final and 'final' or '', sources, terms, rqlst, solindices, needsel
 newroot = Select()
 self.sources = sorted(sources)
-self.variables = variables
+self.terms = terms
 self.solindices = solindices
 self.final = final
-# variables which appear in unsupported branches
+# terms which appear in unsupported branches
 needsel |= self.extneedsel
 self.needsel = needsel
-# variables which appear in supported branches
+# terms which appear in supported branches
 self.mayneedsel = set()
 # new inserted variables
 self.insertedvars = []
 # other structures (XXX document)
 self.mayneedvar, self.hasvar = {}, {}
 self.use_only_defined = False
 self.scopes = {rqlst: newroot}
 if rqlst.where:
-rqlst = self._rqlst_accept(rqlst, rqlst.where, newroot, variables,
+rqlst = self._rqlst_accept(rqlst, rqlst.where, newroot, terms,
 newroot.set_where)
 if isinstance(rqlst, Select):
 self.use_only_defined = True
 if rqlst.groupby:
 groupby = []
 for node in rqlst.groupby:
-rqlst = self._rqlst_accept(rqlst, node, newroot, variables,
+rqlst = self._rqlst_accept(rqlst, node, newroot, terms,
 groupby.append)
 if groupby:
 newroot.set_groupby(groupby)
 if rqlst.having:
 having = []
 for node in rqlst.having:
-rqlst = self._rqlst_accept(rqlst, node, newroot, variables,
+rqlst = self._rqlst_accept(rqlst, node, newroot, terms,
 having.append)
 if having:
 newroot.set_having(having)
 if final and rqlst.orderby and not self.hasaggrstep:
 orderby = []
 for node in rqlst.orderby:
-rqlst = self._rqlst_accept(rqlst, node, newroot, variables,
+rqlst = self._rqlst_accept(rqlst, node, newroot, terms,
 orderby.append)
 if orderby:
 newroot.set_orderby(orderby)
-self.process_selection(newroot, variables, rqlst)
+self.process_selection(newroot, terms, rqlst)
 elif not newroot.where:
-# no restrictions have been copied, just select variables and add
+# no restrictions have been copied, just select terms and add
 # type restriction (done later by add_types_restriction)
-for v in variables:
+for v in terms:
 if not isinstance(v, Variable):
 continue
 newroot.append_selected(VariableRef(newroot.get_variable(v.name)))
 solutions = self.ppi.copy_solutions(solindices)
 cleanup_solutions(newroot, solutions)
 add_types_restriction(self.schema, rqlst, newroot, solutions)
 if server.DEBUG:
 print '--->', newroot
 return newroot, self.insertedvars
-def visit_and(self, node, newroot, variables):
+def visit_and(self, node, newroot, terms):
 subparts = []
 for i in xrange(len(node.children)):
 child = node.children[i]
 try:
-newchild, child_ = child.accept(self, newroot, variables)
+newchild, child_ = child.accept(self, newroot, terms)
 if not child_ is child:
 node = child_.parent
 if newchild is None:
 continue
 subparts.append(newchild)
 visit_or = visit_and
 def _relation_supported(self, relation):
 rtype = relation.r_type
 for source in self.sources:
-if not source.support_relation(rtype) \
+if not source.support_relation(rtype) or (
-or (rtype in source.cross_relations and not relation in self.variables):#self.ppi.crossed_relation(source, relation):
+rtype in source.cross_relations and not relation in self.terms):
 return False
-if not self.final:
+if not self.final and not relation in self.terms:
 rschema = self.schema.rschema(relation.r_type)
 if not rschema.is_final():
 for term in relation.get_nodes((VariableRef, Constant)):
 term = getattr(term, 'variable', term)
 termsources = sorted(set(x[0] for x in self.ppi._term_sources(term)))
 if termsources and termsources != self.sources:
 return False
 return True
-def visit_relation(self, node, newroot, variables):
+def visit_relation(self, node, newroot, terms):
 if not node.is_types_restriction():
 if node in self.skip and self.solindices.issubset(self.skip[node]):
 if not self.schema.rschema(node.r_type).is_final():
 # can't really skip the relation if one variable is selected and only
 # referenced by this relation
 # 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)
 elif not any(self._relation_supported(rel)
 for rel in node.children[0].variable.stinfo['relations']):
-rel, node = self.visit_default(node, newroot, variables)
+rel, node = self.visit_default(node, newroot, terms)
 return rel, node
 else:
 raise UnsupportedBranch()
 rschema = self.schema.rschema(node.r_type)
-res = self.visit_default(node, newroot, variables)[0]
+try:
+res = self.visit_default(node, newroot, terms)[0]
+except Exception, ex:
+raise
 ored = node.ored()
 if rschema.is_final() or rschema.inlined:
 vrefs = node.children[1].get_nodes(VariableRef)
 if not vrefs:
 if not ored:
 else:
 assert len(vrefs) == 1
 vref = vrefs[0]
 # XXX check operator ?
 self.hasvar[(node.children[0].name, rschema)] = vref
-if self._may_skip_attr_rel(rschema, node, vref, ored, variables, res):
+if self._may_skip_attr_rel(rschema, node, vref, ored, terms, res):
 self.skip.setdefault(node, set()).update(self.solindices)
 elif not ored:
 self.skip.setdefault(node, set()).update(self.solindices)
 return res, node
-def _may_skip_attr_rel(self, rschema, rel, vref, ored, variables, res):
+def _may_skip_attr_rel(self, rschema, rel, vref, ored, terms, res):
 var = vref.variable
 if ored:
 return False
 if var.name in self.extneedsel or var.stinfo['selected']:
 return False
 if not same_scope(var):
 return False
-if any(v for v,_ in var.stinfo['attrvars'] if not v.name in variables):
+if any(v for v,_ in var.stinfo['attrvars'] if not v.name in terms):
 return False
 return True
-def visit_exists(self, node, newroot, variables):
+def visit_exists(self, node, newroot, terms):
 newexists = node.__class__()
 self.scopes = {node: newexists}
-subparts, node = self._visit_children(node, newroot, variables)
+subparts, node = self._visit_children(node, newroot, terms)
 if not subparts:
 return None, node
 newexists.set_where(subparts[0])
 return newexists, node
-def visit_not(self, node, newroot, variables):
+def visit_not(self, node, newroot, terms):
-subparts, node = self._visit_children(node, newroot, variables)
+subparts, node = self._visit_children(node, newroot, terms)
 if not subparts:
 return None, node
 return copy_node(newroot, node, subparts), node
-def visit_group(self, node, newroot, variables):
+def visit_group(self, node, newroot, terms):
 if not self.final:
 return None, node
-return self.visit_default(node, newroot, variables)
+return self.visit_default(node, newroot, terms)
-def visit_variableref(self, node, newroot, variables):
+def visit_variableref(self, node, newroot, terms):
 if self.use_only_defined:
 if not node.variable.name in newroot.defined_vars:
 raise UnsupportedBranch(node.name)
-elif not node.variable in variables:
+elif not node.variable in terms:
 raise UnsupportedBranch(node.name)
 self.mayneedsel.add(node.name)
 # set scope so we can insert types restriction properly
 newvar = newroot.get_variable(node.name)
 newvar.stinfo['scope'] = self.scopes.get(node.variable.scope, newroot)
 return VariableRef(newvar), node
-def visit_constant(self, node, newroot, variables):
+def visit_constant(self, node, newroot, terms):
 return copy_node(newroot, node), node
-def visit_default(self, node, newroot, variables):
+def visit_default(self, node, newroot, terms):
-subparts, node = self._visit_children(node, newroot, variables)
+subparts, node = self._visit_children(node, newroot, terms)
 return copy_node(newroot, node, subparts), node
 visit_comparison = visit_mathexpression = visit_constant = visit_function = visit_default
 visit_sort = visit_sortterm = visit_default
-def _visit_children(self, node, newroot, variables):
+def _visit_children(self, node, newroot, terms):
 subparts = []
 for i in xrange(len(node.children)):
 child = node.children[i]
-newchild, child_ = child.accept(self, newroot, variables)
+newchild, child_ = child.accept(self, newroot, terms)
 if not child is child_:
 node = child_.parent
 if newchild is not None:
 subparts.append(newchild)
 return subparts, node
-def process_selection(self, newroot, variables, rqlst):
+def process_selection(self, newroot, terms, rqlst):
 if self.final:
 for term in rqlst.selection:
 newroot.append_selected(term.copy(newroot))
 for vref in term.get_nodes(VariableRef):
 self.needsel.add(vref.name)
 vrefs = term.get_nodes(VariableRef)
 if vrefs:
 supportedvars = []
 for vref in vrefs:
 var = vref.variable
-if var in variables:
+if var in terms:
 supportedvars.append(vref)
 continue
 else:
 self.needsel.add(vref.name)
 break
 supportedvars = []
 for vref in supportedvars:
 if not vref in newroot.get_selected_variables():
 newroot.append_selected(VariableRef(newroot.get_variable(vref.name)))
-def add_necessary_selection(self, newroot, variables):
+def add_necessary_selection(self, newroot, terms):
 selected = tuple(newroot.get_selected_variables())
-for varname in variables:
+for varname in terms:
 var = newroot.defined_vars[varname]
 for vref in var.references():
 rel = vref.relation()
 if rel is None and vref in selected:
 # already selected

changeset 1228	91ae10ffb611
parent 1124	5548b608b7d9
child 1229	dd9bdcfc03b6