fix issue with VALUES clause containing multiple variables

Author: prohde

DeTrusty/Decomposer/Decomposer.py

@@ -4,11 +4,13 @@
 from enum import Enum
 from functools import partial
 
-from DeTrusty.Logger import get_logger
 from DeTrusty.Decomposer import Tree, utils
+from DeTrusty.Logger import get_logger
 from DeTrusty.Sparql.Parser.services import Service, Triple, Filter, Optional, UnionBlock, JoinBlock, Values, Bind, Argument
 
 logger = get_logger(__name__, '.decompositions.log')
+
+
 class Decomposer(object):
 
     def __init__(self, query, config, decompType="STAR", joinstarslocally=True, sparql_one_dot_one=False):
@@ -105,7 +107,7 @@ def decomposeJoinBlock(self, jb):
                 return None
 
         fl1 = self.includeFilter(sl, fl)
-        fl = list(set(fl) - set(fl1)) # in case VALUES with variables from multiple sources, the VALUES object won't be removed
+        fl = list(set(fl) - set(fl1))  # in case VALUES with variables from multiple sources, the VALUES object won't be removed
         if sl:
             if len(sl) == 1 and isinstance(sl[0], UnionBlock) and fl != []:
                 sl[0] = self.updateFilters(sl[0], fl)
@@ -695,10 +697,6 @@ def includeFilterAux(self, f, sl):
             if set(vars_s) & set(vars_f) == set(vars_f):
                 s.include_filter(f)
                 fl1 = fl1 + [f]
-            elif type(f) is Values and set(vars_s) & set(vars_f) != set():
-                fl2 = f.instantiate(set(vars_s) & set(vars_f))
-                s.include_filter(fl2) # the new decomposed clause need to be reconsidered
-                fl1 = fl1 + [fl2]
         return fl1
 
     def includeFilterUnionBlock(self, jb, f):
@@ -708,11 +706,7 @@ def includeFilterUnionBlock(self, jb, f):
                 if isinstance(jbUS, Service):
                     vars_s = set(jbUS.getVars())
                     vars_f = f.getVars()
-                    if type(f) is Values and set(vars_s) & set(vars_f) != set():
-                        fl2 = f.instantiate(set(vars_s) & set(vars_f))
-                        jbUS.include_filter(fl2)
-                        fl1 = fl1 + [fl2]
-                    elif set(vars_s) & set(vars_f) == set(vars_f):
+                    if set(vars_s) & set(vars_f) == set(vars_f):
                         jbUS.include_filter(f)
                         fl1 = fl1 + [f]
         return fl1
@@ -741,33 +735,22 @@ def includeFilterAuxSK(self, f, sl, sr):
         for s in sl:
             bgpvars.update(set(utils.getVars(s)))
             vars_s = set()
-            if (isinstance(s, Triple)):
+            if isinstance(s, Triple):
                 vars_s.update(set(utils.getVars(s)))
             else:
                 for t in s.triples:
                     vars_s.update(set(utils.getVars(t)))
 
-            if set(vars_s) & set(vars_f) == set(vars_f): # Bind supposed to fall to this category
-                serviceFilter = True
-
-            if type(f) is Values and set(vars_s) & set(vars_f) != set():
-                fl2 = f.instantiate(set(vars_s) & set(vars_f))
+            if set(vars_s) & set(vars_f) == set(vars_f):  # Bind supposed to fall to this category
                 serviceFilter = True
 
         for v in bgpvars:
             if v in fvars:
                 fvars[v] = True
 
-        # if type(f) is Bind: # why is this here? Investigate later
-        #     fvars[f.alias] = True
-
         if serviceFilter:
-            if type(f) is Values:
-                sr.include_filter(fl2)
-                fl1 = fl1 + [fl2]
-            else:
-                sr.include_filter(f)
-                fl1 = fl1 + [f]
+            sr.include_filter(f)
+            fl1 = fl1 + [f]
         else:
             fs = [v for v in fvars if not fvars[v]]
             if len(fs) == 0:

DeTrusty/Decomposer/Planner.py

@@ -18,10 +18,11 @@
 from DeTrusty.Operators.AnapsidOperators.Xorderby import Xorderby
 from DeTrusty.Operators.AnapsidOperators.Xproject import Xproject
 from DeTrusty.Operators.AnapsidOperators.Xunion import Xunion
