kosiew commented on code in PR #25091:
URL: https://github.com/apache/datafusion/pull/25091#discussion_r4047456693
##########
datafusion/substrait/src/logical_plan/consumer/rel/set_rel.rs:
##########
@@ -77,16 +78,87 @@ 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".
+///
+/// [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 right_nullability: Vec<bool> = right
+ .schema()
+ .fields()
+ .iter()
+ .map(|field| field.is_nullable())
+ .collect();
+
+ let plan = LogicalPlanBuilder::intersect(left, right, is_all)?;
+
+ // `intersect` has already checked that both sides have the same width.
+ let narrowed: Vec<bool> = plan
+ .schema()
+ .fields()
+ .iter()
+ .zip(&right_nullability)
+ .map(|(field, right_nullable)| field.is_nullable() && !right_nullable)
+ .collect();
+
+ if !narrowed.contains(&true) {
+ return Ok(plan);
+ }
+
+ let qualified_fields = plan
+ .schema()
+ .iter()
+ .zip(&narrowed)
+ .map(|((qualifier, field), narrow)| {
+ let field = if *narrow {
+ Arc::new(field.as_ref().clone().with_nullable(false))
+ } else {
+ Arc::clone(field)
+ };
+ (qualifier.cloned(), field)
+ })
+ .collect();
+ let schema = Arc::new(DFSchema::new_with_metadata(
+ qualified_fields,
+ plan.schema().metadata().clone(),
+ )?);
+
+ let exprs = plan
+ .schema()
+ .columns()
+ .into_iter()
+ .map(Expr::Column)
+ .collect();
+ Ok(LogicalPlan::Projection(Projection::try_new_with_schema(
Review Comment:
I don't think this `Projection` can carry the nullability narrowing through
to execution. `DefaultPhysicalPlanner` creates it with
`ProjectionExec::try_new_with_schema_metadata`, and that API preserves
nullability from the physical column expressions and input. The supplied schema
is only used for metadata.
That means the left semi join can still expose nullable fields in the
physical plan and collected `RecordBatch` schema, while `plan.schema()` says
those fields are required. This leaves the logical and physical schemas
inconsistent.
Could we use an execution-plan or schema adapter that deliberately rebinds
the batches to the proven non-null schema while preserving the exact field
attributes, or extend the physical projection path so this validated narrowing
is supported there as well?
It would also be good to add a regression assertion that checks both the
physical-plan schema and the collected batch schema. The current `show()` call
exercises execution, but it does not verify that those schemas actually match
the narrowed logical schema.
##########
datafusion/substrait/tests/cases/logical_plans.rs:
##########
@@ -229,6 +229,50 @@ mod tests {
Ok(())
}
+ #[tokio::test]
+ async fn intersect_nullability() -> Result<()> {
+ // Substrait's set operation rules derive an intersection's
nullability from
+ // every input, not only the primary one. Each plan below intersects
three
+ // tables carrying the same four columns, with these nullabilities
+ // (`?` marks a nullable column):
+ //
+ // primary a? b? c? d?
+ // secondary a b c? d?
+ // secondary a b? c d?
+ for (file, expected) in [
Review Comment:
Could we also add a small `NULLABILITY_UNSPECIFIED` case, ideally on one of
the secondary inputs? Since the consumer treats unspecified nullability as
nullable, this would lock down that Substrait boundary in addition to the
explicit `REQUIRED` and `NULLABLE` cases.
--
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]
