feldera · mihaibudiu · Aug 7, 2026 · Aug 6, 2026
diff --git a/...er/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/Expensive.java b/...er/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-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/inner/ReferenceMap.java b/...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/...compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java b/...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/...src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DecomposeExpensiveFilters.java b/...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<DBSPExpression> 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/...ompiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/PullFilterVisitor.java b/...ompiler/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/...iler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/temporal/ContainsNow.java b/...iler/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-compiler/src/main/java/org/dbsp/sqlCompiler/ir/expression/DBSPClosureExpression.java b/.../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,15 +190,24 @@ 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);
         projection.apply(this);
         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);
         }
     }