From bceb5585525d486d820ce44aaa96d8882921d40a Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Wed, 5 Aug 2026 17:52:51 -0700 Subject: [PATCH] [SQL] Optimization to pull expensive operations out of filters Signed-off-by: Mihai Budiu --- .../compiler/visitors/inner/Expensive.java | 25 ++ .../compiler/visitors/inner/ReferenceMap.java | 6 + .../visitors/outer/CircuitOptimizer.java | 5 +- .../outer/DecomposeExpensiveFilters.java | 236 ++++++++++++++++++ .../visitors/outer/PullFilterVisitor.java | 6 +- .../visitors/outer/temporal/ContainsNow.java | 2 +- .../ir/expression/DBSPClosureExpression.java | 14 +- .../outer/DecomposeExpensiveFiltersTests.java | 143 +++++++++++ 8 files changed, 432 insertions(+), 5 deletions(-) create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DecomposeExpensiveFilters.java create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DecomposeExpensiveFiltersTests.java diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/Expensive.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/Expensive.java index e997d71c9bc..25626cf22cc 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/Expensive.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/Expensive.java @@ -6,8 +6,12 @@ import org.dbsp.sqlCompiler.ir.IDBSPInnerNode; import org.dbsp.sqlCompiler.ir.expression.DBSPApplyExpression; import org.dbsp.sqlCompiler.ir.expression.DBSPApplyMethodExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPBinaryExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPCastExpression; import org.dbsp.sqlCompiler.ir.expression.DBSPExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPOpcode; import org.dbsp.sqlCompiler.ir.type.DBSPType; +import org.dbsp.sqlCompiler.ir.type.primitive.DBSPTypeVariant; /** Visitor which detects whether an expression contains "expensive" subexpressions. * Today any external function call is deemed expensive. @@ -66,6 +70,27 @@ public VisitDecision preorder(DBSPApplyMethodExpression unused) { return VisitDecision.STOP; } + @Override + public VisitDecision preorder(DBSPBinaryExpression expression) { + // Lowered to a runtime function call + if (expression.opcode == DBSPOpcode.VARIANT_INDEX) { + this.expensive = true; + return VisitDecision.STOP; + } + return super.preorder(expression); + } + + @Override + public VisitDecision preorder(DBSPCastExpression expression) { + // VARIANT casts are lowered to runtime function calls + if (expression.getType().is(DBSPTypeVariant.class) || + expression.source.getType().is(DBSPTypeVariant.class)) { + this.expensive = true; + return VisitDecision.STOP; + } + return super.preorder(expression); + } + public static boolean isExpensive(DBSPCompiler compiler, DBSPExpression expression) { Expensive expensive = new Expensive(compiler); expensive.apply(expression); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/ReferenceMap.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/ReferenceMap.java index 0a980d9db55..2458624f354 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/ReferenceMap.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/ReferenceMap.java @@ -3,6 +3,7 @@ import org.dbsp.sqlCompiler.compiler.errors.InternalCompilerError; import org.dbsp.sqlCompiler.ir.IDBSPDeclaration; import org.dbsp.sqlCompiler.ir.expression.DBSPVariablePath; +import org.dbsp.util.Linq; import org.dbsp.util.Utilities; import javax.annotation.Nullable; @@ -33,6 +34,11 @@ public IDBSPDeclaration getDeclaration(DBSPVariablePath var) { return Utilities.getExists(this.declarations, var); } + /** Number of references to a specific declaration */ + public int count(IDBSPDeclaration decl) { + return Linq.where(this.declarations.values(), v -> v == decl).size(); + } + @Nullable public IDBSPDeclaration get(DBSPVariablePath var) { return this.declarations.get(var); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java index 32948adb7cc..afc57bc7613 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java @@ -75,6 +75,9 @@ void createOptimizer() { // Example dumping circuit to a png file // this.dump(3); // First part of optimizations may still synthesize some circuit components + // Runs before ImplementNow so temporal filters compare fields instead of + // expensive computations, which makes them implementable as windows + this.add(new DecomposeExpensiveFilters(compiler)); this.add(new ImplementNow(compiler)); if (compiler.options.ioOptions.correlatedColumns) this.add(new Lineage(compiler)); @@ -101,7 +104,6 @@ void createOptimizer() { this.add(new ExpandAggregateZero(compiler)); this.add(new Conditional(compiler, new RemoveStarJoins(compiler), this.compiler.metadata::noStarJoins)); this.add(new DeadCode(compiler, true)); - this.add(new OptimizeWithGraph(compiler, g -> new PullFilterVisitor(compiler, g))); this.add(new PropagateEmptySources(compiler)); this.add(new OptimizeDistinctVisitor(compiler)); // This is useful even without incrementalization if we have recursion @@ -116,6 +118,7 @@ void createOptimizer() { this.add(new Simplify(compiler).circuitRewriter(true)); this.add(new RemoveConstantFilters(compiler)); this.add(new OptimizeWithGraph(compiler, g -> new OptimizeProjectionVisitor(compiler, g))); + this.add(new OptimizeWithGraph(compiler, g -> new PullFilterVisitor(compiler, g))); this.add(new OptimizeWithGraph(compiler, g -> new OptimizeProjections(compiler, true, g, operatorsAnalyzed))); this.add(new ShareIndexes(compiler)); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DecomposeExpensiveFilters.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DecomposeExpensiveFilters.java new file mode 100644 index 00000000000..0d90e034716 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DecomposeExpensiveFilters.java @@ -0,0 +1,236 @@ +package org.dbsp.sqlCompiler.compiler.visitors.outer; + +import org.dbsp.sqlCompiler.circuit.annotation.IsProjection; +import org.dbsp.sqlCompiler.circuit.operator.DBSPFilterOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPMapOperator; +import org.dbsp.sqlCompiler.compiler.DBSPCompiler; +import org.dbsp.sqlCompiler.compiler.visitors.inner.EquivalenceContext; +import org.dbsp.sqlCompiler.compiler.visitors.inner.Expensive; +import org.dbsp.sqlCompiler.compiler.visitors.inner.Simplify; +import org.dbsp.sqlCompiler.compiler.visitors.outer.temporal.ContainsNow; +import org.dbsp.sqlCompiler.ir.DBSPParameter; +import org.dbsp.sqlCompiler.ir.expression.DBSPBinaryExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPClosureExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPCloneExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPDerefExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPFieldExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPIsNullExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPOpcode; +import org.dbsp.sqlCompiler.ir.expression.DBSPTupleExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPUnaryExpression; +import org.dbsp.sqlCompiler.ir.expression.DBSPVariablePath; +import org.dbsp.sqlCompiler.ir.expression.literal.DBSPLiteral; +import org.dbsp.sqlCompiler.ir.type.derived.DBSPTypeRef; +import org.dbsp.sqlCompiler.ir.type.derived.DBSPTypeTuple; + +import java.util.ArrayList; +import java.util.List; + +/** Decomposes a filter whose predicate contains expensive computations and now() into + * map -> filter -> map: + * - the first map appends the results of the expensive computations to the input row + * - the filter evaluates only cheap Boolean computations over the appended fields + * - the last map restores the original row type. + * Computations that involve now() are never hoisted. */ +public class DecomposeExpensiveFilters extends CircuitCloneVisitor { + final ContainsNow containsNow; + + public DecomposeExpensiveFilters(DBSPCompiler compiler) { + super(compiler, false); + this.containsNow = new ContainsNow(compiler, true); + } + + static boolean isConnective(DBSPOpcode opcode) { + return opcode == DBSPOpcode.AND || opcode == DBSPOpcode.OR; + } + + static boolean isComparison(DBSPOpcode opcode) { + return switch (opcode) { + case EQ, NEQ, LT, GT, LTE, GTE, IS_DISTINCT -> true; + default -> false; + }; + } + + static boolean isBooleanUnary(DBSPOpcode opcode) { + return switch (opcode) { + case NOT, WRAP_BOOL, IS_TRUE, IS_FALSE, IS_NOT_TRUE, IS_NOT_FALSE -> true; + default -> false; + }; + } + + /** Expressions that are as cheap as a hoisted field reference */ + static boolean isTrivial(DBSPExpression expression) { + while (expression.is(DBSPCloneExpression.class)) + expression = expression.to(DBSPCloneExpression.class).expression; + if (expression.is(DBSPLiteral.class)) + return true; + if (expression.is(DBSPFieldExpression.class)) + return expression.to(DBSPFieldExpression.class).expression.is(DBSPDerefExpression.class); + return false; + } + + /** Finds the maximal now-free subexpressions of a filter predicate worth + * hoisting into a preceding map, then rewrites the predicate to reference + * them as fields of a widened input tuple. */ + class Hoister { + final DBSPParameter param; + final List hoisted = new ArrayList<>(); + + Hoister(DBSPParameter param) { + this.param = param; + } + + /** Index of an equivalent computation present in this.hoisted, or -1. */ + int indexOf(DBSPExpression expression) { + for (int i = 0; i < this.hoisted.size(); i++) + if (EquivalenceContext.equiv( + this.hoisted.get(i).closure(this.param), + expression.closure(this.param))) + return i; + return -1; + } + + void collect(DBSPExpression expression) { + if (expression.is(DBSPBinaryExpression.class)) { + DBSPBinaryExpression bin = expression.to(DBSPBinaryExpression.class); + if (isConnective(bin.opcode)) { + this.collect(bin.left); + this.collect(bin.right); + return; + } + if (isComparison(bin.opcode)) { + this.operand(bin.left); + this.operand(bin.right); + return; + } + } else if (expression.is(DBSPUnaryExpression.class)) { + DBSPUnaryExpression unary = expression.to(DBSPUnaryExpression.class); + if (isBooleanUnary(unary.opcode)) { + this.collect(unary.source); + return; + } + } else if (expression.is(DBSPIsNullExpression.class)) { + this.operand(expression.to(DBSPIsNullExpression.class).expression); + return; + } + this.operand(expression); + } + + void operand(DBSPExpression expression) { + DecomposeExpensiveFilters.this.containsNow.apply(expression); + if (DecomposeExpensiveFilters.this.containsNow.found) + return; + if (isTrivial(expression)) + return; + if (!Expensive.isExpensive(DecomposeExpensiveFilters.this.compiler(), expression)) + return; + if (this.indexOf(expression) < 0) + this.hoisted.add(expression); + } + + DBSPExpression rewrite(DBSPExpression expression, DBSPVariablePath newVar, int base) { + if (expression.is(DBSPBinaryExpression.class)) { + DBSPBinaryExpression bin = expression.to(DBSPBinaryExpression.class); + if (isConnective(bin.opcode)) { + return new DBSPBinaryExpression(bin.getNode(), bin.getType(), bin.opcode, + this.rewrite(bin.left, newVar, base), + this.rewrite(bin.right, newVar, base)); + } + if (isComparison(bin.opcode)) { + return new DBSPBinaryExpression(bin.getNode(), bin.getType(), bin.opcode, + this.rewriteOperand(bin.left, newVar, base), + this.rewriteOperand(bin.right, newVar, base)); + } + } else if (expression.is(DBSPUnaryExpression.class)) { + DBSPUnaryExpression unary = expression.to(DBSPUnaryExpression.class); + if (isBooleanUnary(unary.opcode)) { + return new DBSPUnaryExpression(unary.getNode(), unary.getType(), unary.opcode, + this.rewrite(unary.source, newVar, base)); + } + } else if (expression.is(DBSPIsNullExpression.class)) { + DBSPIsNullExpression isNull = expression.to(DBSPIsNullExpression.class); + return new DBSPIsNullExpression(isNull.getNode(), + this.rewriteOperand(isNull.expression, newVar, base)); + } + return this.rewriteOperand(expression, newVar, base); + } + + /** Hoisted operands become field references; other operands keep + * referencing the original parameter. */ + DBSPExpression rewriteOperand(DBSPExpression expression, DBSPVariablePath newVar, int base) { + int index = this.indexOf(expression); + if (index >= 0) + return newVar.deref().field(base + index).applyCloneIfNeeded(); + return expression; + } + } + + @Override + public void postorder(DBSPFilterOperator operator) { + DBSPClosureExpression function = operator.getClosureFunction(); + Simplify simplify = new Simplify(this.compiler()); + function = simplify.apply(function).to(DBSPClosureExpression.class); + DBSPParameter param = function.parameters[0]; + Hoister hoister = new Hoister(param); + hoister.collect(function.body); + final boolean shouldHoist; + if (hoister.hoisted.isEmpty()) { + shouldHoist = false; + } else if (hoister.hoisted.size() == 1) { + // Heuristic: one expensive expression is hoisted only if the + // filter may be a temporal filter. + this.containsNow.apply(function); + shouldHoist = this.containsNow.found(); + } else { + shouldHoist = true; + } + if (!shouldHoist) { + super.postorder(operator); + return; + } + + DBSPTypeTuple inputType = param.type.to(DBSPTypeRef.class).deref().to(DBSPTypeTuple.class); + int n = inputType.size(); + int m = hoister.hoisted.size(); + + // Map appending the hoisted computations to the input row + DBSPExpression[] fields = new DBSPExpression[n + m]; + DBSPVariablePath t = param.asVariable(); + for (int i = 0; i < n; i++) + fields[i] = t.deref().field(i).applyCloneIfNeeded(); + for (int j = 0; j < m; j++) + fields[n + j] = hoister.hoisted.get(j); + DBSPTupleExpression tuple = new DBSPTupleExpression(fields); + DBSPClosureExpression mapFunction = tuple.closure(param); + DBSPMapOperator map = new DBSPMapOperator( + operator.getRelNode(), mapFunction, this.mapped(operator.input())); + this.addOperator(map); + + // Filter with the cheap predicate over the widened tuple. + DBSPVariablePath filterVar = tuple.getType().ref().var(); + DBSPExpression[] row = new DBSPExpression[n]; + for (int i = 0; i < n; i++) + row[i] = filterVar.deref().field(i).applyCloneIfNeeded(); + DBSPExpression newBody = hoister.rewrite(function.body, filterVar, n) + .closure(param) + .call(new DBSPTupleExpression(operator.getNode(), inputType, row).borrow()) + .reduce(this.compiler()); + DBSPFilterOperator filter = new DBSPFilterOperator( + operator.getRelNode(), + newBody.wrapBoolIfNeeded().closure(filterVar), map.outputPort()); + this.addOperator(filter); + + // Projection restoring the original row type + DBSPVariablePath projVar = tuple.getType().ref().var(); + DBSPExpression[] back = new DBSPExpression[n]; + for (int i = 0; i < n; i++) + back[i] = projVar.deref().field(i).applyCloneIfNeeded(); + DBSPClosureExpression projFunction = + new DBSPTupleExpression(operator.getNode(), inputType, back).closure(projVar); + DBSPMapOperator projection = new DBSPMapOperator( + operator.getRelNode(), projFunction, filter.outputPort()) + .addAnnotation(new IsProjection(n + m), DBSPMapOperator.class); + this.map(operator, projection); + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PullFilterVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PullFilterVisitor.java index 3e75c4278e1..985750f0d8b 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PullFilterVisitor.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PullFilterVisitor.java @@ -14,6 +14,7 @@ import org.dbsp.sqlCompiler.circuit.operator.DBSPSimpleOperator; import org.dbsp.sqlCompiler.compiler.DBSPCompiler; import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject; +import org.dbsp.sqlCompiler.compiler.visitors.inner.Expensive; import org.dbsp.sqlCompiler.ir.expression.DBSPBinaryExpression; import org.dbsp.sqlCompiler.ir.expression.DBSPClosureExpression; import org.dbsp.sqlCompiler.ir.expression.DBSPExpression; @@ -73,7 +74,10 @@ public void postorder(DBSPFilterOperator operator) { || source.node().is(DBSPMapIndexOperator.class)) { DBSPClosureExpression mapClosure = source.simpleNode().getClosureFunction(); DBSPClosureExpression filterClosure = operator.getClosureFunction(); - if (filterClosure.shouldInlineComposition(this.compiler, mapClosure)) { + // If we combine the two, the body of the map will be essentially executed twice + // so do it only if the map body is not expensive + boolean isExpensive = Expensive.isExpensive(compiler, mapClosure); + if (!isExpensive) { final DBSPClosureExpression newFilter; if (source.node().is(DBSPMapOperator.class)) { newFilter = filterClosure.applyAfter(this.compiler, mapClosure, Maybe.YES); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/temporal/ContainsNow.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/temporal/ContainsNow.java index c47836da6c9..fbe956199b5 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/temporal/ContainsNow.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/temporal/ContainsNow.java @@ -13,7 +13,7 @@ import javax.annotation.Nullable; /** Discovers whether an expression contains a call to the now() function. */ -class ContainsNow extends InnerVisitor { +public class ContainsNow extends InnerVisitor { public boolean found; /** If true the 'found' is reset for each invocation. */ public final boolean perExpression; diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPClosureExpression.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPClosureExpression.java index 692020924c6..4077b43e649 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPClosureExpression.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPClosureExpression.java @@ -33,6 +33,7 @@ import org.dbsp.sqlCompiler.compiler.visitors.inner.Expensive; import org.dbsp.sqlCompiler.compiler.visitors.inner.InnerVisitor; import org.dbsp.sqlCompiler.compiler.visitors.inner.Projection; +import org.dbsp.sqlCompiler.compiler.visitors.inner.ResolveReferences; import org.dbsp.sqlCompiler.ir.DBSPParameter; import org.dbsp.sqlCompiler.ir.IDBSPInnerNode; import org.dbsp.sqlCompiler.ir.type.DBSPType; @@ -189,6 +190,13 @@ public IIndentStream toString(IIndentStream builder) { .append(")"); } + /** Counts how many times a parameter is referenced within the closure body */ + int parameterReferences(DBSPCompiler compiler, DBSPParameter param) { + ResolveReferences resolver = new ResolveReferences(compiler, true); + resolver.apply(this); + return resolver.reference.count(param); + } + /** True if the composition this(before) can productively inline before */ public boolean shouldInlineComposition(DBSPCompiler compiler, DBSPClosureExpression before) { Projection projection = new Projection(compiler, true, true); @@ -196,8 +204,10 @@ public boolean shouldInlineComposition(DBSPCompiler compiler, DBSPClosureExpress if (projection.isProjection && before.body.is(DBSPBaseTupleExpression.class)) { return true; } else { - // TODO: this could be refined by checking how many times the source expression - // is substituted in the result. + int refCount = this.parameterReferences(compiler, this.parameters[0]); + if (refCount <= 1) + // Parameter referenced only once: allow inlining + return true; return !Expensive.isExpensive(compiler, before); } } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DecomposeExpensiveFiltersTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DecomposeExpensiveFiltersTests.java new file mode 100644 index 00000000000..0b983fed96d --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DecomposeExpensiveFiltersTests.java @@ -0,0 +1,143 @@ +package org.dbsp.sqlCompiler.compiler.visitors.outer; + +import org.dbsp.sqlCompiler.circuit.operator.DBSPSimpleOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPSourceBaseOperator; +import org.dbsp.sqlCompiler.compiler.DBSPCompiler; +import org.dbsp.sqlCompiler.compiler.sql.tools.SqlIoTest; +import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision; +import org.dbsp.sqlCompiler.compiler.visitors.inner.InnerVisitor; +import org.dbsp.sqlCompiler.ir.expression.DBSPApplyExpression; +import org.dbsp.sqlCompiler.ir.type.DBSPType; +import org.junit.Assert; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +/** Tests for {@link DecomposeExpensiveFilters} */ +public class DecomposeExpensiveFiltersTests extends SqlIoTest { + /** Counts the calls to a function in the whole circuit. + * Matches by prefix. */ + static class CountCalls extends InnerVisitor { + final String function; + int count = 0; + + public CountCalls(DBSPCompiler compiler, String function) { + super(compiler); + this.function = function; + } + + @Override + public VisitDecision preorder(DBSPType type) { + return VisitDecision.STOP; + } + + @Override + public void postorder(DBSPApplyExpression node) { + String name = node.getFunctionName(); + if (name != null && name.startsWith(this.function)) + this.count++; + } + } + + /** Collects non-source operators whose output contains a VARIANT field */ + static class VariantOutputs extends CircuitVisitor { + final List operators = new ArrayList<>(); + + public VariantOutputs(DBSPCompiler compiler) { + super(compiler); + } + + @Override + public void postorder(DBSPSimpleOperator operator) { + if (operator.is(DBSPSourceBaseOperator.class)) + return; + if (operator.outputType.toString().contains("VARIANT")) + this.operators.add(operator.getClass().getSimpleName() + " " + operator.getIdString()); + } + } + + /** A view filtering on expensive computations over a VARIANT column, + * compared against NOW() windows. The filter must be decomposed so that + * - the expensive function is evaluated once per distinct argument + * - the VARIANT column does not flow past the map holding the + * hoisted computations. */ + @Test + public void testDecomposition() { + String sql = """ + CREATE TABLE data ( + id VARCHAR NOT NULL PRIMARY KEY, + properties VARIANT, + sid VARCHAR + ); + CREATE FUNCTION to_ts(d VARCHAR) RETURNS TIMESTAMP AS + PARSE_TIMESTAMP('%Y-%m-%d %H:%M:%S', d); + CREATE VIEW segment AS + SELECT id, sid FROM data u + WHERE u.sid = 'x' + AND TO_TS(SAFE_CAST(u.properties['created'] AS VARCHAR)) + BETWEEN NOW() - INTERVAL 93 DAYS AND NOW() - INTERVAL 1 DAYS + AND TO_TS(SAFE_CAST(u.properties['deleted'] AS VARCHAR)) + >= NOW() - INTERVAL 1 MONTHS + AND SAFE_CAST(u.properties['opt'] AS VARCHAR) = 'true';"""; + DBSPCompiler compiler = this.testCompiler(); + compiler.submitStatementsForCompilation(sql); + var ccs = this.getCCS(compiler); + + CountCalls counter = new CountCalls(compiler, "to_ts"); + ccs.visit(counter.getCircuitVisitor(false)); + Assert.assertEquals(2, counter.count); + + VariantOutputs variants = new VariantOutputs(compiler); + ccs.visit(variants); + Assert.assertEquals(List.of(), variants.operators); + } + + /** Operands containing nested closures (array lambdas) must be hoisted + * whole, and the two structurally equal copies that BETWEEN creates must + * share one hoisted column despite containing closures. */ + @Test + public void testNestedClosure() { + String sql = """ + CREATE TABLE data ( + id VARCHAR NOT NULL PRIMARY KEY, + properties VARIANT, + sid VARCHAR + ); + CREATE VIEW segment AS + SELECT id, sid FROM data u + WHERE u.sid = 'x' + AND ARRAY_EXISTS(CAST(u.properties['tags'] AS VARCHAR ARRAY), t -> t = 'pro') + AND CARDINALITY(TRANSFORM(CAST(u.properties['tags'] AS VARCHAR ARRAY), t -> UPPER(t))) + BETWEEN 1 AND 5;"""; + DBSPCompiler compiler = this.testCompiler(); + compiler.submitStatementsForCompilation(sql); + var ccs = this.getCCS(compiler).withStringTrim(); + + CountCalls exists = new CountCalls(compiler, "array_exists"); + ccs.visit(exists.getCircuitVisitor(false)); + Assert.assertEquals(1, exists.count); + + CountCalls transform = new CountCalls(compiler, "transform"); + ccs.visit(transform.getCircuitVisitor(false)); + Assert.assertEquals(1, transform.count); + + VariantOutputs variants = new VariantOutputs(compiler); + ccs.visit(variants); + Assert.assertEquals(List.of(), variants.operators); + + // Only 'a' passes: 'b' lacks the 'pro' tag, 'c' has the wrong sid, + // 'd' has more than 5 tags, 'e' has no properties + ccs.stepWeightOne(""" + INSERT INTO data VALUES + ('a', PARSE_JSON('{"tags": ["pro", "basic"]}'), 'x'), + ('b', PARSE_JSON('{"tags": ["basic"]}'), 'x'), + ('c', PARSE_JSON('{"tags": ["pro"]}'), 'y'), + ('d', PARSE_JSON('{"tags": ["pro", "t1", "t2", "t3", "t4", "t5"]}'), 'x'), + ('e', NULL, 'x');""", + """ + id | sid + ---------- + a | x"""); + } +}