kosiew commented on code in PR #25091:
URL: https://github.com/apache/datafusion/pull/25091#discussion_r4092559621
##########
datafusion/substrait/src/logical_plan/consumer/rel/set_rel.rs:
##########
@@ -77,16 +82,199 @@ 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 describes the left input's metadata exactly at the
+/// field level, and does so at the schema level for the plan this function
+/// returns, though not necessarily after later rewrites; see the two notes
+/// below.
+///
+/// At the field level: a column read from the right is cast to an explicit
+/// target field carrying the left field's metadata, and an explicit-field
+/// cast target replaces the source's metadata outright rather than merging
+/// into it, so a key only the right input's field carries is dropped rather
+/// than surviving into the result. A column read from the left is already
+/// exactly the left input's own field, so it needs no such treatment. This is
+/// robust: a `Field`'s own metadata travels with it through any later
+/// rewrite, since a rebuilt schema is still assembled from the same `Field`s.
+///
+/// At the schema level:
[`build_join_schema`](datafusion::logical_expr::build_join_schema)
+/// chains the two inputs' schema metadata maps (right's entries, then
+/// left's), so a conflicting key resolves to the left's value but a key
+/// present only on the right's schema survives, and [`projection_schema`]
+/// copies that merged map verbatim onto a projection built directly on top
+/// of the join. The projection this function returns is therefore built with
+/// an explicit schema instead, whose metadata is the left input's own schema
+/// metadata, unmodified - correct for the plan as returned here, and for a
+/// caller that reads its schema without executing it.
+///
+/// That schema-level fix is not preserved through arbitrary later rewrites,
+/// unlike the field-level one. `LogicalPlan::recompute_schema` rebuilds a
+/// `Join`'s or a `Projection`'s schema-level metadata from its (possibly
+/// rewritten) children's *current* schema-level metadata, discarding
+/// whatever this function set. DataFusion's optimizer calls it whenever it
+/// judges anything in the plan changed, including something unrelated
+/// elsewhere in the same query, so by the time this plan reaches physical
+/// planning its schema-level metadata may again be the join's merged one
+/// rather than the left input's. This holds even for the simplest two-input,
+/// non-correlated case, with no join filter or `INTERSECTION_PRIMARY` union
+/// involved. There is currently no supported way from this crate to opt a
+/// join's or a projection's schema-level metadata out of that
+/// recomputation.
+///
+/// [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 substituted for its own.
+ 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 {
+ // `alias_qualified_with_metadata` would not do:
`Expr::Alias`'s
+ // field derivation extends the aliased expression's own
+ // metadata with the alias's, so a key only the right column
+ // carries would survive alongside the left field's metadata.
+ // An explicit-field `Cast` target's metadata is instead used
+ // exactly as given, in both the logical and the physical
+ // plan, so casting to the left field's type (already proven
+ // equal to the right's) and metadata drops the right's own
+ // metadata outright. The qualifier and name still need
+ // `alias_qualified` on top, since a `Cast`'s own field is not
+ // renamed to its target field's name.
+ let target_field = Arc::new(
+ Field::new(&left.name, field.data_type().clone(), false)
+ .with_metadata(field.metadata().clone()),
+ );
+ Expr::Cast(Cast::new_from_field(
+ Box::new(Expr::Column(right.clone())),
+ target_field,
+ ))
+ .alias_qualified(left.relation.clone(), &left.name)
+ } else {
+ Expr::Column(left.clone())
+ }
+ })
+ .collect::<Vec<_>>();
+
+ // Captured before `left` is consumed by `join_detailed` below: this is
+ // the metadata the projection's schema must end up with, not the join's.
+ let left_schema_metadata = left.schema().metadata().clone();
Review Comment:
Could we simplify this by dropping the captured left schema metadata and the
explicit projection schema, then finish with
`LogicalPlanBuilder::from(joined).project(exprs)?.build()`? The explicit schema
only lasts until an optimizer rewrite, and the existing semi-join intersection
path already merges schema-level metadata. That would also let us remove the
unused imports and shorten the comment to explain that the explicit-field cast
preserves left field metadata, while the join merges schema metadata with the
left value winning conflicts. The test would need to change with it: check that
the `table` key is `data` instead of asserting exact schema metadata equality,
while keeping the exact per-field checks on converted and optimized plans. The
physical/batch comment could be shortened as well. This is optional, and I'm
sorry my earlier request led to the extra code.
--
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]
