This is an automated email from the ASF dual-hosted git repository.
julianhyde pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/calcite.git
The following commit(s) were added to refs/heads/main by this push:
new 9089f40a0a [CALCITE-7628] In the interpreter, MINUS and INTERSECT with
3 or more inputs return wrong result
9089f40a0a is described below
commit 9089f40a0ab3220ef6344f3e6e067b32c438d9bd
Author: Julian Hyde <[email protected]>
AuthorDate: Sun Jun 28 13:13:38 2026 -0700
[CALCITE-7628] In the interpreter, MINUS and INTERSECT with 3 or more
inputs return wrong result
In the interpreter, a query where MINUS or INTERSECT has 3 or more
inputs previously returned the wrong result, because SetOpNode evaluated
only the first two inputs. It now evaluates all inputs.
We add tests in a new file `interpreter.iq`, that runs SQL queries using
the interpreter.
Add optimized implementations of MINUS DISTINCT and INTERSECT DISTINCT
using mutable counts of the number of occurrences of each key; avoid
materializing every input of a MINUS or INTERSECT if the output becomes
empty.
Remove dead code in `RelFieldTrimmer` (killed by CALCITE-3399).
Close apache/calcite#5055
---
.../org/apache/calcite/interpreter/SetOpNode.java | 236 +++++++++++++++++----
.../apache/calcite/sql2rel/RelFieldTrimmer.java | 15 +-
.../org/apache/calcite/test/InterpreterTest.java | 2 +-
.../java/org/apache/calcite/test/JdbcTest.java | 9 +-
core/src/test/resources/sql/interpreter.iq | 171 +++++++++++++++
5 files changed, 374 insertions(+), 59 deletions(-)
diff --git a/core/src/main/java/org/apache/calcite/interpreter/SetOpNode.java
b/core/src/main/java/org/apache/calcite/interpreter/SetOpNode.java
index 2d5750efd7..9e803cb809 100644
--- a/core/src/main/java/org/apache/calcite/interpreter/SetOpNode.java
+++ b/core/src/main/java/org/apache/calcite/interpreter/SetOpNode.java
@@ -18,10 +18,15 @@
import org.apache.calcite.rel.core.SetOp;
-import com.google.common.collect.HashMultiset;
-
-import java.util.Collection;
+import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import static org.apache.calcite.util.Util.transformIndexed;
/**
* Interpreter node that implements a
@@ -31,63 +36,216 @@
* {@link org.apache.calcite.rel.core.Intersect}.
*/
public class SetOpNode implements Node {
- private final Source leftSource;
- private final Source rightSource;
+ private final List<Source> sources;
private final Sink sink;
private final SetOp setOp;
public SetOpNode(Compiler compiler, SetOp setOp) {
- leftSource = compiler.source(setOp, 0);
- rightSource = compiler.source(setOp, 1);
+ final int arity = setOp.getInputs().size();
+ checkArgument(arity >= 2, "invalid set op arity %s", arity);
+ sources = transformIndexed(setOp.getInputs(), (r, i) ->
compiler.source(setOp, i));
+ assert sources.size() == arity;
sink = compiler.sink(setOp);
this.setOp = setOp;
}
@Override public void close() {
- leftSource.close();
- rightSource.close();
+ for (Source source : sources) {
+ source.close();
+ }
}
@Override public void run() throws InterruptedException {
- final Collection<Row> leftRows;
- final Collection<Row> rightRows;
- if (setOp.all) {
- leftRows = HashMultiset.create();
- rightRows = HashMultiset.create();
- } else {
- leftRows = new HashSet<>();
- rightRows = new HashSet<>();
- }
- Row row;
- while ((row = leftSource.receive()) != null) {
- leftRows.add(row);
- }
- while ((row = rightSource.receive()) != null) {
- rightRows.add(row);
- }
switch (setOp.kind) {
- case INTERSECT:
- for (Row leftRow : leftRows) {
- if (rightRows.remove(leftRow)) {
- sink.send(leftRow);
- }
+ case UNION:
+ if (setOp.all) {
+ unionAll();
+ } else {
+ unionDistinct();
}
break;
- case EXCEPT:
- for (Row leftRow : leftRows) {
- if (!rightRows.remove(leftRow)) {
- sink.send(leftRow);
- }
+ case INTERSECT:
+ if (setOp.all) {
+ intersectAll();
+ } else {
+ intersectDistinct();
}
break;
- case UNION:
- leftRows.addAll(rightRows);
- for (Row r : leftRows) {
- sink.send(r);
+ case EXCEPT:
+ if (setOp.all) {
+ minusAll();
+ } else {
+ minusDistinct();
}
break;
default:
break;
}
}
+
+ /** Evaluates UNION ALL. Does not need to buffer: sends each row to the
+ * output as it arrives. */
+ private void unionAll() throws InterruptedException {
+ for (Source source : sources) {
+ Row row;
+ while ((row = source.receive()) != null) {
+ sink.send(row);
+ }
+ }
+ }
+
+ /** Evaluates UNION DISTINCT. Does not need to buffer: sends each row to the
+ * output as it arrives, eliminating duplicates on the fly. */
+ private void unionDistinct() throws InterruptedException {
+ final Set<Row> seen = new HashSet<>();
+ for (Source source : sources) {
+ Row row;
+ while ((row = source.receive()) != null) {
+ if (seen.add(row)) {
+ sink.send(row);
+ }
+ }
+ }
+ }
+
+ /** Evaluates INTERSECT ALL by counting each value's occurrences in every
+ * input and emitting it the minimum number of times. 'min' holds the
+ * smallest count seen across the inputs processed so far; 'current' counts
+ * the occurrences in the input being processed. */
+ private void intersectAll() throws InterruptedException {
+ final Map<Row, CountPair> counts = new HashMap<>();
+ Row row;
+ final Source first = sources.get(0);
+ while ((row = first.receive()) != null) {
+ counts.computeIfAbsent(row, k -> new CountPair()).min++;
+ }
+ final int last = sources.size() - 1;
+ for (int i = 1; i < last && !counts.isEmpty(); i++) {
+ final Source source = sources.get(i);
+ while ((row = source.receive()) != null) {
+ final CountPair pair = counts.get(row);
+ if (pair != null) {
+ pair.current++;
+ }
+ }
+ // Reduce the running minimum, dropping values absent from this input.
+ counts.values().removeIf(pair -> {
+ if (pair.current == 0) {
+ return true;
+ }
+ pair.min = Math.min(pair.min, pair.current);
+ pair.current = 0;
+ return false;
+ });
+ }
+ // Last input: count occurrences and emit each value that occurs in it
+ // min(min, current) times.
+ if (counts.isEmpty()) {
+ return;
+ }
+ final Source source = sources.get(last);
+ while ((row = source.receive()) != null) {
+ final CountPair pair = counts.get(row);
+ if (pair != null) {
+ pair.current++;
+ }
+ }
+ for (Map.Entry<Row, CountPair> entry : counts.entrySet()) {
+ final CountPair pair = entry.getValue();
+ if (pair.current > 0) {
+ for (int n = Math.min(pair.min, pair.current); n > 0; n--) {
+ sink.send(entry.getKey());
+ }
+ }
+ }
+ }
+
+ /** Evaluates INTERSECT DISTINCT by retaining, for each successive input, the
+ * rows it has in common with the result so far. Inputs after the first are
+ * streamed, so only the result set (which only shrinks) is held in memory.
*/
+ private void intersectDistinct() throws InterruptedException {
+ Set<Row> result = read(sources.get(0));
+ for (int i = 1; i < sources.size() && !result.isEmpty(); i++) {
+ final Source source = sources.get(i);
+ final Set<Row> next = new HashSet<>();
+ Row row;
+ while ((row = source.receive()) != null) {
+ if (result.contains(row)) {
+ next.add(row);
+ }
+ }
+ result = next;
+ }
+ for (Row r : result) {
+ sink.send(r);
+ }
+ }
+
+ /** Evaluates EXCEPT ALL by counting occurrences of each value in a map. A
row
+ * from the first input increments its value's count; a row from a later
input
+ * decrements it, and a value whose count reaches zero is removed. After all
+ * inputs have been read, emit each surviving value as many times as its
+ * remaining count. */
+ private void minusAll() throws InterruptedException {
+ final Map<Row, Count> counts = new HashMap<>();
+ Row row;
+ final Source first = sources.get(0);
+ while ((row = first.receive()) != null) {
+ counts.computeIfAbsent(row, k -> new Count()).i++;
+ }
+ for (int i = 1; i < sources.size() && !counts.isEmpty(); i++) {
+ final Source source = sources.get(i);
+ while ((row = source.receive()) != null) {
+ final Count count = counts.get(row);
+ if (count != null && --count.i == 0) {
+ counts.remove(row);
+ }
+ }
+ }
+ for (Map.Entry<Row, Count> entry : counts.entrySet()) {
+ for (int n = entry.getValue().i; n > 0; n--) {
+ sink.send(entry.getKey());
+ }
+ }
+ }
+
+ /** Evaluates EXCEPT DISTINCT by removing from the running result every row
+ * that occurs in a later input. Later inputs are streamed, so only the
result
+ * set is held in memory. */
+ private void minusDistinct() throws InterruptedException {
+ final Set<Row> result = read(sources.get(0));
+ for (int i = 1; i < sources.size() && !result.isEmpty(); i++) {
+ final Source source = sources.get(i);
+ Row row;
+ while ((row = source.receive()) != null) {
+ result.remove(row);
+ }
+ }
+ for (Row r : result) {
+ sink.send(r);
+ }
+ }
+
+ /** Reads a single input into a set, eliminating duplicates. */
+ private static Set<Row> read(Source source) {
+ final Set<Row> rows = new HashSet<>();
+ Row row;
+ while ((row = source.receive()) != null) {
+ rows.add(row);
+ }
+ return rows;
+ }
+
+ /** Mutable count of the occurrences of a row, used by {@link #minusAll()}.
*/
+ private static class Count {
+ int i;
+ }
+
+ /** Mutable pair of counts used by {@link #intersectAll()}: the minimum
+ * multiplicity of a row across the inputs seen so far, and its count in the
+ * input currently being processed. */
+ private static class CountPair {
+ int min;
+ int current;
+ }
}
diff --git a/core/src/main/java/org/apache/calcite/sql2rel/RelFieldTrimmer.java
b/core/src/main/java/org/apache/calcite/sql2rel/RelFieldTrimmer.java
index 48cb1f0521..d47adb17b5 100644
--- a/core/src/main/java/org/apache/calcite/sql2rel/RelFieldTrimmer.java
+++ b/core/src/main/java/org/apache/calcite/sql2rel/RelFieldTrimmer.java
@@ -1076,20 +1076,7 @@ public TrimResult trimFields(
return result(setOp, mapping);
}
- switch (setOp.kind) {
- case UNION:
- relBuilder.union(setOp.all, setOp.getInputs().size());
- break;
- case INTERSECT:
- relBuilder.intersect(setOp.all, setOp.getInputs().size());
- break;
- case EXCEPT:
- assert setOp.getInputs().size() == 2;
- relBuilder.minus(setOp.all);
- break;
- default:
- throw new AssertionError("unknown setOp " + setOp);
- }
+ relBuilder.union(true, setOp.getInputs().size());
return result(relBuilder.build(), mapping, setOp);
}
diff --git a/core/src/test/java/org/apache/calcite/test/InterpreterTest.java
b/core/src/test/java/org/apache/calcite/test/InterpreterTest.java
index 1409b02d4a..7470e21636 100644
--- a/core/src/test/java/org/apache/calcite/test/InterpreterTest.java
+++ b/core/src/test/java/org/apache/calcite/test/InterpreterTest.java
@@ -471,7 +471,7 @@ private static void assertRows(Interpreter interpreter,
+ "(select x, y from (values (1, 'a'), (2, 'b'), (2, 'b'), (3, 'c'))
as t(x, y))\n"
+ "except all\n"
+ "(select x, y from (values (1, 'a'), (2, 'c'), (4, 'x')) as t2(x,
y))";
- sql(sql).returnsRows("[2, b]", "[2, b]", "[3, c]");
+ sql(sql).returnsRowsUnordered("[2, b]", "[2, b]", "[3, c]");
}
@Test void testDuplicateRowInterpretMinusAll() {
diff --git a/core/src/test/java/org/apache/calcite/test/JdbcTest.java
b/core/src/test/java/org/apache/calcite/test/JdbcTest.java
index a3c004c441..dbeb4f7a46 100644
--- a/core/src/test/java/org/apache/calcite/test/JdbcTest.java
+++ b/core/src/test/java/org/apache/calcite/test/JdbcTest.java
@@ -8249,11 +8249,10 @@ void checkCalciteSchemaGetSubSchemaMap(boolean cache) {
+ " BindableTableScan(table=[[hr, emps]])\n"
+ " BindableProject(empid=[$0], deptno=[$1])\n"
+ " BindableTableScan(table=[[hr, emps]])")
- .returns(""
- + "empid=150; deptno=10\n"
- + "empid=100; deptno=10\n"
- + "empid=200; deptno=20\n"
- + "empid=110; deptno=10\n");
+ .returnsUnordered("empid=150; deptno=10",
+ "empid=100; deptno=10",
+ "empid=200; deptno=20",
+ "empid=110; deptno=10");
}
}
diff --git a/core/src/test/resources/sql/interpreter.iq
b/core/src/test/resources/sql/interpreter.iq
new file mode 100644
index 0000000000..1b33d63c21
--- /dev/null
+++ b/core/src/test/resources/sql/interpreter.iq
@@ -0,0 +1,171 @@
+# interpreter.iq - Queries executed by the interpreter
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to you 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.
+#
+# Setting 'bindable' makes the planner target BindableConvention. Operators
+# that have no native bindable implementation - including Minus, Intersect and
+# Union - are executed by the interpreter (org.apache.calcite.interpreter); see
+# Bindables and SetOpNode.
+!use blank
+!set outputformat mysql
+!set bindable true
+
+# A project and filter, to confirm queries run via the interpreter.
+select i, i * 10 as j
+from (values 1, 2, 3, 4) as t (i)
+where i <> 2
+order by i;
++---+----+
+| I | J |
++---+----+
+| 1 | 10 |
+| 3 | 30 |
+| 4 | 40 |
++---+----+
+(3 rows)
+
+!ok
+
+# The plan uses bindable/interpreter operators, not enumerable ones.
+BindableProject(I=[$0], J=[*($0, 10)])
+ BindableFilter(condition=[<>($0, 2)])
+ BindableValues(tuples=[[{ 1 }, { 2 }, { 3 }, { 4 }]])
+!plan
+
+# Aggregate.
+select count(*) as c, sum(i) as s, min(i) as mn, max(i) as mx
+from (values 1, 2, 3, 4) as t (i);
++---+----+----+----+
+| C | S | MN | MX |
++---+----+----+----+
+| 4 | 10 | 1 | 4 |
++---+----+----+----+
+(1 row)
+
+!ok
+
+# Join.
+select t.i, u.y
+from (values (1, 'a'), (2, 'b'), (3, 'c')) as t (i, x)
+join (values (1, 'p'), (3, 'q')) as u (j, y) on t.i = u.j
+order by t.i;
++---+---+
+| I | Y |
++---+---+
+| 1 | p |
+| 3 | q |
++---+---+
+(2 rows)
+
+!ok
+
+# Multi-input set operations.
+#
+# Chained set operations are merged into a single n-input SetOp, so the
+# interpreter's SetOpNode must evaluate every input, not just the first two.
+# [CALCITE-7628] In the interpreter, MINUS with 3 or more inputs returns wrong
+# result
+
+# Except, three inputs
+select i from (
+ values 1, 2, 3 except values 3, 4, 5 except values 4, 5, 1
+) as t (i)
+order by i;
++---+
+| I |
++---+
+| 2 |
++---+
+(1 row)
+
+!ok
+
+# A single three-input BindableMinus confirms the interpreter path.
+BindableSort(sort0=[$0], dir0=[ASC])
+ BindableMinus(all=[false])
+ BindableValues(tuples=[[{ 1 }, { 2 }, { 3 }]])
+ BindableValues(tuples=[[{ 3 }, { 4 }, { 5 }]])
+ BindableValues(tuples=[[{ 4 }, { 5 }, { 1 }]])
+!plan
+
+# Except all, three inputs
+select i from (
+ values 1, 1, 1, 2, 2, 3 except all values 1 except all values 1, 2
+) as t (i)
+order by i;
++---+
+| I |
++---+
+| 1 |
+| 2 |
+| 3 |
++---+
+(3 rows)
+
+!ok
+
+# Intersect, three inputs
+select i from (
+ values 1, 2, 3, 4 intersect values 2, 3, 4, 5 intersect values 3, 4, 5, 6
+) as t (i)
+order by i;
++---+
+| I |
++---+
+| 3 |
+| 4 |
++---+
+(2 rows)
+
+!ok
+
+# Intersect all, three inputs
+select i from (
+ values 1, 1, 2, 2, 3 intersect all values 1, 2, 2, 3, 3
+ intersect all values 1, 2, 3, 3
+) as t (i)
+order by i;
++---+
+| I |
++---+
+| 1 |
+| 2 |
+| 3 |
++---+
+(3 rows)
+
+!ok
+
+# Union, three inputs
+select i from (
+ values 1, 2 union values 2, 3 union values 3, 4
+) as t (i)
+order by i;
++---+
+| I |
++---+
+| 1 |
+| 2 |
+| 3 |
+| 4 |
++---+
+(4 rows)
+
+!ok
+
+!set bindable false
+
+# End interpreter.iq
