Merge pull request #369 from EvgSkv/ti2023

Make recursion 32 deep and iterative when DuckDB is the engine.

compiler/dialect_libraries/recursion_library.py

@@ -135,7 +135,8 @@ def GetFlatIterativeRecursionFunctor(depth, cover, direct_args_of,
   result_rules = []
   iterate_over_upper_half = []
   iterate_over_lower_half = []
-  inset = ignition_steps // 2
+  # inset = ignition_steps // 2
+  inset = 2
   stop_file_name = ''
   if stop:
     stop_file_name = '/tmp/logical_stop_%s_%s.json' % (

compiler/functors.py

@@ -108,6 +108,7 @@ def __init__(self, rules):
     self.cached_calls = {}
 
     try:
+      # This import takes about 500ms.
       import numpy
       numpy_is_here = True
     except:
@@ -530,16 +531,23 @@ def GetStop(self, depth_map, p):
       stop = stop['predicate_name']
     return stop
 
-  def UnfoldRecursions(self, depth_map):
+  def UnfoldRecursions(self, depth_map, default_iterative, default_depth):
     """Unfolds all recursions."""
-    should_recurse, my_cover = self.RecursiveAnalysis(depth_map)
+    should_recurse, my_cover = self.RecursiveAnalysis(
+      depth_map, default_iterative, default_depth)
     new_rules = copy.deepcopy(self.rules)
     for p, style in should_recurse.items():
-      depth = depth_map.get(p, {}).get('1', 8)
+      depth = depth_map.get(p, {}).get('1', default_depth)
       if style == 'vertical':
         self.UnfoldRecursivePredicate(p, my_cover[p], depth, new_rules)
       elif style == 'horizontal' or style == 'iterative_horizontal':
-        ignition = len(my_cover[p]) * 3 + 4
+        # Old ad-hoc formula:
+        # ignition = len(my_cover[p]) * 3 + 4
+        # Calculation of ignition:
+        #   len(cover) to fill all predicates,
+        #   + 2 for iterations
+        #   + 1 for final step.
+        ignition = len(my_cover[p]) + 3
         if ignition % 2 == depth % 2:
           ignition += 1
         stop = self.GetStop(depth_map, p)
@@ -631,7 +639,7 @@ def IsCutOfCover(self, p, cover_leaf):
           stack.append((x, (u | {t})))
     return True
 
-  def RecursiveAnalysis(self, depth_map):
+  def RecursiveAnalysis(self, depth_map, default_iterative, default_depth):
     """Finds recursive cycles and predicates that would unfold them."""
     # TODO: Select unfolding predicates to guarantee unfolding.
     cover = []
@@ -661,9 +669,9 @@ def RecursiveAnalysis(self, depth_map):
         p = min(c)
       # Iterate if explicitly requested or unspecified
       # and number of steps is greater than 20.
-      if (depth_map.get(p, {}).get('iterative') or
+      if (depth_map.get(p, {}).get('iterative', default_iterative) or
           depth_map.get(p, {}).get('iterative', True) == True and
-          depth_map.get(p, {}).get('1', 8) > 20):
+          depth_map.get(p, {}).get('1', default_depth) > 20):
         should_recurse[p] = 'iterative_horizontal'
       elif self.IsCutOfCover(p, c):
         should_recurse[p] = 'vertical'

compiler/universe.py

@@ -645,7 +645,15 @@ def UnfoldRecursion(self, rules):
     #   for p in depth_map:
     #     # DuckDB struggles with long querries.
     #     depth_map[p]['iterative'] = True
-    return f.UnfoldRecursions(depth_map)
+    quacks_like_a_duck = (
+      annotations.annotations.get(
+        '@Engine', {}).get('duckdb', None) is not None)
+    default_iterative = False
+    default_depth = 8
+    if quacks_like_a_duck:
+      default_iterative = True
+      default_depth = 32
+    return f.UnfoldRecursions(depth_map, default_iterative, default_depth)
 
   def BuildUdfs(self):
     """Build UDF definitions."""
@@ -976,6 +984,23 @@ def UpdateExecutionWithTyping(self):
     if self.execution.dialect.IsPostgreSQLish():
       self.execution.preamble += '\n' + self.typing_preamble
 
+  def PerformIterationClosure(self, allocator):
+    """Iteration closure of current execution.
+    
+    If one predicates of the iteration runs, then all of them must run.
+    """
+    participating_predicates = list(
+      self.execution.table_to_defined_table_map.keys())
+    translator = self.MakeSubqueryTranslator(allocator)
+    for iteration in self.execution.iterations.values():
+      for p in participating_predicates:
+        iteration_predicates = set(iteration['predicates'])
+        if p in iteration_predicates:
+          for d in iteration_predicates:
+            # We are not doing anything with the resulting SQL,
+            # as we only need the execution state updated.
+            translator.TranslateTable(d, None)
+
   def FormattedPredicateSql(self, name, allocator=None):
     """Printing top-level formatted SQL statement with defines and exports."""
     self.InitializeExecution(name)
@@ -990,6 +1015,7 @@ def FormattedPredicateSql(self, name, allocator=None):
       sql = self.FunctionSql(name, allocator)
     else:
       sql = self.PredicateSql(name, allocator)
+    self.PerformIterationClosure(allocator)
 
     self.UpdateExecutionWithTyping()
 

integration_tests/duckdb_iteration_closure_test.l

@@ -0,0 +1,26 @@
+#
+# Copyright 2020 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# This program is executed incorrectly unless iteration closure is
+# performed.
+
+@Engine("duckdb");
+
+A() = 0;
+B() = A() + 1;
+C() = B() + 1;
+A() = C() + 1;
+
+Test() Max= A();

integration_tests/duckdb_iteration_closure_test.txt

@@ -0,0 +1,5 @@
++--------------+
+| logica_value |
++--------------+
+| 30           |
++--------------+

integration_tests/psql_flow_test.l

@@ -40,7 +40,7 @@ V() = x distinct :- x in [i, j], G(i, j);
 @Ground(Opportunity);
 @Ground(ActivePath);
 @OrderBy(Test, "col0", "col1");
-@Recursive(Flow, 20);
+@Recursive(Flow, 20, iterative: false);
 
 Flow(i, j) += 0 :- G(i, j);
 

integration_tests/run_tests.py

@@ -109,6 +109,8 @@ def RunAll(test_presto=False, test_trino=False):
   RunTest("sqlite_reachability")
   RunTest("sqlite_element_test")
 
+  RunTest("duckdb_iteration_closure_test",
+          use_concertina=True)
   RunTest("duckdb_stop_test",
           src="duckdb_stop_test.l",
           use_concertina=True)
@@ -138,7 +140,8 @@ def RunAll(test_presto=False, test_trino=False):
 
   RunTest("psql_udf_test")
   RunTest("psql_flow_test")
-  RunTest("psql_graph_coloring_test")
+  RunTest("psql_graph_coloring_test",
+          use_concertina=True)
   RunTest("psql_win_move_test")
   RunTest("psql_plusplus_test")
   RunTest("psql_argmax2_combine_test")