namanjain24-sudo commented on code in PR #25091:
URL: https://github.com/apache/datafusion/pull/25091#discussion_r4070486211
##########
datafusion/substrait/src/logical_plan/consumer/rel/set_rel.rs:
##########
@@ -77,16 +80,112 @@ async fn intersect_rels(
let mut rel = consumer.consume_rel(&rels[0]).await?;
for input in &rels[1..] {
- rel = LogicalPlanBuilder::intersect(
- rel,
- consumer.consume_rel(input).await?,
- is_all,
- )?;
+ rel = intersect_rel(rel, consumer.consume_rel(input).await?, is_all)?;
}
Ok(rel)
}
+/// Intersects two relations, giving the result the nullability the Substrait
+/// [Set Operation rules] prescribe.
+///
+/// [`LogicalPlanBuilder::intersect`] compiles an intersection into a left semi
+/// join, so on its own the result keeps the left input's nullability. The join
+/// matches nulls with nulls, so a left row holding a null in some field only
+/// survives when the right input holds a null there too. A field is therefore
+/// nullable in the result only when it is nullable in *both* inputs.
+///
+/// Applied to each step of a chain, that gives the spec's rule for the
multiset
+/// intersections - a field is required when any input requires it. For
+/// `INTERSECTION_PRIMARY` the right side is the union of the secondary inputs,
+/// whose field is nullable exactly when some secondary input makes it
nullable,
+/// so the same rule yields "nullable in the primary input and in at least one
+/// secondary input".
+///
+/// When the right input requires a field the left input leaves nullable, the
+/// intersection is built as an inner join against the distinct right rows
+/// instead, and that field is read from the right side. Matched rows hold
equal
+/// values, so the result is unchanged, and the field is non-nullable because
+/// its source is: the logical and the physical planner both derive that from
+/// the input schema, so the plan, the physical plan and the batches agree.
+/// Joining against distinct right rows keeps each left row at most once, as
the
+/// semi join does.
+///
+/// Differing metadata does not change which path is taken, as it is no part of
+/// nullability. The result should describe the left input, as
+/// [`LogicalPlanBuilder::intersect`] does, so on conflicting keys the left's
+/// metadata wins: a column read from the right is aliased with the left
field's
+/// metadata, and an inner join's schema already lets the left input's schema
+/// metadata win, in the logical and in the physical plan alike. Keys only the
+/// right input carries are merged in, and the physical plan and the batches
+/// carry the same metadata as the logical plan.
+///
+/// [Set Operation rules]:
https://substrait.io/relations/logical_relations/#set-operation
+fn intersect_rel(
+ left: LogicalPlan,
+ right: LogicalPlan,
+ is_all: bool,
+) -> datafusion::common::Result<LogicalPlan> {
+ let left_fields = left.schema().fields();
+ let right_fields = right.schema().fields();
+ // A field is read from the right side when the left leaves it nullable and
+ // the right requires it. Its metadata does not matter here: it is read
from
+ // the right with the left field's metadata layered over it.
+ let from_right: Vec<bool> = left_fields
+ .iter()
+ .zip(right_fields.iter())
+ .map(|(left, right)| {
+ left.is_nullable()
+ && !right.is_nullable()
+ && left.data_type() == right.data_type()
+ })
+ .collect();
+
+ // `intersect` also reports inputs of different widths.
+ if left_fields.len() != right_fields.len() || !from_right.contains(&true) {
+ return LogicalPlanBuilder::intersect(left, right, is_all);
+ }
+
+ let (left, right, _) = requalify_sides_if_needed(
+ LogicalPlanBuilder::from(left),
+ LogicalPlanBuilder::from(right),
+ )?;
+ let left = if is_all { left } else { left.distinct()? };
+ let right = right.distinct()?.build()?;
+
+ let left_columns = left.schema().columns();
+ let right_columns = right.schema().columns();
+ let exprs = left
+ .schema()
+ .fields()
+ .iter()
+ .zip(&left_columns)
+ .zip(&right_columns)
+ .zip(&from_right)
+ .map(|(((field, left), right), from_right)| {
+ if *from_right {
+ Expr::Column(right.clone()).alias_qualified_with_metadata(
Review Comment:
Good catch — you're right that `alias_qualified_with_metadata` merges rather
than replaces, since `Expr::Alias`'s field derivation extends the aliased
expression's own metadata with the alias's supplied metadata instead of
overriding it.
Fixed in 2a8d8f5: instead of aliasing the right column directly, I route it
through a `Cast` to an explicit target field (`Cast::new_from_field`) carrying
the left field's exact type and metadata. Both `cast_output_field` (logical)
and `cast_with_target_field`/`CastExpr::new_with_target_field` (physical) use
an explicit target field's metadata exactly, with nothing merged in from the
source - that's the documented, intended way to get "target metadata exactly"
rather than `Alias`'s merge semantics. The qualifier/name still need
`alias_qualified` on top since a bare `Cast`'s own field isn't renamed to the
target's name.
Extended the `intersect_nullability` regression test to compare the complete
metadata map (not just the `"column"` key) for the tagged case, so a leaked
`only_in_secondary` key now fails it. Since the test already asserts
`physical_plan.schema() == logical_schema` and `batch.schema() ==
logical_schema`, that same check covers the physical plan and collected batches
too.
--
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]
