kosiew commented on code in PR #25767:
URL: https://github.com/apache/datafusion/pull/25767#discussion_r4282906263
##########
datafusion/sqllogictest/test_files/intersection.slt:
##########
@@ -42,3 +42,150 @@ query IR
SELECT int_col, double_col FROM alltypes_plain where int_col > 0 INTERSECT
SELECT int_col, double_col FROM alltypes_plain
----
1 10.1
+
+# INTERSECT ALL keeps min(m, n) copies of each row and EXCEPT ALL keeps
+# max(m - n, 0), with NULLs compared as equal.
+statement ok
Review Comment:
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.
##########
datafusion/substrait/tests/cases/roundtrip_logical_plan.rs:
##########
@@ -1803,46 +1803,37 @@ async fn self_referential_except() -> Result<()> {
.await
}
+// `INTERSECT ALL` and `EXCEPT ALL` are planned with `unnest(range(..))` to
+// repeat each row, and the producer cannot serialize `Unnest` yet, so these
+// plans cannot be sent to Substrait. Consuming Substrait multiset set
+// operations still works (see `multiset_intersect_all_consume`).
#[tokio::test]
async fn self_referential_intersect_all() -> Result<()> {
- // Test INTERSECT ALL with the same table on both sides
- // INTERSECT ALL preserves duplicates and does not include DISTINCT
- // Uses **LeftSemi** join (returns rows from left that exist in right)
- // The requalification ensures no duplicate field name errors
- assert_expected_plan(
+ assert_unnest_not_supported(
Review Comment:
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?
##########
datafusion/expr/src/logical_plan/builder.rs:
##########
@@ -1497,6 +1510,184 @@ impl LogicalPlanBuilder {
)
}
+ /// Build an `INTERSECT ALL` plan, preserving the multiplicity of each row.
+ ///
+ /// Each distinct row is returned `min(m, n)` times, where `m` and `n` are
+ /// its counts on the left and right. `count` must be the `count` aggregate
+ /// function and `range` the `range` scalar function: rows are counted on
+ /// each side and then expanded back with `unnest(range(..))`.
+ pub fn intersect_all(
+ left_plan: LogicalPlan,
+ right_plan: LogicalPlan,
+ count: &Arc<AggregateUDF>,
+ range: &Arc<ScalarUDF>,
+ ) -> Result<LogicalPlan> {
+ Self::multiset_set_operation(left_plan, right_plan, count, range,
JoinType::Inner)
+ }
+
+ /// Build an `EXCEPT ALL` plan, subtracting matching row multiplicities.
+ ///
+ /// Each distinct row is returned `m - n` times when `m > n`. `count` and
+ /// `range` are as for [`Self::intersect_all`].
+ pub fn except_all(
+ left_plan: LogicalPlan,
+ right_plan: LogicalPlan,
+ count: &Arc<AggregateUDF>,
+ range: &Arc<ScalarUDF>,
+ ) -> Result<LogicalPlan> {
+ Self::multiset_set_operation(left_plan, right_plan, count, range,
JoinType::Left)
+ }
+
+ fn multiset_set_operation(
+ left_plan: LogicalPlan,
+ right_plan: LogicalPlan,
+ count: &Arc<AggregateUDF>,
+ range: &Arc<ScalarUDF>,
+ join_type: JoinType,
+ ) -> Result<LogicalPlan> {
+ let left_columns = left_plan.schema().columns();
+ let right_columns = right_plan.schema().columns();
+ if left_columns.len() != right_columns.len() {
+ return plan_err!(
+ "INTERSECT/EXCEPT query must have the same number of columns.
Left is {} and right is {}.",
+ left_columns.len(),
+ right_columns.len()
+ );
+ }
+ let from_right = left_plan
+ .schema()
+ .fields()
+ .iter()
+ .zip(right_plan.schema().fields())
+ .map(|(left, right)| {
+ join_type == JoinType::Inner
+ && left.is_nullable()
+ && !right.is_nullable()
+ && left.data_type() == right.data_type()
+ })
+ .collect::<Vec<_>>();
+ // The synthetic columns get names that differ from each other and from
+ // every input column, so the sides are only requalified when the
+ // user's own columns conflict.
+ let mut base = "__datafusion_set_operation".to_string();
+ let name_in_use = |name: &str| {
+ [left_plan.schema(), right_plan.schema()]
+ .iter()
+ .any(|schema| schema.fields().iter().any(|field| field.name()
== name))
+ };
+ let suffixes = ["_left_count", "_right_count", "_copies"];
+ while suffixes
+ .iter()
+ .any(|suffix| name_in_use(&format!("{base}{suffix}")))
+ {
+ base.push('_');
+ }
+ let [left_count, right_count, copies] =
+ suffixes.map(|suffix| format!("{base}{suffix}"));
+ // Count each distinct row on both sides. Grouping treats NULLs, and
+ // 0.0 and -0.0, as equal, like the rest of the set operations.
+ let count_rows = |plan: LogicalPlan, columns: &[Column], name: &str| {
+ let count_star =
Expr::AggregateFunction(AggregateFunction::new_udf(
+ Arc::clone(count),
+ vec![lit(COUNT_STAR_EXPANSION)],
+ false,
+ None,
+ vec![],
+ None,
+ ));
+ LogicalPlanBuilder::from(plan)
+ .aggregate(
+ columns.iter().cloned().map(Expr::Column),
+ vec![count_star.alias(name)],
+ )?
+ .build()
+ };
+ let left_plan = count_rows(left_plan, &left_columns, &left_count)?;
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
