1fanwang commented on code in PR #25767:
URL: https://github.com/apache/datafusion/pull/25767#discussion_r4284738466
##########
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:
Done in
https://github.com/apache/datafusion/commit/bc0acb698d2d80e0e3a52b2dbc6f5757b4e06bb8.
Both sides are now cast to one common type before grouping.
--
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]
