ReduceScan and ReduceTraversal respect variable scopes (#1070)

* ReduceScan and ReduceTraversal respect variable scopes

* Fix source lookup

* PR fixes

Co-authored-by: Roi Lipman <swilly22@users.noreply.github.com>

src/execution_plan/execution_plan_modify.c

@@ -264,6 +264,12 @@ void ExecutionPlan_BoundVariables(const OpBase *op, rax *modifiers) {
 		}
 	}
 
+	/* Project and Aggregate operations demarcate variable scopes,
+	 * collect their projections but do not recurse into their children.
+	 * Note that future optimizations which operate across scopes will require different logic
+	 * than this for application. */
+	if(op->type == OPType_PROJECT || op->type == OPType_AGGREGATE) return;
+
 	for(int i = 0; i < op->childCount; i++) {
 		ExecutionPlan_BoundVariables(op->children[i], modifiers);
 	}

src/execution_plan/optimizations/reduce_scans.c

@@ -29,25 +29,25 @@ static void _reduceScans(ExecutionPlan *plan, OpBase *scan) {
 	assert(array_len(scan->modifies) == 1);
 	const char *scanned_alias = scan->modifies[0];
 
-	// Check if that alias is already bound by any of the scan's children.
+	// Collect variables bound before this operation.
+	rax *bound_vars = raxNew();
 	for(int i = 0; i < scan->childCount; i ++) {
-		if(ExecutionPlan_LocateOpResolvingAlias(scan->children[i], scanned_alias)) {
-			/* Here is the case where a node is already populated in the record when it arrived to a label scan operation.
-			 * The label scan operation here is redundent since it will re-scan the entire graph, so we will replace it with
-			 * a conditional traverse operation which will filter only the nodes with the requested labels.
-			 * The conditional traverse is done over the label matrix, and is more efficient then a filter operation. */
-			if(scan->type == OPType_NODE_BY_LABEL_SCAN) {
-				OpBase *traverse = _LabelScanToConditionalTraverse((NodeByLabelScan *)scan);
-				ExecutionPlan_ReplaceOp(plan, scan, traverse);
-				OpBase_Free(scan);
-				break;
-			}
-			// The scanned alias is already bound, remove the redundant scan op.
+		ExecutionPlan_BoundVariables(scan->children[i], bound_vars);
+	}
+
+	if(raxFind(bound_vars, (unsigned char *)scanned_alias, strlen(scanned_alias)) != raxNotFound) {
+		// If the alias the scan operates on is already bound, the scan operation is redundant.
+		if(scan->type == OPType_NODE_BY_LABEL_SCAN) {
+			// If we are performing a label scan, introduce a conditional traversal to filter by label.
+			OpBase *traverse = _LabelScanToConditionalTraverse((NodeByLabelScan *)scan);
+			ExecutionPlan_ReplaceOp(plan, scan, traverse);
+		} else {
+			// Remove the redundant scan op.
 			ExecutionPlan_RemoveOp(plan, scan);
-			OpBase_Free(scan);
-			break;
 		}
+		OpBase_Free(scan);
 	}
+	raxFree(bound_vars);
 }
 
 void reduceScans(ExecutionPlan *plan) {

src/execution_plan/optimizations/reduce_traversal.c

@@ -43,7 +43,6 @@ void reduceTraversal(ExecutionPlan *plan) {
 	for(uint i = 0; i < traversals_count; i++) {
 		OpBase *op = traversals[i];
 		AlgebraicExpression *ae;
-
 		if(op->type == OPType_CONDITIONAL_TRAVERSE) {
 			CondTraverse *traverse = (CondTraverse *)op;
 			ae = traverse->ae;
@@ -65,9 +64,18 @@ void reduceTraversal(ExecutionPlan *plan) {
 		   AlgebraicExpression_OperandCount(ae) == 1 &&
 		   AlgebraicExpression_DiagonalOperand(ae, 0)) continue;
 
-		/* Search to see if dest is already resolved */
-		if(!ExecutionPlan_LocateOpResolvingAlias(op->children[0],
-												 AlgebraicExpression_Destination(ae))) continue;
+		// Collect variables bound before this op.
+		rax *bound_vars = raxNew();
+		for(int i = 0; i < op->childCount; i ++) {
+			ExecutionPlan_BoundVariables(op->children[i], bound_vars);
+		}
+
+		const char *dest = AlgebraicExpression_Destination(ae);
+		if(raxFind(bound_vars, (unsigned char *)dest, strlen(dest)) == raxNotFound) {
+			// The destination could not be resolved, cannot optimize.
+			raxFree(bound_vars);
+			continue;
+		}
 
 		/* Both src and dest are already known
 		 * perform expand into instaed of traverse. */
@@ -108,6 +116,7 @@ void reduceTraversal(ExecutionPlan *plan) {
 				}
 			}
 		}
+		raxFree(bound_vars);
 	}
 
 	// Remove redundant traversals

tests/flow/test_null_handling.py

@@ -86,3 +86,31 @@ def test06_null_named_path_function_inputs(self):
         # The path and function calls on it should return NULL, while collect() returns an empty array.
         expected_result = [[None, None, []]]
         self.env.assertEquals(actual_result.result_set, expected_result)
+
+    # Scan and traversal operations should gracefully handle NULL inputs.
+    def test07_null_graph_entity_inputs(self):
+        query = """WITH NULL AS a MATCH (a) RETURN a"""
+        actual_result = redis_graph.query(query)
+        # Expect one NULL entity to be returned.
+        expected_result = [[None]]
+        self.env.assertEquals(actual_result.result_set, expected_result)
+
+        query = """WITH NULL AS a MATCH (a)-[e]->(b) RETURN a, e, b"""
+        plan = redis_graph.execution_plan(query)
+        # Verify that we are attempting to perform a traversal but no scan.
+        self.env.assertNotIn("Scan", plan)
+        self.env.assertIn("Conditional Traverse", plan)
+        actual_result = redis_graph.query(query)
+        # Expect no results.
+        expected_result = []
+        self.env.assertEquals(actual_result.result_set, expected_result)
+
+        query = """WITH NULL AS e MATCH (a:L)-[e]->(b) RETURN a, e, b"""
+        plan = redis_graph.execution_plan(query)
+        # Verify that we are performing a scan and traversal.
+        self.env.assertIn("Label Scan", plan)
+        self.env.assertIn("Conditional Traverse", plan)
+        actual_result = redis_graph.query(query)
+        # Expect no results.
+        expected_result = []
+        self.env.assertEquals(actual_result.result_set, expected_result)

tests/flow/test_optimizations_plan.py

@@ -325,10 +325,32 @@ def test19_test_filter_compaction_remove_true_filter(self):
         executionPlan = graph.execution_plan(query)
         self.env.assertNotIn("Filter", executionPlan)
 
-    def test19_test_filter_compaction_not_removing_false_filter(self):
+    def test20_test_filter_compaction_not_removing_false_filter(self):
         query = "MATCH (n) WHERE 1 > 1 RETURN n"
         executionPlan = graph.execution_plan(query)
         self.env.assertIn("Filter", executionPlan)
         resultset = graph.query(query).result_set
         expected = []
         self.env.assertEqual(resultset, expected)
+
+    # ExpandInto should be applied where possible on projected graph entities.
+    def test21_expand_into_projected_endpoints(self):
+        query = """MATCH (a)-[]->(b) WITH a, b MATCH (a)-[e]->(b) RETURN a.val, b.val ORDER BY a.val, b.val LIMIT 3"""
+        executionPlan = graph.execution_plan(query)
+        self.env.assertIn("Expand Into", executionPlan)
+        resultset = graph.query(query).result_set
+        expected = [[0, 1],
+                    [0, 2],
+                    [0, 3]]
+        self.env.assertEqual(resultset, expected)
+
+    # Variables bound in one scope should not be used to introduce ExpandInto ops in later scopes.
+    def test22_no_expand_into_across_scopes(self):
+        query = """MATCH (reused_1)-[]->(reused_2) WITH COUNT(reused_2) as edge_count MATCH (reused_1)-[]->(reused_2) RETURN edge_count, reused_1.val, reused_2.val ORDER BY reused_1.val, reused_2.val LIMIT 3"""
+        executionPlan = graph.execution_plan(query)
+        self.env.assertNotIn("Expand Into", executionPlan)
+        resultset = graph.query(query).result_set
+        expected = [[14, 0, 1],
+                    [14, 0, 2],
+                    [14, 0, 3]]
+        self.env.assertEqual(resultset, expected)

tests/tck/features/WithAcceptance.feature

@@ -265,7 +265,6 @@ Feature: WithAcceptance
             | 'B'  |
         And no side effects
 
-    @skip
     Scenario: A simple pattern with one bound endpoint
         Given an empty graph
         And having executed: