perf: speed up INTERSECT ALL and EXCEPT ALL on mostly distinct rows

[1fanwang]

Which issue does this PR close?

Related to #12955 and #25766. Depends on #25742.

Rationale for this change

INTERSECT ALL and EXCEPT ALL are about 20x slower than INTERSECT on mostly distinct rows with the duplicate-count fix, as kita-renji measured in the parent review. Grouped counts avoid the per-row window partitions.

Grouping each side before the join also miscounted rows when the input column types differ. The join compares groups in a common type, where distinct groups of one side can be equal, such as the strings '1' and '01' against the integer 1. EXCEPT ALL of the integers 1, 1 and the strings '1', '01' returned two rows instead of none.

What changes are included in this PR?

Each side groups equal rows and counts them. A NULL-safe join matches the counts, and range plus unnest emits min(m, n) copies for INTERSECT ALL or max(m - n, 0) for EXCEPT ALL. Grouping also treats signed zero as equal.

Columns whose types differ are first cast to the type equality compares them in, so values equal there count as one row, and the result has that type. Columns of the same type are untouched, and a timestamp with a time zone against one without is left to type coercion, which knows the session time zone.

What is the testing strategy for this PR?

SQL and DataFrame regressions cover duplicate counts, NULLs, column names and qualifiers. The intersection.slt and negative_zero.slt cases cover SQL execution, signed zero and empty inputs. Mixed-type inputs run through SQL and the DataFrame API in both operand orders, checked against the same operation on inputs cast to the common type first.

Testing Done

The mixed-type queries ran with datafusion-cli, dev profile:

datafusion-cli --quiet -f mixed.sql

Reproducer source: mixed.sql

CREATE TABLE ints(a BIGINT) AS VALUES (1), (1);
CREATE TABLE one_int(a BIGINT) AS VALUES (1);
CREATE TABLE padded(a VARCHAR) AS VALUES ('1'), ('01');
CREATE TABLE big_ints(a BIGINT) AS VALUES (9007199254740992), (9007199254740993);
CREATE TABLE a_float(a DOUBLE) AS VALUES (9007199254740992.0);
SELECT a, arrow_typeof(a) AS type FROM (SELECT a FROM ints EXCEPT ALL SELECT a FROM padded);
SELECT a, arrow_typeof(a) AS type FROM (SELECT a FROM one_int INTERSECT ALL SELECT a FROM padded);
SELECT a, arrow_typeof(a) AS type FROM (SELECT a FROM padded EXCEPT ALL SELECT a FROM one_int);
SELECT a, arrow_typeof(a) AS type FROM (SELECT a FROM big_ints EXCEPT ALL SELECT a FROM a_float);

The correct results are no rows, one row, one row and one row. The strings '1' and '01' are both the integer 1, and 9007199254740992 and 9007199254740993 are both 9007199254740992.0 as doubles. The output is:

+---+------+
| a | type |
+---+------+
+---+------+
+---+-------+
| a | type  |
+---+-------+
| 1 | Int64 |
+---+-------+
+---+-------+
| a | type  |
+---+-------+
| 1 | Int64 |
+---+-------+
+--------------------+---------+
| a                  | type    |
+--------------------+---------+
| 9007199254740992.0 | Float64 |
+--------------------+---------+

The benchmark below was measured before this change. Its queries use inputs of the same type, which are planned as before, and a builder unit test pins that equal types get no extra projection. Casting mismatched inputs adds one projection below the aggregate of each side that needs a cast, and I did not benchmark that cost.

The benchmark ran with datafusion-cli, release-nonlto, and 18 cores:

datafusion-cli -f bench.sql

Reproducer source: bench.sql

CREATE TABLE l AS SELECT value AS a, concat('v', value) AS b FROM generate_series(1, 10000000);
CREATE TABLE r AS SELECT value AS a, concat('v', value) AS b FROM generate_series(5000001, 15000000);
CREATE TABLE ld AS SELECT value % 1000 AS a, concat('v', value % 1000) AS b FROM generate_series(1, 10000000);
CREATE TABLE rd AS SELECT value % 1000 AS a, concat('v', value % 1000) AS b FROM generate_series(5000001, 15000000);
SELECT count(*) FROM (SELECT a, b FROM l INTERSECT ALL SELECT a, b FROM r) t;
SELECT count(*) FROM (SELECT a, b FROM l EXCEPT ALL SELECT a, b FROM r) t;
SELECT count(*) FROM (SELECT a, b FROM l INTERSECT SELECT a, b FROM r) t;
SELECT count(*) FROM (SELECT a, b FROM ld INTERSECT ALL SELECT a, b FROM rd) t;
SELECT count(*) FROM (SELECT a, b FROM ld EXCEPT ALL SELECT a, b FROM rd) t;
EXPLAIN ANALYZE SELECT count(*) FROM (SELECT a, b FROM l INTERSECT ALL SELECT a, b FROM r) t;