+from DeTrusty.Operators.AnapsidOperators.Xvalues import Xvalues
 from DeTrusty.Operators.BlockingOperators.Union import Union
 from DeTrusty.Operators.NonBlockingOperators.NestedHashJoinFilter import NestedHashJoinFilter as NestedHashJoin
 from DeTrusty.Operators.NonBlockingOperators.NestedHashOptionalFilter import  NestedHashOptionalFilter as NestedHashOptional
-from DeTrusty.Sparql.Parser.services import Bind, Filter, Service, Optional, UnionBlock, JoinBlock
+from DeTrusty.Sparql.Parser.services import Bind, Filter, Values, Service, Optional, UnionBlock, JoinBlock
 
 
 class Planner(object):
@@ -52,7 +53,7 @@ def createPlan(self):
             raise SyntaxError('Query contains projections that are neither grouping variables nor aggregates.')
 
         # Adds the group by operator to the plan.
-        if (len(query.group_by) > 0 or over_all_triples):
+        if (len(query.group_by) > 0) or over_all_triples:
             operatorTree = TreePlan(Xgroupby(query.group_by, over_all_triples), operatorTree.vars, operatorTree)
 
         # Adds the having operator to the plan.
@@ -188,6 +189,8 @@ def includePhysicalOperators(self, tree):
                     if set(n.vars) & set(vars_f) == set(vars_f):
                         if isinstance(f, Filter):
                             n = TreePlan(Xfilter(f), n.vars, n)
+                        elif isinstance(f, Values):
+                            n = TreePlan(Xvalues(f), n.vars, n)
                     if isinstance(f, Bind):
                         n.vars = set(n.vars) | set(vars_f)
                         if n.vars & set(vars_f) == set(vars_f):

DeTrusty/Operators/AnapsidOperators/Xvalues.py

@@ -1,5 +1,7 @@
-from multiprocessing import Queue
 import datetime
+from multiprocessing import Queue
+
+from DeTrusty.Sparql.Parser.services import Argument
 
 data_types = {
     'integer': (int, 'numerical'),
@@ -23,6 +25,8 @@
     'positiveInteger': (int, 'numerical')
 }
 
+UNDEF = Argument('UNDEF')
+
 
 class Xvalues(object):
 
@@ -31,49 +35,50 @@ class Xvalues(object):
     def __init__(self, values): 
         self.input = Queue()
         self.qresults = Queue()
-        self.values = values  
+        self.values = values
+        self.left = None
 
     def execute(self, left, dummy, out, processqueue=Queue()):
         self.left = left
         self.qresults = out
-        tuple = self.left.get(True)
+        tuple_ = self.left.get(True)
 
-        while (tuple != "EOF"):
-            shouldInclude = self.filterByValues(tuple)
-            if shouldInclude:
-                self.qresults.put(tuple)
+        while tuple_ != "EOF":
+            should_include = self.filterByValues(tuple_)
+            if should_include:
+                self.qresults.put(tuple_)
 
-            tuple = self.left.get(True)           
+            tuple_ = self.left.get(True)
 
         # Put EOF in queue and exit.
         self.qresults.put("EOF")
 
-    def filterByValues(self, tuple):        
+    def filterByValues(self, tuple_):
         for row in self.values.data_block_val:
-            isValid = True
+            is_valid = True
             for idx, variable in enumerate(self.values.var): 
-                value = tuple[variable.name[1:]]
-                rowArg = row[idx]
-                if rowArg is None:
+                value = tuple_[variable.name[1:]]['value']
+                row_arg = row[idx]
+                if row_arg == UNDEF:
                     continue 
-                extractedRowValue = self.extractValue(rowArg.name)
-                rowValue = extractedRowValue[1:-1]
+                extracted_row_value = self.extractValue(row_arg.name)
+                row_value = extracted_row_value[1:-1]
 
-                if value != rowValue:
-                    isValid = False
+                if value != row_value:
+                    is_valid = False
                     break
-            if isValid:
+            if is_valid:
                 return True
         return False 
 
     def extractValue(self, val):
         pos = val.find("^^")
         # Handles when the literal is typed.
-        if (pos > -1):
+        if pos > -1:
             for t in data_types.keys():
-                if (t in val[pos:]):
+                if t in val[pos:]:
                     (python_type, general_type) = data_types[t]
-                    if (general_type == bool):
+                    if general_type == bool:
                         return val[:pos]
                     else:
                         return python_type(val[:pos])