1 # copyright 2003-2011 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

     2 # contact http://www.logilab.fr/ -- mailto:contact@logilab.fr

     3 #

     4 # This file is part of CubicWeb.

     5 #

     6 # CubicWeb is free software: you can redistribute it and/or modify it under the

     7 # terms of the GNU Lesser General Public License as published by the Free

     8 # Software Foundation, either version 2.1 of the License, or (at your option)

     9 # any later version.

    10 #

    11 # CubicWeb is distributed in the hope that it will be useful, but WITHOUT

    12 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

    13 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more

    14 # details.

    15 #

    16 # You should have received a copy of the GNU Lesser General Public License along

    17 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.

    18 """plan execution of rql queries on multiple sources

    19 

    20 the best way to understand what are we trying to acheive here is to read the

    21 unit-tests in unittest_msplanner.py

    22 

    23 

    24 What you need to know

    25 ~~~~~~~~~~~~~~~~~~~~~

    26 1. The system source is expected  to support every entity and relation types

    27 

    28 2. Given "X relation Y":

    29 

    30    * if relation, X and Y types are supported by the external source, we suppose

    31      by default that X and Y should both come from the same source as the

    32      relation. You can specify otherwise by adding relation into the

    33      "cross_relations" set in the source's mapping file and it that case, we'll

    34      consider that we can also find in the system source some relation between

    35      X and Y coming from different sources.

    36 

    37    * if "relation" isn't supported by the external source but X or Y

    38      types (or both) are, we suppose by default that can find in the system

    39      source some relation where X and/or Y come from the external source. You

    40      can specify otherwise by adding relation into the "dont_cross_relations"

    41      set in the source's mapping file and it that case, we'll consider that we

    42      can only find in the system source some relation between X and Y coming

    43      the system source.

    44 

    45 

    46 Implementation

    47 ~~~~~~~~~~~~~~

    48 XXX explain algorithm

    49 

    50 

    51 Exemples of multi-sources query execution

    52 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    53 For a system source and a ldap user source (only CWUser and its attributes

    54 is supported, no group or such):

    55 

    56 :CWUser X:

    57 1. fetch CWUser X from both sources and return concatenation of results

    58 

    59 :CWUser X WHERE X in_group G, G name 'users':

    60 * catch 1

    61   1. fetch CWUser X from both sources, store concatenation of results into a

    62      temporary table

    63   2. return the result of TMP X WHERE X in_group G, G name 'users' from the

    64      system source

    65 * catch 2

    66   1. return the result of CWUser X WHERE X in_group G, G name 'users' from system

    67      source, that's enough (optimization of the sql querier will avoid join on

    68      CWUser, so we will directly get local eids)

    69 

    70 :CWUser X,L WHERE X in_group G, X login L, G name 'users':

    71 1. fetch Any X,L WHERE X is CWUser, X login L from both sources, store

    72    concatenation of results into a temporary table

    73 2. return the result of Any X, L WHERE X is TMP, X login LX in_group G,

    74    G name 'users' from the system source

    75 

    76 

    77 :Any X WHERE X owned_by Y:

    78 * catch 1

    79   1. fetch CWUser X from both sources, store concatenation of results into a

    80      temporary table

    81   2. return the result of Any X WHERE X owned_by Y, Y is TMP from the system

    82      source

    83 * catch 2

    84   1. return the result of Any X WHERE X owned_by Y from system source, that's

    85      enough (optimization of the sql querier will avoid join on CWUser, so we

    86      will directly get local eids)

    87 """

    88 

    89 __docformat__ = "restructuredtext en"

    90 

    91 from itertools import imap, ifilterfalse

    92 

    93 from logilab.common.compat import any

    94 from logilab.common.decorators import cached

    95 from logilab.common.deprecation import deprecated

    96 

    97 from rql import BadRQLQuery

    98 from rql.stmts import Union, Select

    99 from rql.nodes import (VariableRef, Comparison, Relation, Constant, Variable,

   100                        Not, Exists, SortTerm, Function)

   101 

   102 from cubicweb import server

   103 from cubicweb.utils import make_uid

   104 from cubicweb.rqlrewrite import add_types_restriction, cleanup_solutions

   105 from cubicweb.server.ssplanner import SSPlanner, OneFetchStep

   106 from cubicweb.server.mssteps import *

   107 

   108 Variable._ms_table_key = lambda x: x.name

   109 Relation._ms_table_key = lambda x: x.r_type

   110 # str() Constant.value to ensure generated table name won't be unicode

   111 Constant._ms_table_key = lambda x: str(x.value)

   112 

   113 Variable._ms_may_be_processed = lambda x, terms, linkedterms: any(

   114     t for t in terms if t in linkedterms.get(x, ()))

   115 Relation._ms_may_be_processed = lambda x, terms, linkedterms: all(

   116     getattr(hs, 'variable', hs) in terms for hs in x.get_variable_parts())

   117 

   118 def ms_scope(term):

   119     rel = None

   120     scope = term.scope

   121     if isinstance(term, Variable) and len(term.stinfo['relations']) == 1:

   122         rel = iter(term.stinfo['relations']).next().relation()

   123     elif isinstance(term, Constant):

   124         rel = term.relation()

   125     elif isinstance(term, Relation):

   126         rel = term

   127     if rel is not None and (

   128         rel.r_type != 'identity' and rel.scope is scope

   129         and isinstance(rel.parent, Exists) and rel.parent.neged(strict=True)):

   130         return scope.parent.scope

   131     return scope

   132 

   133 def need_intersect(select, getrschema):

   134     for rel in select.iget_nodes(Relation):

   135         if isinstance(rel.parent, Exists) and rel.parent.neged(strict=True) and not rel.is_types_restriction():

   136             rschema = getrschema(rel.r_type)

   137             if not rschema.final:

   138                 # if one of the relation's variable is ambiguous but not

   139                 # invariant, an intersection will be necessary

   140                 for vref in rel.get_nodes(VariableRef):

   141                     var = vref.variable

   142                     if (var.valuable_references() == 1

   143                         and len(var.stinfo['possibletypes']) > 1):

   144                         return True

   145     return False

   146 

   147 def neged_relation(rel):

   148     parent = rel.parent

   149     return isinstance(parent, Not) or (isinstance(parent, Exists) and

   150                                        isinstance(parent.parent, Not))

   151 

   152 def need_source_access_relation(vargraph):

   153     if not vargraph:

   154         return False

   155     # check vargraph contains some other relation than the identity relation

   156     # test of key nature since it may be a variable name (don't care about that)

   157     # or a 2-uple (var1, var2) associated to the relation to traverse to go from

   158     # var1 to var2

   159     return any(key for key, val in vargraph.iteritems()

   160                if isinstance(key, tuple) and val != 'identity')

   161 

   162 def need_aggr_step(select, sources, stepdefs=None):

   163     """return True if a temporary table is necessary to store some partial

   164     results to execute the given query

   165     """

   166     if len(sources) == 1:

   167         # can do everything at once with a single source

   168         return False

   169     if select.orderby or select.groupby or select.has_aggregat:

   170         # if more than one source, we need a temp table to deal with sort /

   171         # groups / aggregat if :

   172         # * the rqlst won't be splitted (in the other case the last query

   173         #   using partial temporary table can do sort/groups/aggregat without

   174         #   the need for a later AggrStep)

   175         # * the rqlst is splitted in multiple steps and there are more than one

   176         #   final step

   177         if stepdefs is None:

   178             return True

   179         has_one_final = False

   180         fstepsolindices = set()

   181         for stepdef in stepdefs:

   182             if stepdef[-1]:

   183                 if has_one_final or frozenset(stepdef[2]) != fstepsolindices:

   184                     return True

   185                 has_one_final = True

   186             else:

   187                 fstepsolindices.update(stepdef[2])

   188     return False

   189 

   190 def select_group_sort(select): # XXX something similar done in rql2sql

   191     # add variables used in groups and sort terms to the selection

   192     # if necessary

   193     if select.groupby:

   194         for vref in select.groupby:

   195             if not vref in select.selection:

   196                 select.append_selected(vref.copy(select))

   197     for sortterm in select.orderby:

   198         for vref in sortterm.iget_nodes(VariableRef):

   199             if not vref in select.get_selected_variables():

   200                 # we can't directly insert sortterm.term because it references

   201                 # a variable of the select before the copy.

   202                 # XXX if constant term are used to define sort, their value

   203                 # may necessite a decay

   204                 select.append_selected(vref.copy(select))

   205                 if select.groupby and not vref in select.groupby:

   206                     select.add_group_var(vref.copy(select))

   207 

   208 def allequals(solutions):

   209     """return true if all solutions are identical"""

   210     sol = solutions.next()

   211     noconstsol = None

   212     for sol_ in solutions:

   213         if sol_ != sol:

   214             return False

   215     return True

   216 

   217 # XXX move functions below to rql ##############################################

   218 

   219 def is_ancestor(n1, n2):

   220     """return True if n2 is a parent scope of n1"""

   221     p = n1.parent

   222     while p is not None:

   223         if p is n2:

   224             return True

   225         p = p.parent

   226     return False

   227 

   228 def copy_node(newroot, node, subparts=()):

   229     newnode = node.__class__(*node.initargs(newroot))

   230     for part in subparts:

   231         newnode.append(part)

   232     return newnode

   233 

   234 def used_in_outer_scope(var, scope):

   235     """return true if the variable is used in an outer scope of the given scope

   236     """

   237     for rel in var.stinfo['relations']:

   238         rscope = ms_scope(rel)

   239         if not rscope is scope and is_ancestor(scope, rscope):

   240             return True

   241     return False

   242 

   243 ################################################################################

   244 

   245 class PartPlanInformation(object):

   246     """regroups necessary information to execute some part of a "global" rql

   247     query ("global" means as received by the querier, which may result in

   248     several internal queries, e.g. parts, due to security insertions). Actually

   249     a PPI is created for each subquery and for each query in a union.

   250 

   251     It exposes as well some methods helping in executing this part on a

   252     multi-sources repository, modifying its internal structure during the

   253     process.

   254 

   255     :attr plan:

   256       the execution plan

   257     :attr rqlst:

   258       the original rql syntax tree handled by this part

   259 

   260     :attr needsplit:

   261       bool telling if the query has to be split into multiple steps for

   262       execution or if it can be executed at once

   263 

   264     :attr temptable:

   265       a SQL temporary table name or None, if necessary to handle aggregate /

   266       sorting for this part of the query

   267 

   268     :attr finaltable:

   269       a SQL table name or None, if results for this part of the query should be

   270       written into a temporary table (usually shared by multiple PPI)

   271 

   272     :attr sourcesterms:

   273       a dictionary {source : {term: set([solution index, ])}} telling for each

   274       source which terms are supported for which solutions. A "term" may be

   275       either a rql Variable, Constant or Relation node.

   276     """

   277     def __init__(self, plan, rqlst, rqlhelper=None):

   278         self.plan = plan

   279         self.rqlst = rqlst

   280         self.needsplit = False

   281         self.temptable = None

   282         self.finaltable = None

   283         # shortcuts

   284         self._schema = plan.schema

   285         self._session = plan.session

   286         self._repo = self._session.repo

   287         self._solutions = rqlst.solutions

   288         self._solindices = range(len(self._solutions))

   289         self.system_source = self._repo.system_source

   290         # source : {term: [solution index, ]}

   291         self.sourcesterms = self._sourcesterms = {}

   292         # source : {relation: set(child variable and constant)}

   293         self._crossrelations = {}

   294         # term : set(sources)

   295         self._discarded_sources = {}

   296         # dictionary of variables and constants which are linked to each other

   297         # using a non final relation supported by multiple sources (crossed or

   298         # not).

   299         self._linkedterms = {}

   300         # processing

   301         termssources = self._compute_sourcesterms()

   302         self._remove_invalid_sources(termssources)

   303         self._compute_needsplit()

   304         # after initialisation, .sourcesterms contains the same thing as

   305         # ._sourcesterms though during plan construction, ._sourcesterms will

   306         # be modified while .sourcesterms will be kept unmodified

   307         self.sourcesterms = {}

   308         for k, v in self._sourcesterms.iteritems():

   309             self.sourcesterms[k] = {}

   310             for k2, v2 in v.iteritems():

   311                 self.sourcesterms[k][k2] = v2.copy()

   312         # cleanup linked var

   313         for var, linkedrelsinfo in self._linkedterms.iteritems():

   314             self._linkedterms[var] = frozenset(x[0] for x in linkedrelsinfo)

   315         # map output of a step to input of a following step

   316         self._inputmaps = {}

   317         # record input map conflicts to resolve them on final step generation

   318         self._conflicts = []

   319         if rqlhelper is not None: # else test

   320             self._insert_identity_variable = rqlhelper._annotator.rewrite_shared_optional

   321         if server.DEBUG & server.DBG_MS:

   322             print 'sourcesterms:'

   323             self._debug_sourcesterms()

   324 

   325     def _debug_sourcesterms(self):

   326         for source in self._sourcesterms:

   327             print '-', source

   328             for term, sols in self._sourcesterms[source].items():

   329                 print '  -', term, id(term), ':', sols

   330 

   331     def copy_solutions(self, solindices):

   332         return [self._solutions[solidx].copy() for solidx in solindices]

   333 

   334     @property

   335     @cached

   336     def part_sources(self):

   337         if self._sourcesterms:

   338             return tuple(sorted(self._sourcesterms))

   339         return (self.system_source,)

   340 

   341     @property

   342     @cached

   343     def _sys_source_set(self):

   344         return frozenset((self.system_source, solindex)

   345                          for solindex in self._solindices)

   346 

   347     @cached

   348     def _norel_support_set(self, relation):

   349         """return a set of (source, solindex) where source doesn't support the

   350         relation

   351         """

   352         return frozenset((source, solidx) for source in self._repo.sources

   353                          for solidx in self._solindices

   354                          if not ((source.support_relation(relation.r_type))

   355                                  or relation.r_type in source.dont_cross_relations))

   356 

   357     def _compute_sourcesterms(self):

   358         """compute for each term (variable, rewritten constant, relation) and

   359         for each solution in the rqlst which sources support them

   360         """

   361         repo = self._repo

   362         eschema = self._schema.eschema

   363         sourcesterms = self._sourcesterms

   364         # find for each source which variable/solution are supported

   365         for varname, varobj in self.rqlst.defined_vars.items():

   366             # if variable has an eid specified, we can get its source directly

   367             # NOTE: use uidrel and not constnode to deal with "X eid IN(1,2,3,4)"

   368             if varobj.stinfo['uidrel'] is not None:

   369                 rel = varobj.stinfo['uidrel']

   370                 hasrel = len(varobj.stinfo['relations']) > 1

   371                 for const in rel.children[1].get_nodes(Constant):

   372                     eid = const.eval(self.plan.args)

   373                     source = self._session.source_from_eid(eid)

   374                     if (source is self.system_source

   375                         or (hasrel and varobj._q_invariant and

   376                             not any(source.support_relation(r.r_type)

   377                                     for r in varobj.stinfo['relations']

   378                                     if not r is rel))):

   379                         self._set_source_for_term(self.system_source, varobj)

   380                     else:

   381                         self._set_source_for_term(source, varobj)

   382                 continue

   383             rels = varobj.stinfo['relations']

   384             if not rels and varobj.stinfo['typerel'] is None:

   385                 # (rare) case where the variable has no type specified nor

   386                 # relation accessed ex. "Any MAX(X)"

   387                 self._set_source_for_term(self.system_source, varobj)

   388                 continue

   389             for i, sol in enumerate(self._solutions):

   390                 vartype = sol[varname]

   391                 # skip final variable

   392                 if eschema(vartype).final:

   393                     break

   394                 for source in repo.sources:

   395                     if source.support_entity(vartype):

   396                         # the source support the entity type, though we will

   397                         # actually have to fetch from it only if

   398                         # * the variable isn't invariant

   399                         # * at least one supported relation specified

   400                         if not varobj._q_invariant or \

   401                                any(imap(source.support_relation,

   402                                         (r.r_type for r in rels if r.r_type not in ('identity', 'eid')))):

   403                             sourcesterms.setdefault(source, {}).setdefault(varobj, set()).add(i)

   404                         # if variable is not invariant and is used by a relation

   405                         # not supported by this source, we'll have to split the

   406                         # query

   407                         if not varobj._q_invariant and any(ifilterfalse(

   408                             source.support_relation, (r.r_type for r in rels))):

   409                             self.needsplit = True

   410         # add source for rewritten constants to sourcesterms

   411         self._const_vars = {}

   412         for vconsts in self.rqlst.stinfo['rewritten'].itervalues():

   413             # remember those consts come from the same variable

   414             for const in vconsts:

   415                 self._const_vars[const] = vconsts

   416             source = self._session.source_from_eid(const.eval(self.plan.args))

   417             if source is self.system_source:

   418                 for const in vconsts:

   419                     self._set_source_for_term(source, const)

   420             elif not self._sourcesterms:

   421                 for const in vconsts:

   422                     self._set_source_for_term(source, const)

   423             elif source in self._sourcesterms:

   424                 source_scopes = frozenset(ms_scope(t) for t in self._sourcesterms[source])

   425                 for const in vconsts:

   426                     if ms_scope(const) in source_scopes:

   427                         self._set_source_for_term(source, const)

   428                         # if system source is used, add every rewritten constant

   429                         # to its supported terms even when associated entity

   430                         # doesn't actually come from it so we get a changes that

   431                         # allequals will return True as expected when computing

   432                         # needsplit

   433                         # check const is used in a relation restriction

   434                         if const.relation() and self.system_source in sourcesterms:

   435                             self._set_source_for_term(self.system_source, const)

   436         # add source for relations

   437         rschema = self._schema.rschema

   438         termssources = {}

   439         sourcerels = []

   440         for rel in self.rqlst.iget_nodes(Relation):

   441             # process non final relations only

   442             # note: don't try to get schema for 'is' relation (not available

   443             # during bootstrap)

   444             if rel.r_type == 'cw_source':

   445                 sourcerels.append(rel)

   446             if not (rel.is_types_restriction() or rschema(rel.r_type).final):

   447                 # nothing to do if relation is not supported by multiple sources

   448                 # or if some source has it listed in its cross_relations

   449                 # attribute

   450                 #

   451                 # XXX code below don't deal if some source allow relation

   452                 #     crossing but not another one

   453                 relsources = [s for s in repo.rel_type_sources(rel.r_type)

   454                                if s is self.system_source

   455                                or s in self._sourcesterms]

   456                 if len(relsources) < 2:

   457                     # filter out sources being there because they have this

   458                     # relation in their dont_cross_relations attribute

   459                     relsources = [source for source in relsources

   460                                   if source.support_relation(rel.r_type)]

   461                     if relsources:

   462                         # this means the relation is using a variable inlined as

   463                         # a constant and another unsupported variable, in which

   464                         # case we put the relation in sourcesterms

   465                         self._sourcesterms.setdefault(relsources[0], {})[rel] = set(self._solindices)

   466                     continue

   467                 lhs, rhs = rel.get_variable_parts()

   468                 lhsv, rhsv = getattr(lhs, 'variable', lhs), getattr(rhs, 'variable', rhs)

   469                 # update dictionary of sources supporting lhs and rhs vars

   470                 if not lhsv in termssources:

   471                     termssources[lhsv] = self._term_sources(lhs)

   472                 if not rhsv in termssources:

   473                     termssources[rhsv] = self._term_sources(rhs)

   474                 self._handle_cross_relation(rel, relsources, termssources)

   475                 self._linkedterms.setdefault(lhsv, set()).add((rhsv, rel))

   476                 self._linkedterms.setdefault(rhsv, set()).add((lhsv, rel))

   477         # extract information from cw_source relation

   478         for srel in sourcerels:

   479             vref = srel.children[1].children[0]

   480             sourceeids, sourcenames = [], []

   481             if isinstance(vref, Constant):

   482                 # simplified variable

   483                 sourceeids = None, (vref.eval(self.plan.args),)

   484                 var = vref

   485             else:

   486                 var = vref.variable

   487                 for rel in var.stinfo['relations'] - var.stinfo['rhsrelations']:

   488                     # skip neged eid relation since it's the kind of query

   489                     # generated when clearing old value of '?1" relation,

   490                     # cw_source included. See

   491                     # unittest_ldapuser.test_copy_to_system_source

   492                     if rel.r_type == 'name' or \

   493                        (rel.r_type == 'eid' and not rel.neged(strict=True)):

   494                         if rel.r_type == 'eid':

   495                             slist = sourceeids

   496                         else:

   497                             slist = sourcenames

   498                         sources = [cst.eval(self.plan.args)

   499                                    for cst in rel.children[1].get_nodes(Constant)]

   500                         if sources:

   501                             if slist:

   502                                 # don't attempt to do anything

   503                                 sourcenames = sourceeids = None

   504                                 break

   505                             slist[:] = (rel, sources)

   506             if sourceeids:

   507                 rel, values = sourceeids

   508                 sourcesdict = self._repo.sources_by_eid

   509             elif sourcenames:

   510                 rel, values = sourcenames

   511                 sourcesdict = self._repo.sources_by_uri

   512             else:

   513                 sourcesdict = None

   514             if sourcesdict is not None:

   515                 lhs = srel.children[0]

   516                 try:

   517                     sources = [sourcesdict[key] for key in values]

   518                 except KeyError:

   519                     raise BadRQLQuery('source conflict for term %s' % lhs.as_string())

   520                 if isinstance(lhs, Constant):

   521                     source = self._session.source_from_eid(lhs.eval(self.plan.args))

   522                     if not source in sources:

   523                         raise BadRQLQuery('source conflict for term %s' % lhs.as_string())

   524                 else:

   525                     lhs = getattr(lhs, 'variable', lhs)

   526                 invariant = getattr(lhs, '_q_invariant', False)

   527                 # XXX NOT NOT

   528                 neged = srel.neged(traverse_scope=True) or (rel and rel.neged(strict=True))

   529                 has_copy_based_source = False

   530                 sources_ = []

   531                 for source in sources:

   532                     if source.copy_based_source:

   533                         has_copy_based_source = True

   534                         if not self.system_source in sources_:

   535                             sources_.append(self.system_source)

   536                     else:

   537                         sources_.append(source)

   538                 sources = sources_

   539                 if neged:

   540                     for source in sources:

   541                         if invariant and source is self.system_source:

   542                             continue

   543                         self._remove_source_term(source, lhs)

   544                         self._discarded_sources.setdefault(lhs, set()).add(source)

   545                     usesys = self.system_source not in sources

   546                 else:

   547                     for source, terms in sourcesterms.items():

   548                         if lhs in terms and not source in sources:

   549                             if invariant and source is self.system_source:

   550                                 continue

   551                             self._remove_source_term(source, lhs)

   552                             self._discarded_sources.setdefault(lhs, set()).add(source)

   553                     usesys = self.system_source in sources

   554                 if rel is None or (len(var.stinfo['relations']) == 2 and

   555                                    not var.stinfo['selected']):

   556                     self._remove_source_term(self.system_source, var)

   557                     if not (has_copy_based_source or len(sources) > 1

   558                             or usesys or invariant):

   559                         if rel is None:

   560                             srel.parent.remove(srel)

   561                         else:

   562                             self.rqlst.undefine_variable(var)

   563                         self._remove_source_term(self.system_source, srel)

   564         return termssources

   565 

   566     def _handle_cross_relation(self, rel, relsources, termssources):

   567         for source in relsources:

   568             if rel.r_type in source.cross_relations:

   569                 ssource = self.system_source

   570                 crossvars = set(x.variable for x in rel.get_nodes(VariableRef))

   571                 for const in rel.get_nodes(Constant):

   572                     if source.uri != 'system' and not const in self._sourcesterms.get(source, ()):

   573                         continue

   574                     crossvars.add(const)

   575                 self._crossrelations.setdefault(source, {})[rel] = crossvars

   576                 if len(crossvars) < 2:

   577                     # this means there is a constant in the relation which is

   578                     # not supported by the source, so we can stop here

   579                     continue

   580                 self._sourcesterms.setdefault(ssource, {})[rel] = set(self._solindices)

   581                 solindices = None

   582                 for term in crossvars:

   583                     if len(termssources[term]) == 1 and iter(termssources[term]).next()[0].uri == 'system':

   584                         for ov in crossvars:

   585                             if ov is not term and (isinstance(ov, Constant) or ov._q_invariant):

   586                                 ssset = frozenset((ssource,))

   587                                 self._remove_sources(ov, termssources[ov] - ssset)

   588                         break

   589                     if solindices is None:

   590                         solindices = set(sol for s, sol in termssources[term]

   591                                          if s is source)

   592                     else:

   593                         solindices &= set(sol for s, sol in termssources[term]

   594                                           if s is source)

   595                 else:

   596                     self._sourcesterms.setdefault(source, {})[rel] = solindices

   597 

   598     def _remove_invalid_sources(self, termssources):

   599         """removes invalid sources from `sourcesterms` member according to

   600         traversed relations and their properties (which sources support them,

   601         can they cross sources, etc...)

   602         """

   603         for term in self._linkedterms:

   604             self._remove_sources_until_stable(term, termssources)

   605         if len(self._sourcesterms) > 1 and hasattr(self.plan.rqlst, 'main_relations'):

   606             # the querier doesn't annotate write queries, need to do it here

   607             self.plan.annotate_rqlst()

   608             # insert/update/delete queries, we may get extra information from

   609             # the main relation (eg relations to the left of the WHERE

   610             if self.plan.rqlst.TYPE == 'insert':

   611                 inserted = dict((vref.variable, etype)

   612                                 for etype, vref in self.plan.rqlst.main_variables)

   613             else:

   614                 inserted = {}

   615             repo = self._repo

   616             rschema = self._schema.rschema

   617             for rel in self.plan.rqlst.main_relations:

   618                 if not rschema(rel.r_type).final:

   619                     # nothing to do if relation is not supported by multiple sources

   620                     if len(repo.rel_type_sources(rel.r_type)) < 2:

   621                         continue

   622                     lhs, rhs = rel.get_variable_parts()

   623                     try:

   624                         lhsv = self._extern_term(lhs, termssources, inserted)

   625                         rhsv = self._extern_term(rhs, termssources, inserted)

   626                     except KeyError:

   627                         continue

   628                     self._remove_term_sources(lhsv, rel, rhsv, termssources)

   629                     self._remove_term_sources(rhsv, rel, lhsv, termssources)

   630 

   631     def _extern_term(self, term, termssources, inserted):

   632         var = term.variable

   633         if var.stinfo['constnode']:

   634             termv = var.stinfo['constnode']

   635             termssources[termv] = self._term_sources(termv)

   636         elif var in inserted:

   637             termv = var

   638             source = self._repo.locate_etype_source(inserted[var])

   639             termssources[termv] = set((source, solindex)

   640                                       for solindex in self._solindices)

   641         else:

   642             termv = self.rqlst.defined_vars[var.name]

   643             if not termv in termssources:

   644                 termssources[termv] = self._term_sources(termv)

   645         return termv

   646 

   647     def _remove_sources_until_stable(self, term, termssources):

   648         sourcesterms = self._sourcesterms

   649         for oterm, rel in self._linkedterms.get(term, ()):

   650             tscope = ms_scope(term)

   651             otscope = ms_scope(oterm)

   652             rscope = ms_scope(rel)

   653             if not tscope is otscope and rscope.neged(strict=True):

   654                 # can't get information from relation inside a NOT exists

   655                 # where terms don't belong to the same scope

   656                 continue

   657             need_ancestor_scope = False

   658             if not (tscope is rscope and otscope is rscope):

   659                 if rel.ored():

   660                     continue

   661                 if rel.ored(traverse_scope=True):

   662                     # if relation has some OR as parent, constraints should only

   663                     # propagate from parent scope to child scope, nothing else

   664                     need_ancestor_scope = True

   665             relsources = self._repo.rel_type_sources(rel.r_type)

   666             if neged_relation(rel) and (

   667                 len(relsources) < 2

   668                 or not isinstance(oterm, Variable)

   669                 or oterm.valuable_references() != 1

   670                 or any(sourcesterms[source][term] != sourcesterms[source][oterm]

   671                        for source in relsources

   672                        if term in sourcesterms.get(source, ())

   673                        and oterm in sourcesterms.get(source, ()))):

   674                 # neged relation doesn't allow to infer term sources unless

   675                 # we're on a multisource relation for a term only used by this

   676                 # relation (eg "Any X WHERE NOT X multisource_rel Y" and over is

   677                 # Y)

   678                 continue

   679             # compute invalid sources for terms and remove them

   680             if not need_ancestor_scope or is_ancestor(tscope, otscope):

   681                 self._remove_term_sources(term, rel, oterm, termssources)

   682             if not need_ancestor_scope or is_ancestor(otscope, tscope):

   683                 self._remove_term_sources(oterm, rel, term, termssources)

   684 

   685     def _remove_term_sources(self, term, rel, oterm, termssources):

   686         """remove invalid sources for term according to oterm's sources and the

   687         relation between those two terms.

   688         """

   689         norelsup = self._norel_support_set(rel)

   690         termsources = termssources[term]

   691         invalid_sources = termsources - (termssources[oterm] | norelsup)

   692         if invalid_sources and self._repo.can_cross_relation(rel.r_type):

   693             invalid_sources -= self._sys_source_set

   694             if invalid_sources and isinstance(term, Variable) \

   695                    and self._need_ext_source_access(term, rel):

   696                 # if the term is a not invariant variable, we should filter out

   697                 # source where the relation is a cross relation from invalid

   698                 # sources

   699                 invalid_sources = frozenset((s, solidx) for s, solidx in invalid_sources

   700                                             if not (s in self._crossrelations and

   701                                                     rel in self._crossrelations[s]))

   702         if invalid_sources:

   703             self._remove_sources(term, invalid_sources)

   704             discarded = self._discarded_sources.get(term)

   705             if discarded is not None and not any(x[0] for x in (termsources-invalid_sources)

   706                                                  if not x[0] in discarded):

   707                 raise BadRQLQuery('relation %s cant be crossed but %s and %s should '

   708                               'come from difference sources' %

   709                               (rel.r_type, term.as_string(), oterm.as_string()))

   710             # if term is a rewritten const, we can apply the same changes to

   711             # all other consts inserted from the same original variable

   712             for const in self._const_vars.get(term, ()):

   713                 if const is not term:

   714                     self._remove_sources(const, invalid_sources)

   715             termsources -= invalid_sources

   716             self._remove_sources_until_stable(term, termssources)

   717             if isinstance(oterm, Constant):

   718                 self._remove_sources(oterm, invalid_sources)

   719 

   720     def _compute_needsplit(self):

   721         """tell according to sourcesterms if the rqlst has to be splitted for

   722         execution among multiple sources

   723 

   724         the execution has to be split if

   725         * a source support an entity (non invariant) but doesn't support a

   726           relation on it

   727         * a source support an entity which is accessed by an optional relation

   728         * there is more than one source and either all sources'supported

   729           variable/solutions are not equivalent or multiple variables have to

   730           be fetched from some source

   731         """

   732         # NOTE: < 2 since may be 0 on queries such as Any X WHERE X eid 2

   733         if len(self._sourcesterms) < 2:

   734             self.needsplit = False

   735             # if this is not the system source but we have only constant terms

   736             # and no relation (other than eid), apply query on the system source

   737             #

   738             # testing for rqlst with nothing in vargraph nor defined_vars is the

   739             # simplest way the check the condition explained below

   740             if not self.system_source in self._sourcesterms and \

   741                    not self.rqlst.defined_vars and \

   742                    not need_source_access_relation(self.rqlst.vargraph):

   743                 self._sourcesterms = {self.system_source: {}}

   744         elif not self.needsplit:

   745             if not allequals(self._sourcesterms.itervalues()):

   746                 for source, terms in self._sourcesterms.iteritems():

   747                     if source is self.system_source:

   748                         continue

   749                     if any(x for x in terms if not isinstance(x, Constant)):

   750                         self.needsplit = True

   751                         return

   752                 self._sourcesterms = {self.system_source: {}}

   753                 self.needsplit = False

   754             else:

   755                 sample = self._sourcesterms.itervalues().next()

   756                 if len(sample) > 1:

   757                     for term in sample:

   758                         # need split if unlinked variable

   759                         if isinstance(term, Variable) and not term in self._linkedterms:

   760                             self.needsplit = True

   761                             break

   762                     else:

   763                         # need split if there are some cross relation on non

   764                         # invariant variable or if the variable is used in

   765                         # multi-sources relation

   766                         if self._crossrelations:

   767                             for reldict in self._crossrelations.itervalues():

   768                                 for rel, terms in reldict.iteritems():

   769                                     for term in terms:

   770                                         if isinstance(term, Variable) \

   771                                                and self._need_ext_source_access(term, rel):

   772                                             self.needsplit = True

   773                                             return

   774         else:

   775             # remove sources only accessing to constant nodes

   776             for source, terms in self._sourcesterms.items():

   777                 if source is self.system_source:

   778                     continue

   779                 if not any(x for x in terms if not isinstance(x, Constant)):

   780                     del self._sourcesterms[source]

   781             if len(self._sourcesterms) < 2:

   782                 self.needsplit = False

   783 

   784     @cached

   785     def _need_ext_source_access(self, var, rel):

   786         if not var._q_invariant:

   787             return True

   788         if  any(r for x, r in self._linkedterms[var]

   789                 if not r is rel and self._repo.is_multi_sources_relation(r.r_type)):

   790             return True

   791         return False

   792 

   793     def _set_source_for_term(self, source, term):

   794         self._sourcesterms.setdefault(source, {})[term] = set(self._solindices)

   795 

   796     def _term_sources(self, term):

   797         """returns possible sources for terms `term`"""

   798         if isinstance(term, Constant):

   799             source = self._session.source_from_eid(term.eval(self.plan.args))

   800             return set((source, solindex) for solindex in self._solindices)

   801         else:

   802             var = getattr(term, 'variable', term)

   803             sources = [source for source, varobjs in self.sourcesterms.iteritems()

   804                        if var in varobjs]

   805             return set((source, solindex) for source in sources

   806                        for solindex in self.sourcesterms[source][var])

   807 

   808     def _remove_sources(self, term, sources):

   809         """removes invalid sources (`sources`) from `sourcesterms`

   810 

   811         :param sources: the list of sources to remove

   812         :param term: the analyzed term

   813         """

   814         sourcesterms = self._sourcesterms

   815         for source, solindex in sources:

   816             try:

   817                 sourcesterms[source][term].remove(solindex)

   818             except KeyError:

   819                 import rql.base as rqlb

   820                 assert isinstance(term, (rqlb.BaseNode, Variable)), repr(term)

   821                 continue # may occur with subquery column alias

   822             if not sourcesterms[source][term]:

   823                 self._remove_source_term(source, term)

   824 

   825     def _remove_source_term(self, source, term):

   826         try:

   827             poped = self._sourcesterms[source].pop(term, None)

   828         except KeyError:

   829             pass

   830         else:

   831             if not self._sourcesterms[source]:

   832                 del self._sourcesterms[source]

   833 

   834     def crossed_relation(self, source, relation):

   835         return relation in self._crossrelations.get(source, ())

   836 

   837     def part_steps(self):

   838         """precompute necessary part steps before generating actual rql for

   839         each step. This is necessary to know if an aggregate step will be

   840         necessary or not.

   841         """

   842         steps = []

   843         select = self.rqlst

   844         rschema = self._schema.rschema

   845         for source in self.part_sources:

   846             try:

   847                 sourceterms = self._sourcesterms[source]

   848             except KeyError:

   849                 continue # already proceed

   850             while sourceterms:

   851                 # take a term randomly, and all terms supporting the

   852                 # same solutions

   853                 term, solindices = self._choose_term(source, sourceterms)

   854                 if source.uri == 'system':

   855                     # ensure all variables are available for the latest step

   856                     # (missing one will be available from temporary tables

   857                     # of previous steps)

   858                     scope = select

   859                     terms = scope.defined_vars.values() + scope.aliases.values()

   860                     sourceterms.clear()

   861                     sources = [source]

   862                 else:

   863                     scope = ms_scope(term)

   864                     # find which sources support the same term and solutions

   865                     sources = self._expand_sources(source, term, solindices)

   866                     # no try to get as much terms as possible

   867                     terms = self._expand_terms(term, sources, sourceterms,

   868                                                scope, solindices)

   869                     if len(terms) == 1 and isinstance(terms[0], Constant):

   870                         # we can't generate anything interesting with a single

   871                         # constant term (will generate an empty "Any" query),

   872                         # go to the next iteration directly!

   873                         continue

   874                     if not sourceterms:

   875                         try:

   876                             del self._sourcesterms[source]

   877                         except KeyError:

   878                             # XXX already cleaned

   879                             pass

   880                 # set of terms which should be additionaly selected when

   881                 # possible

   882                 needsel = set()

   883                 if not self._sourcesterms and scope is select:

   884                     terms += scope.defined_vars.values() + scope.aliases.values()

   885                     if isinstance(term, Relation) and len(sources) > 1:

   886                         variants = set()

   887                         partterms = [term]

   888                         for vref in term.get_nodes(VariableRef):

   889                             if not vref.variable._q_invariant:

   890                                 variants.add(vref.name)

   891                         if len(variants) == 2:

   892                             # we need an extra-step to fetch relations from each source

   893                             # before a join with prefetched inputs

   894                             # (see test_crossed_relation_noeid_needattr in

   895                             #  unittest_msplanner / unittest_multisources)

   896                             lhs, rhs = term.get_variable_parts()

   897                             steps.append( (sources, [term, getattr(lhs, 'variable', lhs),

   898                                                      getattr(rhs, 'variable', rhs)],

   899                                            solindices, scope, variants, False) )

   900                             sources = [self.system_source]

   901                     final = True

   902                 else:

   903                     # suppose this is a final step until the contrary is proven

   904                     final = scope is select

   905                     # add attribute variables and mark variables which should be

   906                     # additionaly selected when possible

   907                     for var in select.defined_vars.itervalues():

   908                         if not var in terms:

   909                             stinfo = var.stinfo

   910                             for ovar, rtype in stinfo.get('attrvars', ()):

   911                                 if ovar in terms:

   912                                     needsel.add(var.name)

   913                                     terms.append(var)

   914                                     break

   915                             else:

   916                                 needsel.add(var.name)

   917                                 final = False

   918                     # check all relations are supported by the sources

   919                     for rel in scope.iget_nodes(Relation):

   920                         if rel.is_types_restriction():

   921                             continue

   922                         # take care not overwriting the existing "source" identifier

   923                         for _source in sources:

   924                             if not _source.support_relation(rel.r_type) or (

   925                                 self.crossed_relation(_source, rel) and not rel in terms):

   926                                 for vref in rel.iget_nodes(VariableRef):

   927                                     needsel.add(vref.name)

   928                                 final = False

   929                                 break

   930                         else:

   931                             if not scope is select:

   932                                 self._exists_relation(rel, terms, needsel, source)

   933                             # if relation is supported by all sources and some of

   934                             # its lhs/rhs variable isn't in "terms", and the

   935                             # other end *is* in "terms", mark it have to be

   936                             # selected

   937                             if source.uri != 'system' and not rschema(rel.r_type).final:

   938                                 lhs, rhs = rel.get_variable_parts()

   939                                 try:

   940                                     lhsvar = lhs.variable

   941                                 except AttributeError:

   942                                     lhsvar = lhs

   943                                 try:

   944                                     rhsvar = rhs.variable

   945                                 except AttributeError:

   946                                     rhsvar = rhs

   947                                 try:

   948                                     if lhsvar in terms and not rhsvar in terms:

   949                                         needsel.add(lhsvar.name)

   950                                     elif rhsvar in terms and not lhsvar in terms:

   951                                         needsel.add(rhsvar.name)

   952                                 except AttributeError:

   953                                     continue # not an attribute, no selection needed

   954                 if final and source.uri != 'system':

   955                     # check rewritten constants

   956                     for vconsts in select.stinfo['rewritten'].itervalues():

   957                         const = vconsts[0]

   958                         eid = const.eval(self.plan.args)

   959                         _source = self._session.source_from_eid(eid)

   960                         if len(sources) > 1 or not _source in sources:

   961                             # if there is some rewriten constant used by a not

   962                             # neged relation while there are some source not

   963                             # supporting the associated entity, this step can't

   964                             # be final (unless the relation is explicitly in

   965                             # `terms`, eg cross relations)

   966                             for c in vconsts:

   967                                 rel = c.relation()

   968                                 if rel is None or not (rel in terms or neged_relation(rel)):

   969                                     final = False

   970                                     break

   971                             break

   972                 if final:

   973                     self._cleanup_sourcesterms(sources, solindices)

   974                 steps.append((sources, terms, solindices, scope, needsel, final)

   975                              )

   976         if not steps[-1][-1]:

   977             # add a final step

   978             terms = select.defined_vars.values() + select.aliases.values()

   979             steps.append( ([self.system_source], terms, set(self._solindices),

   980                            select, set(), True) )

   981         return steps

   982 

   983     def _exists_relation(self, rel, terms, needsel, source):

   984         rschema = self._schema.rschema(rel.r_type)

   985         lhs, rhs = rel.get_variable_parts()

   986         try:

   987             lhsvar, rhsvar = lhs.variable, rhs.variable

   988         except AttributeError:

   989             pass

   990         else:

   991             # supported relation with at least one end supported, check the

   992             # other end is in as well. If not this usually means the

   993             # variable is refed by an outer scope and should be substituted

   994             # using an 'identity' relation (else we'll get a conflict of

   995             # temporary tables)

   996             relscope = ms_scope(rel)

   997             lhsscope = ms_scope(lhsvar)

   998             rhsscope = ms_scope(rhsvar)

   999             if rhsvar in terms and not lhsvar in terms and lhsscope is lhsvar.stmt:

  1000                 self._identity_substitute(rel, lhsvar, terms, needsel, relscope)

  1001             elif lhsvar in terms and not rhsvar in terms and rhsscope is rhsvar.stmt:

  1002                 self._identity_substitute(rel, rhsvar, terms, needsel, relscope)

  1003             elif self.crossed_relation(source, rel):

  1004                 if lhsscope is not relscope:

  1005                     self._identity_substitute(rel, lhsvar, terms, needsel,

  1006                                               relscope, lhsscope)

  1007                 if rhsscope is not relscope:

  1008                     self._identity_substitute(rel, rhsvar, terms, needsel,

  1009                                               relscope, rhsscope)

  1010 

  1011     def _identity_substitute(self, relation, var, terms, needsel, exist,

  1012                              idrelscope=None):

  1013         newvar = self._insert_identity_variable(exist, var, idrelscope)

  1014         # ensure relation is using '=' operator, else we rely on a

  1015         # sqlgenerator side effect (it won't insert an inequality operator

  1016         # in this case)

  1017         relation.children[1].operator = '='

  1018         terms.append(newvar)

  1019         needsel.add(newvar.name)

  1020 

  1021     def _choose_term(self, source, sourceterms):

  1022         """pick one term among terms supported by a source, which will be used

  1023         as a base to generate an execution step

  1024         """

  1025         secondchoice = None

  1026         if len(self._sourcesterms) > 1:

  1027             # first, return non invariant variable of crossed relation, then the

  1028             # crossed relation itself

  1029             for term in sourceterms:

  1030                 if (isinstance(term, Relation)

  1031                     and self.crossed_relation(source, term)

  1032                     and not ms_scope(term) is self.rqlst):

  1033                     for vref in term.get_variable_parts():

  1034                         try:

  1035                             var = vref.variable

  1036                         except AttributeError:

  1037                             # Constant

  1038                             continue

  1039                         if ((len(var.stinfo['relations']) > 1 or var.stinfo['selected'])

  1040                             and var in sourceterms):

  1041                             return var, sourceterms.pop(var)

  1042                     return term, sourceterms.pop(term)

  1043             # priority to variable from subscopes

  1044             for term in sourceterms:

  1045                 if not ms_scope(term) is self.rqlst:

  1046                     if isinstance(term, Variable):

  1047                         return term, sourceterms.pop(term)

  1048                     secondchoice = term

  1049         else:

  1050             # priority to variable from outer scope

  1051             for term in sourceterms:

  1052                 if ms_scope(term) is self.rqlst:

  1053                     if isinstance(term, Variable):

  1054                         return term, sourceterms.pop(term)

  1055                     secondchoice = term

  1056         if secondchoice is not None:

  1057             return secondchoice, sourceterms.pop(secondchoice)

  1058         # priority to variable with the less solutions supported and with the

  1059         # most valuable refs. Add variable name for test predictability

  1060         variables = sorted([(var, sols) for (var, sols) in sourceterms.items()

  1061                             if isinstance(var, Variable)],

  1062                            key=lambda (v, s): (len(s), -v.valuable_references(), v.name))

  1063         if variables:

  1064             var = variables[0][0]

  1065             return var, sourceterms.pop(var)

  1066         # priority to constant

  1067         for term in sourceterms:

  1068             if isinstance(term, Constant):

  1069                 return term, sourceterms.pop(term)

  1070         # whatever (relation)

  1071         term = iter(sourceterms).next()

  1072         return term, sourceterms.pop(term)

  1073 

  1074     def _expand_sources(self, selected_source, term, solindices):

  1075         """return all sources supporting given term / solindices"""

  1076         sources = [selected_source]

  1077         sourcesterms = self._sourcesterms

  1078         for source in list(sourcesterms):

  1079             if source is selected_source:

  1080                 continue

  1081             if not (term in sourcesterms[source] and

  1082                     solindices.issubset(sourcesterms[source][term])):

  1083                 continue

  1084             sources.append(source)

  1085             if source.uri != 'system' or not (isinstance(term, Variable) and not term in self._linkedterms):

  1086                 termsolindices = sourcesterms[source][term]

  1087                 termsolindices -= solindices

  1088                 if not termsolindices:

  1089                     del sourcesterms[source][term]

  1090                     if not sourcesterms[source]:

  1091                         del sourcesterms[source]

  1092         return sources

  1093 

  1094     def _expand_terms(self, term, sources, sourceterms, scope, solindices):

  1095         terms = [term]

  1096         sources = sorted(sources)

  1097         sourcesterms = self._sourcesterms

  1098         linkedterms = self._linkedterms

  1099         # term has to belong to the same scope if there is more

  1100         # than the system source remaining

  1101         if len(sourcesterms) > 1 and not scope is self.rqlst:

  1102             candidates = (t for t in sourceterms if scope is ms_scope(t))

  1103         else:

  1104             candidates = sourceterms

  1105         # we only want one unlinked term in each generated query

  1106         candidates = [t for t in candidates

  1107                       if isinstance(t, (Constant, Relation)) or

  1108                       (solindices.issubset(sourceterms[t]) and t in linkedterms)]

  1109         cross_rels = {}

  1110         for source in sources:

  1111             cross_rels.update(self._crossrelations.get(source, {}))

  1112         exclude = {}

  1113         for crossvars in cross_rels.itervalues():

  1114             vars = [t for t in crossvars if isinstance(t, Variable)]

  1115             try:

  1116                 exclude[vars[0]] = vars[1]

  1117                 exclude[vars[1]] = vars[0]

  1118             except IndexError:

  1119                 pass

  1120         accept_term = lambda x: (not any(s for s in sources

  1121                                          if not x in sourcesterms.get(s, ()))

  1122                                  and x._ms_may_be_processed(terms, linkedterms)

  1123                                  and not exclude.get(x) in terms)

  1124         if isinstance(term, Relation) and term in cross_rels:

  1125             cross_terms = cross_rels.pop(term)

  1126             base_accept_term = accept_term

  1127             accept_term = lambda x: (base_accept_term(x) or x in cross_terms)

  1128             for refed in cross_terms:

  1129                 if not refed in candidates:

  1130                     terms.append(refed)

  1131         # repeat until no term can't be added, since addition of a new

  1132         # term may permit to another one to be added

  1133         modified = True

  1134         while modified and candidates:

  1135             modified = False