diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/AntiJoinDistinctRemoveRule.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/AntiJoinDistinctRemoveRule.java
new file mode 100644
index 00000000000..48bb8ffa629
--- /dev/null
+++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/AntiJoinDistinctRemoveRule.java
@@ -0,0 +1,110 @@
+package org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.optimizer;
+
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Aggregate;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.rules.TransformationRule;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlKind;
+
+/**
+ * Rule that removes a DISTINCT (an Aggregate with no aggregate calls grouping
+ * by all its columns) from the right input of a LEFT JOIN, when a filter above
+ * the join requires a non-nullable right column to be NULL.
+ * This is the SQL idiom for an anti-join.
+ *
+ *
+ * LogicalFilter(condition=[IS NULL($r)]) $r a right column, non-nullable below
+ * LogicalJoin(joinType=[left])
+ * any(left)
+ * LogicalAggregate(group=[{0..n}]) no aggregate calls
+ *
+ * becomes
+ *
+ * LogicalFilter(condition=[IS NULL($r)])
+ * LogicalJoin(joinType=[left])
+ * any(left)
+ * input of the LogicalAggregate
+ *
+ *
+ * The plans are equivalent because the filter output is identical:
+ * a left row with matches only produces join rows where $r is not NULL,
+ * so the IS NULL conjunct deletes all of them, deduplicated or not;
+ * a left row without matches produces exactly one NULL-padded row,
+ * and deduplication cannot change which left rows have no matches.
+ *
+ *
It does not hold for FULL joins: the unmatched rows of the
+ * deduplicated input are themselves part of the output. */
+public class AntiJoinDistinctRemoveRule
+ extends RelRule>
+ implements TransformationRule {
+ protected AntiJoinDistinctRemoveRule() {
+ super(CONFIG);
+ }
+
+ /** True if the aggregate only deduplicates its input:
+ * it groups by all its columns and computes nothing */
+ static boolean isDistinct(Aggregate aggregate) {
+ return aggregate.getGroupType() == Aggregate.Group.SIMPLE
+ && aggregate.getAggCallList().isEmpty()
+ && aggregate.getGroupSet().cardinality()
+ == aggregate.getInput().getRowType().getFieldCount();
+ }
+
+ /** True if some conjunct has the form IS NULL(column), where column is
+ * a right-side column of the join that cannot be NULL below the join.
+ * Such a conjunct only accepts left rows with no matches. */
+ static boolean requiresNoMatch(Filter filter, Join join, Aggregate right) {
+ int leftCount = join.getLeft().getRowType().getFieldCount();
+ for (RexNode conjunct : RelOptUtil.conjunctions(filter.getCondition())) {
+ if (conjunct.getKind() != SqlKind.IS_NULL)
+ continue;
+ RexNode operand = ((RexCall) conjunct).getOperands().get(0);
+ if (!(operand instanceof RexInputRef ref))
+ continue;
+ int index = ref.getIndex() - leftCount;
+ if (index < 0)
+ continue;
+ // The join row type declares right columns nullable because the
+ // join pads them; the nullability below the join is what matters
+ if (!right.getRowType().getFieldList().get(index).getType().isNullable())
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final Filter filter = call.rel(0);
+ final Join join = call.rel(1);
+ final Aggregate aggregate = call.rel(3);
+ if (join.getJoinType() != JoinRelType.LEFT)
+ return;
+ if (!isDistinct(aggregate))
+ return;
+ if (!requiresNoMatch(filter, join, aggregate))
+ return;
+
+ Join newJoin = join.copy(join.getTraitSet(), join.getCondition(),
+ join.getLeft(), aggregate.getInput(), join.getJoinType(), join.isSemiJoinDone());
+ Filter newFilter = filter.copy(filter.getTraitSet(), newJoin, filter.getCondition());
+ call.transformTo(newFilter);
+ call.getPlanner().prune(filter);
+ }
+
+ public static final DefaultOptRuleConfig CONFIG =
+ DefaultOptRuleConfig.create()
+ .withOperandSupplier(
+ b0 -> b0.operand(Filter.class)
+ .oneInput(b1 -> b1.operand(Join.class)
+ .inputs(
+ b2 -> b2.operand(RelNode.class).anyInputs(),
+ b3 -> b3.operand(Aggregate.class).anyInputs())));
+}
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java
index 1f0b7813a69..74a896b7458 100644
--- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java
+++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java
@@ -438,6 +438,7 @@ HepProgram getProgram(RelNode node, int level) {
this.addStep(merge);
this.addStep(new SimpleOptimizerStep("Remove dead code", 0,
CoreRules.AGGREGATE_REMOVE,
+ new AntiJoinDistinctRemoveRule(),
CoreRules.UNION_REMOVE,
CoreRules.PROJECT_REMOVE,
CoreRules.PROJECT_JOIN_JOIN_REMOVE,
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/AntiJoinDistinctRemoveRuleTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/AntiJoinDistinctRemoveRuleTests.java
new file mode 100644
index 00000000000..10b35af7819
--- /dev/null
+++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/AntiJoinDistinctRemoveRuleTests.java
@@ -0,0 +1,209 @@
+package org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.optimizer;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.hep.HepPlanner;
+import org.apache.calcite.plan.hep.HepProgram;
+import org.apache.calcite.plan.hep.HepProgramBuilder;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Aggregate;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.RelFactories;
+import org.apache.calcite.rel.logical.LogicalFilter;
+import org.apache.calcite.rel.type.RelDataTypeSystem;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeFactoryImpl;
+import org.apache.calcite.tools.RelBuilder;
+import org.junit.Assert;
+import org.junit.Test;
+
+/** Unit tests for {@link AntiJoinDistinctRemoveRule}, applied directly
+ * to relational plans built with a {@link RelBuilder}.
+ * Each plan reads two single-column collections built from VALUES:
+ * t with column x, and s with column y; the SQL comments use these names. */
+public class AntiJoinDistinctRemoveRuleTests {
+ static RelBuilder createBuilder() {
+ // Not RelBuilder.create(FrameworkConfig): that route needs a JDBC connection
+ RexBuilder rexBuilder = new RexBuilder(new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT));
+ RelOptCluster cluster = RelOptCluster.create(
+ new HepPlanner(new HepProgramBuilder().build()), rexBuilder);
+ return RelFactories.LOGICAL_BUILDER.create(cluster, null)
+ // The builder must not simplify the plans the tests specify:
+ // it can remove a distinct() over provably-unique VALUES and
+ // fold IS NULL of a non-nullable column to FALSE
+ .transform(config -> config
+ .withSimplify(false)
+ .withSimplifyValues(false)
+ .withAggregateUnique(true)
+ .withPruneInputOfAggregate(false));
+ }
+
+ static RelNode optimize(RelNode node) {
+ HepProgram program = new HepProgramBuilder()
+ .addRuleInstance(new AntiJoinDistinctRemoveRule())
+ .build();
+ HepPlanner planner = new HepPlanner(program);
+ planner.setRoot(node);
+ return planner.findBestExp();
+ }
+
+ static int countAggregates(RelNode node) {
+ int count = node instanceof Aggregate ? 1 : 0;
+ for (RelNode input : node.getInputs())
+ count += countAggregates(input);
+ return count;
+ }
+
+ /** Optimize the plan, which must contain exactly one Aggregate.
+ * @return True if the optimization removed the Aggregate. */
+ static boolean distinctRemoved(RelNode plan) {
+ Assert.assertEquals(RelOptUtil.toString(plan), 1, countAggregates(plan));
+ RelNode optimized = optimize(plan);
+ int count = countAggregates(optimized);
+ // The rule can only remove the Aggregate, never add one
+ Assert.assertTrue(RelOptUtil.toString(optimized), count <= 1);
+ return count == 0;
+ }
+
+ /** Filter(IS NULL(right col)) over LeftJoin(left, Distinct(right)),
+ * with the right column non-nullable: the distinct must be removed. */
+ @Test
+ public void removesDistinct() {
+ // SELECT * FROM t
+ // LEFT JOIN (SELECT DISTINCT y FROM s) d ON t.x = d.y
+ // WHERE d.y IS NULL
+ RelBuilder builder = createBuilder();
+ RelNode plan = builder
+ .values(new String[]{"x"}, 1, 2)
+ .values(new String[]{"y"}, 1, 1)
+ .distinct()
+ .join(JoinRelType.LEFT,
+ builder.equals(builder.field(2, 0, "x"), builder.field(2, 1, "y")))
+ .filter(builder.isNull(builder.field("y")))
+ .build();
+ Assert.assertTrue(distinctRemoved(plan));
+ }
+
+ /** Extra conjuncts over other columns do not prevent the rewrite */
+ @Test
+ public void removesDistinctExtraConjunct() {
+ // SELECT * FROM t
+ // LEFT JOIN (SELECT DISTINCT y FROM s) d ON t.x = d.y
+ // WHERE d.y IS NULL AND t.x > 0
+ RelBuilder builder = createBuilder();
+ RelNode plan = builder
+ .values(new String[]{"x"}, 1, 2)
+ .values(new String[]{"y"}, 1, 1)
+ .distinct()
+ .join(JoinRelType.LEFT,
+ builder.equals(builder.field(2, 0, "x"), builder.field(2, 1, "y")))
+ .filter(
+ builder.isNull(builder.field("y")),
+ builder.greaterThan(builder.field("x"), builder.literal(0)))
+ .build();
+ Assert.assertTrue(distinctRemoved(plan));
+ }
+
+ /** A nullable right column can be NULL in a matched row,
+ * so IS NULL does not prove the absence of a match. */
+ @Test
+ public void keepsNullableColumn() {
+ // SELECT * FROM t
+ // LEFT JOIN (SELECT DISTINCT y FROM s) d ON t.x = d.y
+ // WHERE d.y IS NULL
+ // where s.y is nullable
+ RelBuilder builder = createBuilder();
+ RelNode plan = builder
+ .values(new String[]{"x"}, 1, 2)
+ .values(new String[]{"y"}, 1, null)
+ .distinct()
+ .join(JoinRelType.LEFT,
+ builder.equals(builder.field(2, 0, "x"), builder.field(2, 1, "y")))
+ .filter(builder.isNull(builder.field("y")))
+ .build();
+ Assert.assertFalse(distinctRemoved(plan));
+ }
+
+ /** An inner join propagates right multiplicities to the output */
+ @Test
+ public void keepsInnerJoin() {
+ // SELECT * FROM t
+ // JOIN (SELECT DISTINCT y FROM s) d ON t.x = d.y
+ // WHERE d.y IS NULL
+ RelBuilder builder = createBuilder();
+ RelNode join = builder
+ .values(new String[]{"x"}, 1, 2)
+ .values(new String[]{"y"}, 1, 1)
+ .distinct()
+ .join(JoinRelType.INNER,
+ builder.equals(builder.field(2, 0, "x"), builder.field(2, 1, "y")))
+ .build();
+ // Build the filter directly to avoid optimization by the builder
+ RexBuilder rexBuilder = join.getCluster().getRexBuilder();
+ RelNode plan = LogicalFilter.create(join,
+ rexBuilder.makeCall(SqlStdOperatorTable.IS_NULL,
+ rexBuilder.makeInputRef(join, 1)));
+ Assert.assertFalse(distinctRemoved(plan));
+ }
+
+ /** An Aggregate that computes something is not a plain DISTINCT */
+ @Test
+ public void keepsRealAggregate() {
+ // SELECT * FROM t
+ // LEFT JOIN (SELECT y, COUNT(*) c FROM s GROUP BY y) d ON t.x = d.y
+ // WHERE d.y IS NULL
+ RelBuilder builder = createBuilder();
+ RelNode plan = builder
+ .values(new String[]{"x"}, 1, 2)
+ .values(new String[]{"y"}, 1, 1)
+ .aggregate(builder.groupKey(0), builder.count(false, "c"))
+ .join(JoinRelType.LEFT,
+ builder.equals(builder.field(2, 0, "x"), builder.field(2, 1, "y")))
+ .filter(builder.isNull(builder.field("y")))
+ .build();
+ Assert.assertFalse(distinctRemoved(plan));
+ }
+
+ /** An Aggregate grouping by a subset of its columns changes the schema,
+ * so it cannot be replaced by its input. */
+ @Test
+ public void keepsPartialGroupBy() {
+ // SELECT * FROM t
+ // LEFT JOIN (SELECT y FROM s2 GROUP BY y) d ON t.x = d.y
+ // WHERE d.y IS NULL
+ // where s2 has columns (y, z) and the Aggregate reads both
+ RelBuilder builder = createBuilder();
+ RelNode plan = builder
+ .values(new String[]{"x"}, 1, 2)
+ .values(new String[]{"y", "z"}, 1, 10, 2, 20)
+ .aggregate(builder.groupKey(0))
+ .join(JoinRelType.LEFT,
+ builder.equals(builder.field(2, 0, "x"), builder.field(2, 1, "y")))
+ .filter(builder.isNull(builder.field("y")))
+ .build();
+ Assert.assertFalse(distinctRemoved(plan));
+ }
+
+ /** IS NOT NULL selects the matched rows, whose multiplicity the
+ * distinct bounds; only IS NULL is rewritten. */
+ @Test
+ public void keepsIsNotNull() {
+ // SELECT * FROM t
+ // LEFT JOIN (SELECT DISTINCT y FROM s) d ON t.x = d.y
+ // WHERE NOT(d.y IS NULL)
+ RelBuilder builder = createBuilder();
+ RexNode isNull = builder
+ .values(new String[]{"x"}, 1, 2)
+ .values(new String[]{"y"}, 1, 1)
+ .distinct()
+ .join(JoinRelType.LEFT,
+ builder.equals(builder.field(2, 0, "x"), builder.field(2, 1, "y")))
+ .isNull(builder.field("y"));
+ RelNode plan = builder
+ .filter(builder.not(isNull))
+ .build();
+ Assert.assertFalse(distinctRemoved(plan));
+ }
+}
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/AntiJoinDistinctTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/AntiJoinDistinctTests.java
new file mode 100644
index 00000000000..12f5cf4468c
--- /dev/null
+++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/AntiJoinDistinctTests.java
@@ -0,0 +1,149 @@
+package org.dbsp.sqlCompiler.compiler.sql.simple;
+
+import org.dbsp.sqlCompiler.circuit.operator.DBSPDistinctOperator;
+import org.dbsp.sqlCompiler.circuit.operator.DBSPStreamDistinctOperator;
+import org.dbsp.sqlCompiler.compiler.DBSPCompiler;
+import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.optimizer.AntiJoinDistinctRemoveRule;
+import org.dbsp.sqlCompiler.compiler.sql.tools.SqlIoTest;
+import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor;
+import org.junit.Assert;
+import org.junit.Test;
+
+/** Tests for {@link AntiJoinDistinctRemoveRule} */
+public class AntiJoinDistinctTests extends SqlIoTest {
+ /** Counts the distinct operators in the circuit */
+ static class CountDistinct extends CircuitVisitor {
+ int count = 0;
+
+ public CountDistinct(DBSPCompiler compiler) {
+ super(compiler);
+ }
+
+ @Override
+ public void postorder(DBSPDistinctOperator operator) {
+ this.count++;
+ }
+
+ @Override
+ public void postorder(DBSPStreamDistinctOperator operator) {
+ this.count++;
+ }
+ }
+
+ int countDistinct(String sql) {
+ DBSPCompiler compiler = this.testCompiler();
+ compiler.submitStatementsForCompilation(sql);
+ var ccs = this.getCCS(compiler);
+ CountDistinct counter = new CountDistinct(compiler);
+ ccs.visit(counter);
+ return counter.count;
+ }
+
+ static final String TABLES = """
+ CREATE TABLE t(x INT NOT NULL);
+ CREATE TABLE s(x INT NOT NULL);
+ """;
+
+ /** The rule removes the GROUP BY: the join result is only null-tested */
+ @Test
+ public void testRemoved() {
+ String sql = TABLES + """
+ CREATE VIEW v AS
+ SELECT t.x FROM t
+ LEFT JOIN (SELECT x FROM s GROUP BY x) g ON t.x = g.x
+ WHERE g.x IS NULL;""";
+ DBSPCompiler compiler = this.testCompiler();
+ compiler.submitStatementsForCompilation(sql);
+ var ccs = this.getCCS(compiler);
+
+ CountDistinct counter = new CountDistinct(compiler);
+ ccs.visit(counter);
+ Assert.assertEquals(0, counter.count);
+
+ // The duplicates in s must not duplicate or revive t's rows
+ ccs.stepWeightOne("""
+ INSERT INTO t VALUES(1), (2);
+ INSERT INTO s VALUES(2), (2);""",
+ """
+ x
+ ---
+ 1""");
+ }
+
+ /** The rule must not fire when the tested column is nullable:
+ * NULL no longer proves the absence of a match. */
+ @Test
+ public void testNullableColumn() {
+ String sql = """
+ CREATE TABLE t(x INT NOT NULL);
+ CREATE TABLE s(x INT);
+ CREATE VIEW v AS
+ SELECT t.x FROM t
+ LEFT JOIN (SELECT x FROM s GROUP BY x) g ON t.x = g.x
+ WHERE g.x IS NULL;""";
+ Assert.assertEquals(1, this.countDistinct(sql));
+ }
+
+ /** The rule must not fire for the IS NOT NULL idiom: there the
+ * GROUP BY bounds the multiplicity of the join result. */
+ @Test
+ public void testNotNullTest() {
+ String sql = TABLES + """
+ CREATE VIEW v AS
+ SELECT t.x FROM t
+ LEFT JOIN (SELECT x FROM s GROUP BY x) g ON t.x = g.x
+ WHERE g.x IS NOT NULL;""";
+ DBSPCompiler compiler = this.testCompiler();
+ compiler.submitStatementsForCompilation(sql);
+ var ccs = this.getCCS(compiler);
+
+ CountDistinct counter = new CountDistinct(compiler);
+ ccs.visit(counter);
+ Assert.assertEquals(1, counter.count);
+
+ // Without the GROUP BY the row 2 would appear twice
+ ccs.stepWeightOne("""
+ INSERT INTO t VALUES(1), (2);
+ INSERT INTO s VALUES(2), (2);""", """
+ x
+ ---
+ 2""");
+ }
+
+ /** Only the GROUP BY under the IS NULL branch is removable.
+ * The view selects the people who started something and never finished. */
+ @Test
+ public void testSegment() {
+ String sql = """
+ CREATE TABLE people(id VARCHAR NOT NULL PRIMARY KEY);
+ CREATE TABLE events(id VARCHAR NOT NULL, kind VARCHAR);
+ CREATE VIEW segment AS
+ SELECT p.id FROM people p
+ LEFT JOIN (SELECT id FROM events WHERE kind = 'start' GROUP BY id) s
+ ON p.id = s.id
+ LEFT JOIN (SELECT id FROM events WHERE kind = 'finish' GROUP BY id) f
+ ON p.id = f.id
+ WHERE s.id IS NOT NULL AND f.id IS NULL;""";
+ DBSPCompiler compiler = this.testCompiler();
+ compiler.submitStatementsForCompilation(sql);
+ var ccs = this.getCCS(compiler).withStringTrim();
+
+ CountDistinct counter = new CountDistinct(compiler);
+ ccs.visit(counter);
+ Assert.assertEquals(1, counter.count);
+
+ // 'a' finished, 'c' never started; the duplicated start events
+ // of 'b' must produce a single output row, and the duplicated finish
+ // events of 'a' exercise the branch whose GROUP BY the rule removed.
+ // PostgreSQL 14.13 returns the same result for this data, both for
+ // this query and for one without the GROUP BY in the 'finish' branch.
+ ccs.stepWeightOne("""
+ INSERT INTO people VALUES('a'), ('b'), ('c');
+ INSERT INTO events VALUES
+ ('a', 'start'), ('a', 'finish'), ('a', 'finish'),
+ ('b', 'start'), ('b', 'start');""", """
+ id
+ ----
+ b""");
+ }
+}