query                                   #25742 (84ef8f7)     this PR
l INTERSECT ALL r                       2.264s, 2.313s       0.157s, 0.122s
l EXCEPT ALL r                          2.496s, 2.387s       0.122s, 0.123s
l INTERSECT r                           0.085s               0.079s
ld INTERSECT ALL rd                     0.204s               0.021s
ld EXCEPT ALL rd                        0.214s               0.017s

Row counts match. EXPLAIN ANALYZE on 84ef8f7:

BoundedWindowAggExec: ..., mode=[Sorted], metrics=[output_rows=10.00 M, elapsed_compute=20.32s, ...]
BoundedWindowAggExec: ..., mode=[Sorted], metrics=[output_rows=10.00 M, elapsed_compute=20.17s, ...]

With this PR:

UnnestExec, metrics=[output_rows=5.00 M, elapsed_compute=37.73ms, ...]
  ProjectionExec: expr=[range(CASE WHEN __datafusion_set_operation_right_count@0 > __datafusion_set_operation_left_count@1 THEN __datafusion_set_operation_left_count@1 ELSE __datafusion_set_operation_right_count@0 END) as __datafusion_set_operation_copies], ...
    HashJoinExec: mode=Partitioned, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)], NullsEqual: true, metrics=[output_rows=5.00 M, elapsed_compute=909.84ms, ...]
      AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b], aggr=[count(1) as __datafusion_set_operation_left_count], metrics=[output_rows=10.00 M, elapsed_compute=439.33ms, ...]
      AggregateExec: mode=FinalPartitioned, gby=[a@0 as a, b@1 as b], aggr=[count(1) as __datafusion_set_operation_right_count], metrics=[output_rows=10.00 M, elapsed_compute=453.54ms, ...]

Signed zero:

SELECT a FROM (VALUES (0.0), (-0.0)) t(a) INTERSECT ALL SELECT a FROM (VALUES (0.0)) s(a);
-- 84ef8f7: -0.0, 0.0    this PR: 0.0
SELECT a FROM (VALUES (0.0), (-0.0)) t(a) EXCEPT ALL SELECT a FROM (VALUES (0.0)) s(a);
-- 84ef8f7: no rows      this PR: 0.0

Are there any user-facing changes?

INTERSECT ALL and EXCEPT ALL run faster. Custom sessions must register count and range, as documented in the upgrade guide. The unreleased LogicalPlanBuilder helpers take these functions instead of row_number.

With different input column types, the result has the type equality compares them in, not the left type, as UNION ALL does. INTERSECT and EXCEPT without ALL still return the left type.

Substrait consumption still works, but exporting these plans now fails with "Unsupported plan type: Unnest", as the self_referential_*_all tests assert. #26372 tracks supporting it.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 86.51316% with 41 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.85%. Comparing base (cf4869c) to head (4dc0cdd).
⚠️ Report is 11 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/expr/src/logical_plan/builder.rs	89.10%	6 Missing and 22 partials ⚠️
...substrait/src/logical_plan/consumer/rel/set_rel.rs	45.00%	2 Missing and 9 partials ⚠️
datafusion/core/src/dataframe/mod.rs	81.81%	0 Missing and 2 partials ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main   #25767    +/-   ##
========================================
  Coverage   82.84%   82.85%            
========================================
  Files        1151     1151            
  Lines      455961   456290   +329     
  Branches   455961   456290   +329     
========================================
+ Hits       377760   378064   +304     
+ Misses      55245    55240     -5     
- Partials    22956    22986    +30     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[kosiew]

@1fanwang,

Thanks for working on this. The grouped-count approach looks promising.

I found one correctness issue with mixed-type join keys that needs to be addressed before approval. I've also included two non-blocking suggestions about Substrait export compatibility and regression coverage.

[kosiew]

Both inputs are grouped before join-key coercion, so distinct values can become equal after casting and their counts are processed separately. For example, (VALUES (1), (1)) EXCEPT ALL (VALUES ('1'), ('01')) returns two rows instead of zero, and INTERSECT ALL can similarly overproduce rows. Could we coerce both inputs to their common comparison types before grouping, as UNION does, and add mixed-type regression tests for SQL and another API? This may change the left output type, so that behavior should be handled explicitly.

[1fanwang]

Done in bc0acb6. Both sides are now cast to one common type before grouping.

[kosiew]

The new Unnest-based plans cannot be exported to Substrait because the producer does not support LogicalPlan::Unnest. The previous export produced incorrect multiplicities, so returning an error is preferable, but the upgrade guide does not mention this compatibility change. Could we document the limitation in the 56.0.0 upgrade guide and link a follow-up issue for proper Substrait serialization support?

[1fanwang]

Done in 4dc0cdd. Tracked in #26372

[kosiew]

Could we add tests for an empty left input, an empty right input, and both inputs empty for INTERSECT ALL and EXCEPT ALL? In particular, EXCEPT ALL with an empty right input should retain duplicate left rows. The current behavior appears correct, but these tests would protect the new aggregation, outer-join, and expansion paths against regressions.

[1fanwang]

Done in 7bed929. Empty left, empty right and both empty are now covered.