Source code for rdflib.plugins.sparql.algebra

Converting the 'parse-tree' output of pyparsing to a SPARQL Algebra expression


import functools
import operator
import collections

from rdflib import Literal, Variable, URIRef, BNode

from rdflib.plugins.sparql.sparql import Prologue, Query
from rdflib.plugins.sparql.parserutils import CompValue, Expr
from rdflib.plugins.sparql.operators import (
    and_, TrueFilter, simplify as simplifyFilters)
from rdflib.paths import (
    InvPath, AlternativePath, SequencePath, MulPath, NegatedPath)

from pyparsing import ParseResults

# ---------------------------
# Some convenience methods
[docs]def OrderBy(p, expr): return CompValue('OrderBy', p=p, expr=expr)
[docs]def ToMultiSet(p): return CompValue('ToMultiSet', p=p)
[docs]def Union(p1, p2): return CompValue('Union', p1=p1, p2=p2)
[docs]def Join(p1, p2): return CompValue('Join', p1=p1, p2=p2)
[docs]def Minus(p1, p2): return CompValue('Minus', p1=p1, p2=p2)
[docs]def Graph(term, graph): return CompValue('Graph', term=term, p=graph)
[docs]def BGP(triples=None): return CompValue('BGP', triples=triples or [])
[docs]def LeftJoin(p1, p2, expr): return CompValue('LeftJoin', p1=p1, p2=p2, expr=expr)
[docs]def Filter(expr, p): return CompValue('Filter', expr=expr, p=p)
[docs]def Extend(p, expr, var): return CompValue('Extend', p=p, expr=expr, var=var)
[docs]def Values(res): return CompValue('values', res=res)
[docs]def Project(p, PV): return CompValue('Project', p=p, PV=PV)
[docs]def Group(p, expr=None): return CompValue('Group', p=p, expr=expr)
def _knownTerms(triple, varsknown, varscount): return (len(filter(None, (x not in varsknown and isinstance( x, (Variable, BNode)) for x in triple))), -sum(varscount.get(x, 0) for x in triple), not isinstance(triple[2], Literal), )
[docs]def reorderTriples(l): """ Reorder triple patterns so that we execute the ones with most bindings first """ def _addvar(term, varsknown): if isinstance(term, (Variable, BNode)): varsknown.add(term) l = [(None, x) for x in l] varsknown = set() varscount = collections.defaultdict(int) for t in l: for c in t[1]: if isinstance(c, (Variable, BNode)): varscount[c] += 1 i = 0 # Done in steps, sort by number of bound terms # the top block of patterns with the most bound terms is kept # the rest is resorted based on the vars bound after the first # block is evaluated # we sort by decorate/undecorate, since we need the value of the sort keys while i < len(l): l[i:] = sorted((_knownTerms(x[ 1], varsknown, varscount), x[1]) for x in l[i:]) t = l[i][0][0] # top block has this many terms bound j = 0 while i+j < len(l) and l[i+j][0][0] == t: for c in l[i+j][1]: _addvar(c, varsknown) j += 1 i += 1 return [x[1] for x in l]
[docs]def triples(l): l = reduce(lambda x, y: x + y, l) if (len(l) % 3) != 0: raise Exception('these aint triples') return reorderTriples((l[x], l[x + 1], l[x + 2]) for x in range(0, len(l), 3))
[docs]def translatePName(p, prologue): """ Expand prefixed/relative URIs """ if isinstance(p, CompValue): if == 'pname': return prologue.absolutize(p) if == 'literal': return Literal(p.string, lang=p.lang, datatype=prologue.absolutize(p.datatype)) elif isinstance(p, URIRef): return prologue.absolutize(p)
[docs]def translatePath(p): """ Translate PropertyPath expressions """ if isinstance(p, CompValue): if == 'PathAlternative': if len(p.part) == 1: return p.part[0] else: return AlternativePath(*p.part) elif == 'PathSequence': if len(p.part) == 1: return p.part[0] else: return SequencePath(*p.part) elif == 'PathElt': if not p.mod: return p.part else: if isinstance(p.part, list): if len(p.part) != 1: raise Exception('Denkfehler!') return MulPath(p.part[0], p.mod) else: return MulPath(p.part, p.mod) elif == 'PathEltOrInverse': if isinstance(p.part, list): if len(p.part) != 1: raise Exception('Denkfehler!') return InvPath(p.part[0]) else: return InvPath(p.part) elif == 'PathNegatedPropertySet': if isinstance(p.part, list): return NegatedPath(AlternativePath(*p.part)) else: return NegatedPath(p.part)
[docs]def translateExists(e): """ Translate the graph pattern used by EXISTS and NOT EXISTS """ def _c(n): if isinstance(n, CompValue): if in ('Builtin_EXISTS', 'Builtin_NOTEXISTS'): n.graph = translateGroupGraphPattern(n.graph) if == 'Filter': # filters inside (NOT) EXISTS can see vars bound outside n.graph.no_isolated_scope = True e = traverse(e, visitPost=_c) return e
[docs]def collectAndRemoveFilters(parts): """ FILTER expressions apply to the whole group graph pattern in which they appear. """ filters = [] i = 0 while i < len(parts): p = parts[i] if == 'Filter': filters.append(translateExists(p.expr)) parts.pop(i) else: i += 1 if filters: return and_(*filters) return None
[docs]def translateGroupOrUnionGraphPattern(graphPattern): A = None for g in graphPattern.graph: g = translateGroupGraphPattern(g) if not A: A = g else: A = Union(A, g) return A
[docs]def translateGraphGraphPattern(graphPattern): return Graph(graphPattern.term, translateGroupGraphPattern(graphPattern.graph))
[docs]def translateInlineData(graphPattern): return ToMultiSet(translateValues(graphPattern))
[docs]def translateGroupGraphPattern(graphPattern): """ """ if == 'SubSelect': return ToMultiSet(translate(graphPattern)[0]) if not graphPattern.part: graphPattern.part = [] # empty { } filters = collectAndRemoveFilters(graphPattern.part) g = [] for p in graphPattern.part: if == 'TriplesBlock': # merge adjacent TripleBlocks if not (g and g[-1].name == 'BGP'): g.append(BGP()) g[-1]["triples"] += triples(p.triples) else: g.append(p) G = BGP() for p in g: if == 'OptionalGraphPattern': A = translateGroupGraphPattern(p.graph) if == 'Filter': G = LeftJoin(G, A.p, A.expr) else: G = LeftJoin(G, A, TrueFilter) elif == 'MinusGraphPattern': G = Minus(p1=G, p2=translateGroupGraphPattern(p.graph)) elif == 'GroupOrUnionGraphPattern': G = Join(p1=G, p2=translateGroupOrUnionGraphPattern(p)) elif == 'GraphGraphPattern': G = Join(p1=G, p2=translateGraphGraphPattern(p)) elif == 'InlineData': G = Join(p1=G, p2=translateInlineData(p)) elif == 'ServiceGraphPattern': G = Join(p1=G, p2=p) elif in ('BGP', 'Extend'): G = Join(p1=G, p2=p) elif == 'Bind': G = Extend(G, p.expr, p.var) else: raise Exception('Unknown part in GroupGraphPattern: %s - %s' % (type(p), if filters: G = Filter(expr=filters, p=G) return G
[docs]class StopTraversal(Exception):
[docs] def __init__(self, rv): self.rv = rv
def _traverse(e, visitPre=lambda n: None, visitPost=lambda n: None): """ Traverse a parse-tree, visit each node if visit functions return a value, replace current node """ _e = visitPre(e) if _e is not None: return _e if e is None: return None if isinstance(e, (list, ParseResults)): return [_traverse(x, visitPre, visitPost) for x in e] elif isinstance(e, tuple): return tuple([_traverse(x, visitPre, visitPost) for x in e]) elif isinstance(e, CompValue): for k, val in e.iteritems(): e[k] = _traverse(val, visitPre, visitPost) _e = visitPost(e) if _e is not None: return _e return e def _traverseAgg(e, visitor=lambda n, v: None): """ Traverse a parse-tree, visit each node if visit functions return a value, replace current node """ res = [] if isinstance(e, (list, ParseResults, tuple)): res = [_traverseAgg(x, visitor) for x in e] elif isinstance(e, CompValue): for k, val in e.iteritems(): if val != None: res.append(_traverseAgg(val, visitor)) return visitor(e, res)
[docs]def traverse( tree, visitPre=lambda n: None, visitPost=lambda n: None, complete=None): """ Traverse tree, visit each node with visit function visit function may raise StopTraversal to stop traversal if complete!=None, it is returned on complete traversal, otherwise the transformed tree is returned """ try: r = _traverse(tree, visitPre, visitPost) if complete is not None: return complete return r except StopTraversal, st: return st.rv
def _hasAggregate(x): """ Traverse parse(sub)Tree return true if any aggregates are used """ if isinstance(x, CompValue): if'Aggregate_'): raise StopTraversal(True) def _aggs(e, A): """ Collect Aggregates in A replaces aggregates with variable references """ # TODO: nested Aggregates? if isinstance(e, CompValue) and'Aggregate_'): A.append(e) aggvar = Variable('__agg_%d__' % len(A)) e["res"] = aggvar return aggvar def _findVars(x, res): """ Find all variables in a tree """ if isinstance(x, Variable): res.add(x) if isinstance(x, CompValue): if == "Bind": res.add(x.var) return x # stop recursion and finding vars in the expr elif == 'SubSelect': if x.projection: res.update(v.var or v.evar for v in x.projection) return x def _addVars(x, children): """ find which variables may be bound by this part of the query """ if isinstance(x, Variable): return set([x]) elif isinstance(x, CompValue): if == "RelationalExpression": x["_vars"] = set() elif == "Extend": # vars only used in the expr for a bind should not be included x["_vars"] = reduce(operator.or_, [ child for child,part in zip(children,x) if part!='expr' ], set()) else: x["_vars"] = set(reduce(operator.or_, children, set())) if == 'SubSelect': if x.projection: s = set(v.var or v.evar for v in x.projection) else: s = set() return s return x["_vars"] return reduce(operator.or_, children, set()) def _sample(e, v=None): """ For each unaggregated variable V in expr Replace V with Sample(V) """ if isinstance(e, CompValue) and"Aggregate_"): return e # do not replace vars in aggregates if isinstance(e, Variable) and v != e: return CompValue('Aggregate_Sample', vars=e) def _simplifyFilters(e): if isinstance(e, Expr): return simplifyFilters(e)
[docs]def translateAggregates(q, M): E = [] A = [] # collect/replace aggs in : # select expr as ?var if q.projection: for v in q.projection: if v.evar: v.expr = traverse(v.expr, functools.partial(_sample, v=v.evar)) v.expr = traverse(v.expr, functools.partial(_aggs, A=A)) # having clause if traverse(q.having, _hasAggregate, complete=False): q.having = traverse(q.having, _sample) traverse(q.having, functools.partial(_aggs, A=A)) # order by if traverse(q.orderby, _hasAggregate, complete=False): q.orderby = traverse(q.orderby, _sample) traverse(q.orderby, functools.partial(_aggs, A=A)) # sample all other select vars # TODO: only allowed for vars in group-by? if q.projection: for v in q.projection: if v.var: rv = Variable('__agg_%d__' % (len(A) + 1)) A.append(CompValue('Aggregate_Sample', vars=v.var, res=rv)) E.append((rv, v.var)) return CompValue('AggregateJoin', A=A, p=M), E
[docs]def translateValues(v): # if len(v.var)!=len(v.value): # raise Exception("Unmatched vars and values in ValueClause: "+str(v)) res = [] if not v.var: return res if not v.value: return res if not isinstance(v.value[0], list): for val in v.value: res.append({v.var[0]: val}) else: for vals in v.value: res.append(dict(zip(v.var, vals))) return Values(res)
[docs]def translate(q): """ """ _traverse(q, _simplifyFilters) q.where = traverse(q.where, visitPost=translatePath) # TODO: Var scope test VS = set() traverse(q.where, functools.partial(_findVars, res=VS)) # all query types have a where part M = translateGroupGraphPattern(q.where) aggregate = False if q.groupby: conditions = [] # convert "GROUP BY (?expr as ?var)" to an Extend for c in q.groupby.condition: if isinstance(c, CompValue) and == 'GroupAs': M = Extend(M, c.expr, c.var) c = c.var conditions.append(c) M = Group(p=M, expr=conditions) aggregate = True elif traverse(q.having, _hasAggregate, complete=False) or \ traverse(q.orderby, _hasAggregate, complete=False) or \ any(traverse(x.expr, _hasAggregate, complete=False) for x in q.projection or [] if x.evar): # if any aggregate is used, implicit group by M = Group(p=M) aggregate = True if aggregate: M, E = translateAggregates(q, M) else: E = [] # HAVING if q.having: M = Filter(expr=and_(*q.having.condition), p=M) # VALUES if q.valuesClause: M = Join(p1=M, p2=ToMultiSet(translateValues(q.valuesClause))) if not q.projection: # select * PV = list(VS) else: PV = list() for v in q.projection: if v.var: if v not in PV: PV.append(v.var) elif v.evar: if v not in PV: PV.append(v.evar) E.append((v.expr, v.evar)) else: raise Exception("I expected a var or evar here!") for e, v in E: M = Extend(M, e, v) # ORDER BY if q.orderby: M = OrderBy(M, [CompValue('OrderCondition', expr=c.expr, order=c.order) for c in q.orderby.condition]) # PROJECT M = Project(M, PV) if q.modifier: if q.modifier == 'DISTINCT': M = CompValue('Distinct', p=M) elif q.modifier == 'REDUCED': M = CompValue('Reduced', p=M) if q.limitoffset: offset = 0 if q.limitoffset.offset != None: offset = q.limitoffset.offset.toPython() if q.limitoffset.limit != None: M = CompValue('Slice', p=M, start=offset, length=q.limitoffset.limit.toPython()) else: M = CompValue('Slice', p=M, start=offset) return M, PV
[docs]def simplify(n): """Remove joins to empty BGPs""" if isinstance(n, CompValue): if == 'Join': if == 'BGP' and len(n.p1.triples) == 0: return n.p2 if == 'BGP' and len(n.p2.triples) == 0: return n.p1 elif == 'BGP': n["triples"] = reorderTriples(n.triples) return n
[docs]def analyse(n, children): """ Some things can be lazily joined. This propegates whether they can up the tree and sets lazy flags for all joins """ if isinstance(n, CompValue): if == 'Join': n["lazy"] = all(children) return False elif in ('Slice', 'Distinct'): return False else: return all(children) else: return True
[docs]def translatePrologue(p, base, initNs=None, prologue=None): if prologue is None: prologue = Prologue() prologue.base = "" if base: prologue.base = base if initNs: for k, v in initNs.iteritems(): prologue.bind(k, v) for x in p: if == 'Base': prologue.base = x.iri elif == 'PrefixDecl': prologue.bind(x.prefix, prologue.absolutize(x.iri)) return prologue
[docs]def translateQuads(quads): if quads.triples: alltriples = triples(quads.triples) else: alltriples = [] allquads = collections.defaultdict(list) if quads.quadsNotTriples: for q in quads.quadsNotTriples: if q.triples: allquads[q.term] += triples(q.triples) return alltriples, allquads
[docs]def translateUpdate1(u, prologue): if in ('Load', 'Clear', 'Drop', 'Create'): pass # no translation needed elif in ('Add', 'Move', 'Copy'): pass elif in ('InsertData', 'DeleteData', 'DeleteWhere'): t, q = translateQuads(u.quads) u["quads"] = q u["triples"] = t if in ('DeleteWhere', 'DeleteData'): pass # TODO: check for bnodes in triples elif == 'Modify': if u.delete: u.delete["triples"], u.delete[ "quads"] = translateQuads(u.delete.quads) if u.insert: u.insert["triples"], u.insert[ "quads"] = translateQuads(u.insert.quads) u["where"] = translateGroupGraphPattern(u.where) else: raise Exception('Unknown type of update operation: %s' % u) u.prologue = prologue return u
[docs]def translateUpdate(q, base=None, initNs=None): """ Returns a list of SPARQL Update Algebra expressions """ res = [] prologue = None if not q.request: return res for p, u in zip(q.prologue, q.request): prologue = translatePrologue(p, base, initNs, prologue) # absolutize/resolve prefixes u = traverse( u, visitPost=functools.partial(translatePName, prologue=prologue)) u = _traverse(u, _simplifyFilters) u = traverse(u, visitPost=translatePath) res.append(translateUpdate1(u, prologue)) return res
[docs]def translateQuery(q, base=None, initNs=None): """ Translate a query-parsetree to a SPARQL Algebra Expression Return a rdflib.plugins.sparql.sparql.Query object """ # We get in: (prologue, query) prologue = translatePrologue(q[0], base, initNs) # absolutize/resolve prefixes q[1] = traverse( q[1], visitPost=functools.partial(translatePName, prologue=prologue)) P, PV = translate(q[1]) datasetClause = q[1].datasetClause if q[1].name == 'ConstructQuery': template = triples(q[1].template) if q[1].template else None res = CompValue(q[1].name, p=P, template=template, datasetClause=datasetClause) else: res = CompValue(q[1].name, p=P, datasetClause=datasetClause, PV=PV) res = traverse(res, visitPost=simplify) _traverseAgg(res, visitor=analyse) _traverseAgg(res, _addVars) return Query(prologue, res)
[docs]def pprintAlgebra(q): def pp(p, ind=" "): # if isinstance(p, list): # print "[ " # for x in p: pp(x,ind) # print "%s ]"%ind # return if not isinstance(p, CompValue): print p return print "%s(" % (, ) for k in p: print "%s%s =" % (ind, k,), pp(p[k], ind + " ") print "%s)" % ind try: pp(q.algebra) except AttributeError: # it's update, just a list for x in q: pp(x)
if __name__ == '__main__': import sys from rdflib.plugins.sparql import parser import os.path if os.path.exists(sys.argv[1]): q = file(sys.argv[1]) else: q = sys.argv[1] pq = parser.parseQuery(q) print pq tq = translateQuery(pq) print pprintAlgebra(tq)