[SQL] Improve key analysis for equijoins (#2232)

mihaibudiu · web-flow · commit 6e5a4b3d955c · 2024-08-12T23:58:46.000-07:00
* [SQL] Improve key analysis for equijoins
* [SQL] Change the implementation of NOW

Signed-off-by: Mihai Budiu &lt;mbudiu@feldera.com&gt;
diff --git a/benchmark/feldera-sql/benchmarks/nexmark/queries/q12.sql b/benchmark/feldera-sql/benchmarks/nexmark/queries/q12.sql
@@ -2,7 +2,7 @@ CREATE VIEW Q12 AS
 SELECT
     B.bidder,
     count(*) as bid_count,
-    TUMBLE_START(B.p_time, INTERVAL '10' SECOND) as starttime,
-    TUMBLE_END(B.p_time, INTERVAL '10' SECOND) as endtime
-FROM (SELECT *, NOW() as p_time FROM bid) B
-GROUP BY B.bidder, TUMBLE(B.p_time, INTERVAL '10' SECOND);
+    TUMBLE_START(B.date_time, INTERVAL '10' SECOND) as starttime,
+    TUMBLE_END(B.date_time, INTERVAL '10' SECOND) as endtime
+FROM bid B
+GROUP BY B.bidder, TUMBLE(B.date_time, INTERVAL '10' SECOND);
diff --git a/docs/sql/datetime.md b/docs/sql/datetime.md
@@ -310,17 +310,18 @@ CREATE TABLE NOW(now TIMESTAMP NOT NULL LATENESS INTERVAL 0 SECONDS);
 ```
 
 All invocations of the `NOW()` function within the program
-will produce the value that currently exists in this table.
-
-This table does not currently get populated automatically.
-Instead, the user is responsible for supplying the data to this table.
-The user has to maintain the invariant that this table always contains a
-single value.  Moreover, deleting a value and inserting a new one
-requires the newly inserted value to be larger than the original
-value.  The user can periodically update the contents of the table with
-the current physical timestamp.  Alternatively, they can use this table to
-evaluate queries over historical data by writing a series of increasing
-past timestamps to it.
+will produce the last value inserted in this table.
+
+This table does not currently get populated automatically.  Instead,
+the user is responsible for supplying the data to this table.  In
+every step of the circuit the user has to insert a new value in this
+table, which should be larger than the previous value.
+
+:::warning
+
+Programs that use `NOW()` can be very inefficient.  For example, a
+program such as `SELECT * FROM T WHERE T.x > NOW()` has to scan the
+entire table T at every step.  Use this function judiciously.
 
 | Operation     | Description         | Example                        |
 |---------------|---------------------|--------------------------------|
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToDotVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToDotVisitor.java
@@ -204,12 +204,14 @@ String getColor(DBSPOperator operator) {
             case "waterline" -> " style=filled fillcolor=lightgreen";
             case "controlled_filter" -> " style=filled fillcolor=cyan";
             case "apply", "apply2" -> " style=filled fillcolor=yellow";
-            case "integrate_trace_retain_keys", "integrate_trace_retain_values" -> " style=filled fillcolor=pink";
+            case "integrate_trace_retain_keys",
+                 "partitioned_rolling_aggregate_with_waterline",
+                 "integrate_trace_retain_values" -> " style=filled fillcolor=pink";
             // stateful operators
             case "distinct",
                  // all aggregates require an upsert, which is stateful, even the ones that are linear
                  "aggregate", "partitioned_rolling_aggregate", "aggregate_linear",
-                 "stream_aggregate", "stream_aggregate_linear", "partitioned_rolling_aggregate_with_waterline",
+                 "stream_aggregate", "stream_aggregate_linear",
                  "partitioned_tree_aggregate",
                  // some joins require integrators
                  "join", "join_flatmap",
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/JoinConditionAnalyzer.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/JoinConditionAnalyzer.java
@@ -24,6 +24,7 @@
 package org.dbsp.sqlCompiler.compiler.frontend;
 
 import org.apache.calcite.rex.*;
+import org.apache.calcite.sql.SqlKind;
 import org.dbsp.sqlCompiler.compiler.errors.InternalCompilerError;
 import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject;
 import org.dbsp.sqlCompiler.ir.type.DBSPType;
@@ -35,15 +36,12 @@
 import java.util.List;
 import java.util.Objects;
 
-public class JoinConditionAnalyzer extends RexVisitorImpl<Void> implements IWritesLogs {
+public class JoinConditionAnalyzer implements IWritesLogs {
     private final int leftTableColumnCount;
-    private final ConditionDecomposition result;
     private final TypeCompiler typeCompiler;
 
     public JoinConditionAnalyzer(CalciteObject object, int leftTableColumnCount, TypeCompiler typeCompiler) {
-        super(true);
         this.leftTableColumnCount = leftTableColumnCount;
-        this.result = new ConditionDecomposition(object);
         this.typeCompiler = typeCompiler;
     }
 
@@ -99,19 +97,80 @@ void validate() {
                 throw new InternalCompilerError("Unexpected empty join condition", this.object);
         }
 
-        /**
-         * Part of the join condition that is not an equality test.
-         * @return Null if the entire condition is an equality test.
-         */
+        /** Part of the join condition that is not an equality test.
+         * @return Null if the entire condition is an equality test. */
         @Nullable
         public RexNode getLeftOver() {
             this.validate();
             return this.leftOver;
         }
-    }
 
-    public boolean completed() {
-        return this.result.leftOver != null;
+        void analyzeAnd(RexCall call) {
+            List<RexNode> operands = call.getOperands();
+            List<RexNode> unprocessed = new ArrayList<>();
+            for (int i = 0; i < operands.size(); i++) {
+                RexNode operand = call.operands.get(i);
+                if (!(operand instanceof RexCall opCall)) {
+                    unprocessed.add(operand);
+                    continue;
+                }
+
+                if (opCall.op.kind != SqlKind.EQUALS &&
+                        opCall.op.kind != SqlKind.IS_NOT_DISTINCT_FROM) {
+                    unprocessed.add(operand);
+                    continue;
+                }
+
+                boolean eq = this.analyzeEquals(opCall);
+                if (!eq) {
+                    unprocessed.add(opCall);
+                }
+            }
+
+            if (!unprocessed.isEmpty()) {
+                if (unprocessed.size() == 1) {
+                    this.setLeftOver(unprocessed.get(0));
+                } else {
+                    call = call.clone(call.type, unprocessed);
+                    this.setLeftOver(call);
+                }
+            }
+        }
+
+        /** Analyze an equality comparison.  Return 'true' if this is suitable for an equijoin */
+        public boolean analyzeEquals(RexCall call) {
+            assert call.operands.size() == 2: "Expected 2 operands for equality checking";
+            RexNode left = call.operands.get(0);
+            RexNode right = call.operands.get(1);
+            @Nullable
+            Boolean leftIsLeft = JoinConditionAnalyzer.this.isLeftTableColumnReference(left);
+            @Nullable
+            Boolean rightIsLeft = JoinConditionAnalyzer.this.isLeftTableColumnReference(right);
+            if (leftIsLeft == null || rightIsLeft == null) {
+                return false;
+            }
+            if (leftIsLeft == rightIsLeft) {
+                // Both columns refer to the same table.
+                return false;
+            }
+            DBSPType leftType = JoinConditionAnalyzer.this.typeCompiler.convertType(
+                    left.getType(), true);
+            DBSPType rightType = JoinConditionAnalyzer.this.typeCompiler.convertType(
+                    right.getType(), true);
+            if (call.op.kind == SqlKind.IS_NOT_DISTINCT_FROM) {
+                // Only used if any of the operands is not nullable
+                if (leftType.mayBeNull && rightType.mayBeNull) {
+                    return false;
+                }
+            }
+            DBSPType commonType = ExpressionCompiler.reduceType(leftType, rightType).setMayBeNull(false);
+            if (leftIsLeft) {
+                this.addEquality(left, right, commonType);
+            } else {
+                this.addEquality(right, left, commonType);
+            }
+            return true;
+        }
     }
 
     @Nullable
@@ -134,79 +193,26 @@ public Boolean isLeftTableColumnReference(RexNode node) {
         return ref.getIndex() < this.leftTableColumnCount;
     }
 
-    @Override
-    public Void visitInputRef(RexInputRef ref) {
-        this.result.setLeftOver(ref);
-        return null;
-    }
-
-    @Override
-    public Void visitLiteral(RexLiteral lit) {
-        this.result.setLeftOver(lit);
-        return null;
-    }
-
-    @Override
-    public Void visitCall(RexCall call) {
-        switch (call.op.kind) {
-            case AND:
-                List<RexNode> operands = call.getOperands();
-                for (int i = 0; i < operands.size(); i++) {
-                    call.operands.get(i).accept(this);
-                    if (this.completed()) {
-                        if (i == operands.size() - 1) {
-                            // Just one left
-                            this.result.setLeftOver(operands.get(i));
-                            return null;
-                        }
-                        List<RexNode> remaining = new ArrayList<>();
-                        for (int j = i; j < operands.size(); j++)
-                            remaining.add(call.operands.get(j));
-                        call = call.clone(call.type, remaining);
-                        this.result.setLeftOver(call);
-                        return null;
-                    }
-                }
-                return null;
-            case EQUALS:
-                assert call.operands.size() == 2: "Expected 2 operands for equality checking";
-                RexNode left = call.operands.get(0);
-                RexNode right = call.operands.get(1);
-                @Nullable
-                Boolean leftIsLeft = this.isLeftTableColumnReference(left);
-                @Nullable
-                Boolean rightIsLeft = this.isLeftTableColumnReference(right);
-                if (leftIsLeft == null || rightIsLeft == null) {
-                    this.result.setLeftOver(call);
-                    return null;
-                }
-                if (leftIsLeft == rightIsLeft) {
-                    // Both columns refer to the same table.
-                    this.result.setLeftOver(call);
-                    return null;
-                }
-                DBSPType leftType = this.typeCompiler.convertType(left.getType(), true);
-                DBSPType rightType = this.typeCompiler.convertType(right.getType(), true);
-                DBSPType commonType = ExpressionCompiler.reduceType(leftType, rightType).setMayBeNull(false);
-                if (leftIsLeft) {
-                    this.result.addEquality(left, right, commonType);
-                } else {
-                    this.result.addEquality(right, left, commonType);
-                }
-                return null;
-            default:
-                // We are done: we don't know how to handle this condition.
-                this.result.setLeftOver(call);
-                return null;
-        }
-    }
-
     JoinConditionAnalyzer.ConditionDecomposition analyze(RexNode expression) {
         Logger.INSTANCE.belowLevel(this, 1)
                 .append("Analyzing ")
                 .append(expression.toString())
                 .newline();
-        expression.accept(this);
-        return this.result;
+        final ConditionDecomposition result = new ConditionDecomposition(CalciteObject.create(expression));
+        if (! (expression instanceof RexCall call)) {
+            result.setLeftOver(expression);
+            return result;
+        }
+        if (call.op.kind == SqlKind.AND) {
+            result.analyzeAnd(call);
+        } else if (call.op.kind == SqlKind.EQUALS || call.op.kind == SqlKind.IS_NOT_DISTINCT_FROM) {
+            boolean success = result.analyzeEquals(call);
+            if (!success) {
+                result.setLeftOver(call);
+            }
+        } else {
+            result.setLeftOver(call);
+        }
+        return result;
     }
 }
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/monotone/MonotoneTransferFunctions.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/monotone/MonotoneTransferFunctions.java
@@ -590,8 +590,10 @@ public void postorder(DBSPApplyExpression expression) {
                         name.startsWith("sign_") ||
                         name.startsWith("numeric_inc") ||
                         name.startsWith("extract_year_") ||
+                        name.startsWith("extract_millennium_") ||
+                        name.startsWith("extract_century_") ||
                         name.startsWith("extract_epoch_") ||
-                        name.startsWith("extract_hour_Time") ||
+                        name.startsWith("extract_hour_Time_") ||
                         name.startsWith("dateadd_") ||
                         name.equals("hop_start_timestamp") ||
                         name.startsWith("to_bound_") ||
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/ImplementNow.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/ImplementNow.java
@@ -1,6 +1,7 @@
 package org.dbsp.sqlCompiler.compiler.visitors.outer;
 
 import org.dbsp.sqlCompiler.circuit.DBSPCircuit;
+import org.dbsp.sqlCompiler.circuit.operator.DBSPDifferentiateOperator;
 import org.dbsp.sqlCompiler.circuit.operator.DBSPFilterOperator;
 import org.dbsp.sqlCompiler.circuit.operator.DBSPMapIndexOperator;
 import org.dbsp.sqlCompiler.circuit.operator.DBSPMapOperator;
@@ -22,6 +23,7 @@
 import org.dbsp.sqlCompiler.ir.IDBSPDeclaration;
 import org.dbsp.sqlCompiler.ir.IDBSPOuterNode;
 import org.dbsp.sqlCompiler.ir.annotation.AlwaysMonotone;
+import org.dbsp.sqlCompiler.ir.annotation.NoInc;
 import org.dbsp.sqlCompiler.ir.expression.DBSPApplyExpression;
 import org.dbsp.sqlCompiler.ir.expression.DBSPClosureExpression;
 import org.dbsp.sqlCompiler.ir.expression.DBSPExpression;
@@ -187,7 +189,7 @@ public RewriteNow(IErrorReporter reporter, ICompilerComponent compiler) {
             this.compiler = compiler;
         }
 
-        DBSPStreamJoinOperator createJoin(DBSPUnaryOperator operator) {
+        DBSPOperator createJoin(DBSPUnaryOperator operator) {
             // Index the input
             DBSPType inputType = operator.input().getOutputZSetElementType();
             DBSPVariablePath var = inputType.ref().var();
@@ -210,8 +212,12 @@ DBSPStreamJoinOperator createJoin(DBSPUnaryOperator operator) {
                     Linq.map(fields, DBSPExpression::getType)));
             DBSPExpression joinFunction = new DBSPTupleExpression(fields, false)
                     .closure(key.asParameter(), left.asParameter(), right.asParameter());
+            assert nowIndexed != null;
+            DBSPDifferentiateOperator dNow = new DBSPDifferentiateOperator(operator.getNode(), nowIndexed);
+            dNow.annotations.add(new NoInc());
+            this.addOperator(dNow);
             DBSPStreamJoinOperator join = new DBSPStreamJoinOperator(operator.getNode(), joinType,
-                    joinFunction, operator.isMultiset, index, Objects.requireNonNull(this.nowIndexed));
+                    joinFunction, operator.isMultiset, index, dNow);
             this.addOperator(join);
             return join;
         }
@@ -222,7 +228,7 @@ public void postorder(DBSPMapOperator operator) {
             DBSPExpression function = operator.getFunction();
             cn.apply(function);
             if (cn.found()) {
-                DBSPStreamJoinOperator join = this.createJoin(operator);
+                DBSPOperator join = this.createJoin(operator);
                 RewriteNowExpression rn = new RewriteNowExpression(this.errorReporter);
                 function = rn.apply(function).to(DBSPExpression.class);
                 DBSPOperator result = new DBSPMapOperator(operator.getNode(), function, operator.getOutputZSetType(), join);
@@ -238,7 +244,7 @@ public void postorder(DBSPFilterOperator operator) {
             DBSPExpression function = operator.getFunction();
             cn.apply(function);
             if (cn.found()) {
-                DBSPStreamJoinOperator join = this.createJoin(operator);
+                DBSPOperator join = this.createJoin(operator);
                 RewriteNowExpression rn = new RewriteNowExpression(this.errorReporter);
                 function = rn.apply(function).to(DBSPExpression.class);
                 DBSPOperator filter = new DBSPFilterOperator(operator.getNode(), function, join);
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/OptimizeIncrementalVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/OptimizeIncrementalVisitor.java
@@ -25,6 +25,7 @@
 
 import org.dbsp.sqlCompiler.circuit.operator.*;
 import org.dbsp.sqlCompiler.compiler.IErrorReporter;
+import org.dbsp.sqlCompiler.ir.annotation.NoInc;
 import org.dbsp.util.Linq;
 
 import java.util.List;
@@ -38,6 +39,11 @@ public OptimizeIncrementalVisitor(IErrorReporter reporter) {
 
     @Override
     public void postorder(DBSPDifferentiateOperator operator) {
+        if (operator.hasAnnotation(p -> p.is(NoInc.class))) {
+            this.linear(operator);
+            return;
+        }
+
         DBSPOperator source = this.mapped(operator.input());
         if (source.is(DBSPIntegrateOperator.class)) {
             DBSPIntegrateOperator integral = source.to(DBSPIntegrateOperator.class);
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/annotation/AlwaysMonotone.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/annotation/AlwaysMonotone.java
@@ -2,9 +2,4 @@
 
 /** This annotation indicates that a DBSPOperator has an output that is always monotone.
  * This is intended to mark the part of the circuit that derives from the NOW table. */
-public class AlwaysMonotone extends Annotation {
-    @Override
-    public String toString() {
-        return "AlwaysMonotone";
-    }
-}
+public class AlwaysMonotone extends Annotation { }
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/annotation/Annotation.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/annotation/Annotation.java
@@ -3,4 +3,9 @@
 import org.dbsp.util.ICastable;
 
 /** Base class for annotations that can be attached to various IR nodes. */
-public abstract class Annotation implements ICastable { }
+public abstract class Annotation implements ICastable {
+    @Override
+    public String toString() {
+        return this.getClass().getSimpleName();
+    }
+}
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/annotation/NoInc.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/annotation/NoInc.java
@@ -0,0 +1,5 @@
+package org.dbsp.sqlCompiler.ir.annotation;
+
+/** Annotation set on a differentiation operator indicating
+ * that it should be preserved by the incremental optimization step. */
+public class NoInc extends Annotation { }
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/streaming/StreamingTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/streaming/StreamingTests.java
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/nexmark/NexmarkTest.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/nexmark/NexmarkTest.java
