You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "dariocurr (via GitHub)" <gi...@apache.org> on 2026/08/05 06:40:37 UTC
[PR] physical-plan: coerce UNION/INTERLEAVE schema mismatches at plan time [datafusion]
dariocurr opened a new pull request, #24094:
URL: https://github.com/apache/datafusion/pull/24094
## Which issue does this PR close?
Follow-up to #23861 (issue #15394). Not closing a new issue.
## Rationale for this change
#23861 fixed `UNION ALL` batches carrying the wrong nullability when one leg
is `NOT NULL` and another isn't, by re-stamping each batch's schema inside
`UnionExec`/`InterleaveExec`'s own `execute()`. In review, @alamb noted:
> ideally we could coerce the schema at plan time but I don't know how to
> coerce nullability
This PR does that: the coercion becomes an explicit node in the plan tree,
inserted when the plan is built, instead of invisible logic inside
`execute()`.
## What changes are included in this PR?
- Adds `CoerceSchemaExec`, a single-child passthrough `ExecutionPlan` node.
`UnionExec::try_new`/`InterleaveExec::try_new` insert it above any child
whose own output schema disagrees with the computed union schema (in
practice, only nullability differs -- `UnionExec::try_new` already rejects
real data-type mismatches via `calculate_union`).
- The actual batch re-stamping logic (`SchemaConformingStream`) is unchanged;
it just lives under `CoerceSchemaExec::execute()` now instead of being
called directly from `UnionExec`/`InterleaveExec::execute()`.
- Because it's a real plan node, `CoerceSchemaExec` implements the full
`ExecutionPlan` surface a pure 1:1 passthrough needs to stay transparent
to the optimizer: statistics passthrough, filter/limit pushdown,
`benefits_from_input_partitioning() -> false` (so it doesn't trigger a
spurious repartition), and proto (de)serialization -- the node erases
itself on encode and is reconstructed by `try_new` on decode, so no
protobuf schema change was needed.
- A genuine data-type mismatch (as opposed to nullability-only) is now
rejected eagerly at plan-build time (via `EquivalenceProperties::
with_new_schema`) rather than lazily at `execute()`.
## Are these changes tested?
- New unit test `test_union_partition_statistics_with_mismatched_nullability`
in `union.rs`, proving statistics aren't poisoned to `Absent` through the
new node.
- Existing `union_nullable`/`union_nullable_spill` regression tests from
#23861 continue to pass unchanged.
- Updated the `sqllogictest` golden file (`union.slt`) where `EXPLAIN`
output now shows the new node for pre-existing nullability-mismatched
`UNION ALL` cases.
- Benchmarked against the previous (inline) approach: no measurable
performance difference in either the coerced or matched-schema case
(differences were within run-to-run noise).
## Are there any user-facing changes?
`EXPLAIN` output for a `UNION ALL`/interleaved plan with a nullability
mismatch across legs will now show a `CoerceSchemaExec` node that wasn't
there before. No behavioral or correctness change.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] physical-plan: coerce UNION/INTERLEAVE schema mismatches at plan time [datafusion]
Posted by "dariocurr (via GitHub)" <gi...@apache.org>.
dariocurr commented on code in PR #24094:
URL: https://github.com/apache/datafusion/pull/24094#discussion_r3720055185
##########
datafusion/physical-plan/src/union.rs:
##########
@@ -141,6 +135,179 @@ fn conform_stream_schema(
}
}
+/// Coerces a single child's declared output schema to `schema`, re-stamping
+/// every batch it produces to match. [`UnionExec::try_new`] and
+/// [`InterleaveExec::try_new`] insert this above any child whose own schema
+/// disagrees with the computed union schema, so that the coercion is visible
+/// in the plan tree (e.g. in `EXPLAIN`) instead of happening invisibly inside
+/// the union operator's own `execute()`.
+///
+/// A genuine data type mismatch (as opposed to a nullability-only one) is
+/// rejected eagerly, at construction time, by `EquivalenceProperties::
+/// with_new_schema` below -- unlike the old purely-runtime approach, a
+/// hand-built union/interleave with mismatched child types now fails in
+/// `try_new` rather than in `execute()`.
+///
+/// This node is a strict 1:1, order-preserving passthrough of `input` (it
+/// only ever changes a batch's declared schema, never its rows), so every
+/// `ExecutionPlan` method below that isn't about the schema itself just
+/// delegates straight to `input`.
+///
+/// See <https://github.com/apache/datafusion/issues/15394>.
+#[derive(Debug)]
+struct CoerceSchemaExec {
+ input: Arc<dyn ExecutionPlan>,
+ cache: Arc<PlanProperties>,
+ metrics: ExecutionPlanMetricsSet,
+}
+
+impl CoerceSchemaExec {
+ /// Wraps `input` in a [`CoerceSchemaExec`] targeting `schema` if its own
+ /// schema disagrees with `schema`, otherwise returns it unchanged.
+ fn wrap_if_needed(
+ input: Arc<dyn ExecutionPlan>,
+ schema: &SchemaRef,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ if &input.schema() == schema {
+ Ok(input)
+ } else {
+ Ok(Arc::new(Self::new(input, schema)?))
+ }
+ }
+
+ fn new(input: Arc<dyn ExecutionPlan>, schema: &SchemaRef) -> Result<Self> {
+ let eq_properties = input
+ .equivalence_properties()
+ .clone()
+ .with_new_schema(Arc::clone(schema))?;
+ let output_partitioning = input.output_partitioning().clone();
+ let cache = PlanProperties::new(
+ eq_properties,
+ output_partitioning,
+ emission_type_from_children(std::iter::once(&input)),
+ boundedness_from_children(std::iter::once(&input)),
+ );
+ Ok(Self {
+ input,
+ cache: Arc::new(cache),
+ metrics: ExecutionPlanMetricsSet::new(),
+ })
+ }
+}
+
+impl DisplayAs for CoerceSchemaExec {
+ fn fmt_as(
+ &self,
+ t: DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ write!(f, "CoerceSchemaExec")
+ }
+ DisplayFormatType::TreeRender => Ok(()),
+ }
+ }
+}
+
+impl ExecutionPlan for CoerceSchemaExec {
+ fn name(&self) -> &'static str {
+ "CoerceSchemaExec"
+ }
+
+ fn properties(&self) -> &Arc<PlanProperties> {
+ &self.cache
+ }
+
+ fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+ vec![&self.input]
+ }
+
+ fn maintains_input_order(&self) -> Vec<bool> {
+ vec![true]
+ }
+
+ // A 1:1 passthrough never combines partitions, so re-deriving the cache
+ // (rather than collapsing back to the raw child via `wrap_if_needed`) is
+ // always safe here -- it just keeps this node from vanishing and
+ // changing arity out from under a caller mid-rewrite.
+ fn with_new_children(
+ self: Arc<Self>,
+ mut children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ assert_or_internal_err!(
+ children.len() == 1,
+ "CoerceSchemaExec expects exactly one child"
+ );
+ Ok(Arc::new(Self::new(children.remove(0), &self.schema())?))
+ }
+
+ fn execute(
+ &self,
+ partition: usize,
+ context: Arc<TaskContext>,
+ ) -> Result<SendableRecordBatchStream> {
+ let baseline_metrics = BaselineMetrics::new(&self.metrics, partition);
+ let stream = self.input.execute(partition, context)?;
+ let stream = conform_stream_schema(self.schema(), stream);
+ Ok(Box::pin(ObservedStream::new(
+ stream,
+ baseline_metrics,
+ None,
+ )))
+ }
+
+ fn metrics(&self) -> Option<MetricsSet> {
+ Some(self.metrics.clone_inner())
+ }
+
+ fn benefits_from_input_partitioning(&self) -> Vec<bool> {
+ vec![false]
+ }
+
+ fn supports_limit_pushdown(&self) -> bool {
+ true
+ }
+
+ fn cardinality_effect(&self) -> CardinalityEffect {
+ CardinalityEffect::Equal
+ }
+
+ fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+ vec![ChildStats::At(partition)]
+ }
+
+ fn statistics_from_inputs(
+ &self,
+ input_stats: &[Arc<Statistics>],
+ _args: &StatisticsArgs,
+ ) -> Result<Arc<Statistics>> {
+ Ok(Arc::clone(&input_stats[0]))
+ }
+
+ fn gather_filters_for_pushdown(
+ &self,
+ _phase: FilterPushdownPhase,
+ parent_filters: Vec<Arc<dyn PhysicalExpr>>,
+ _config: &ConfigOptions,
+ ) -> Result<FilterDescription> {
+ FilterDescription::from_children(parent_filters, &self.children())
+ }
+
+ #[cfg(feature = "proto")]
Review Comment:
Added `roundtrip_union_with_mismatched_nullability_executes` and `roundtrip_interleave_with_mismatched_nullability_executes` — one non-nullable + one nullable literal leg, serialize/deserialize, confirm `CoerceSchemaExec` reappears in the decoded plan, then execute it and check every emitted batch carries the nullable schema.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] physical-plan: coerce UNION/INTERLEAVE schema mismatches at plan time [datafusion]
Posted by "dariocurr (via GitHub)" <gi...@apache.org>.
dariocurr commented on PR #24094:
URL: https://github.com/apache/datafusion/pull/24094#issuecomment-5204381608
Made this change. `CoerceSchemaExec` is gone. `UnionExec`/`InterleaveExec` now insert a `ProjectionExec` with `CastExpr::new_with_target_field` (zero-copy for same-type casts), or a plain `Column` when a leg's field already matches.
A few side effects:
- No more wrapper-erasure trick in proto serialization — `ProjectionExec` has an ordinary proto message.
- The projection-collapsing optimizer pass fuses the coercion into a leg's existing top-level `ProjectionExec` for free in several cases.
- Fixed a real gap this surfaced: `ProjectionExec` statistics were zeroing `sum_value`/`byte_size` for same-type casts. Now passed through exactly, with a test.
Net 119 fewer lines despite added tests. Happy to split the statistics fix into its own PR if you'd rather review it separately.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] physical-plan: coerce UNION/INTERLEAVE schema mismatches at plan time [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #24094:
URL: https://github.com/apache/datafusion/pull/24094
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] physical-plan: coerce UNION/INTERLEAVE schema mismatches at plan time [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #24094:
URL: https://github.com/apache/datafusion/pull/24094#issuecomment-5193439470
> Thanks for working on this. The plan-visible CoerceSchemaExec approach looks good, and I like that it removes the schema restamping from UnionExec::execute() and InterleaveExec::execute() while keeping the coercion explicit in the plan.
I wonder if instead of using an entirely new exec, we could use the existing ProjectionExec operator with Cast exprs? In theory it should already be capable of doing the schema tranformation (assuming the physical CastExpr can have a field -- not just a DataType)
Thank you for follow up on this @dariocurr and @kosiew
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] physical-plan: coerce UNION/INTERLEAVE schema mismatches at plan time [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #24094:
URL: https://github.com/apache/datafusion/pull/24094#issuecomment-5190000175
<!-- semver-check-comment -->
Thank you for opening this pull request!
Reviewer note: [cargo-semver-checks](https://github.com/obi1kenobi/cargo-semver-checks) reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).
<details>
<summary>Details</summary>
```
Cloning apache/main
Building datafusion v54.1.0 (current)
Built [ 102.898s] (current)
Parsing datafusion v54.1.0 (current)
Parsed [ 0.037s] (current)
Building datafusion v54.1.0 (baseline)
Built [ 102.026s] (baseline)
Parsing datafusion v54.1.0 (baseline)
Parsed [ 0.034s] (baseline)
Checking datafusion v54.1.0 -> v54.1.0 (no change; assume patch)
Checked [ 0.611s] 223 checks: 223 pass, 30 skip
Summary no semver update required
Finished [ 207.161s] datafusion
Building datafusion-physical-plan v54.1.0 (current)
Built [ 37.182s] (current)
Parsing datafusion-physical-plan v54.1.0 (current)
Parsed [ 0.141s] (current)
Building datafusion-physical-plan v54.1.0 (baseline)
Built [ 37.122s] (baseline)
Parsing datafusion-physical-plan v54.1.0 (baseline)
Parsed [ 0.140s] (baseline)
Checking datafusion-physical-plan v54.1.0 -> v54.1.0 (no change; assume patch)
Checked [ 0.601s] 223 checks: 222 pass, 1 fail, 0 warn, 30 skip
--- failure inherent_method_missing: pub method removed or renamed ---
Description:
A publicly-visible method or associated fn is no longer available under its prior name. It may have been renamed or removed entirely.
ref: https://doc.rust-lang.org/cargo/reference/semver.html#item-remove
impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.49.0/src/lints/inherent_method_missing.ron
Failed in:
ProjectionExec::try_new_with_schema_metadata, previously in file /home/runner/work/datafusion/datafusion/target/semver-checks/git-apache_main/6e20036b1c30db2252c8977d875440dccfc89118/datafusion/physical-plan/src/projection.rs:157
Summary semver requires new major version: 1 major and 0 minor checks failed
Finished [ 76.251s] datafusion-physical-plan
Building datafusion-sqllogictest v54.1.0 (current)
Built [ 179.157s] (current)
Parsing datafusion-sqllogictest v54.1.0 (current)
Parsed [ 0.020s] (current)
Building datafusion-sqllogictest v54.1.0 (baseline)
Built [ 178.499s] (baseline)
Parsing datafusion-sqllogictest v54.1.0 (baseline)
Parsed [ 0.022s] (baseline)
Checking datafusion-sqllogictest v54.1.0 -> v54.1.0 (no change; assume patch)
Checked [ 0.083s] 223 checks: 223 pass, 30 skip
Summary no semver update required
Finished [ 360.261s] datafusion-sqllogictest
```
</details>
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] physical-plan: coerce UNION/INTERLEAVE schema mismatches at plan time [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #24094:
URL: https://github.com/apache/datafusion/pull/24094#issuecomment-5246107486
I merged up to resolve some conflicts
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] physical-plan: coerce UNION/INTERLEAVE schema mismatches at plan time [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #24094:
URL: https://github.com/apache/datafusion/pull/24094#issuecomment-5190120948
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/24094?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `87.31343%` with `17 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 80.90%. Comparing base ([`31ffab1`](https://app.codecov.io/gh/apache/datafusion/commit/31ffab1418aae8966dd270b1042c386bc9646950?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`8c2409c`](https://app.codecov.io/gh/apache/datafusion/commit/8c2409cd6f4002f6fd2d85ac73f38b08e1ab8586?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 4 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/24094?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [datafusion/physical-plan/src/union.rs](https://app.codecov.io/gh/apache/datafusion/pull/24094?src=pr&el=tree&filepath=datafusion%2Fphysical-plan%2Fsrc%2Funion.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9waHlzaWNhbC1wbGFuL3NyYy91bmlvbi5ycw==) | 87.31% | [7 Missing and 10 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/24094?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #24094 +/- ##
==========================================
- Coverage 80.90% 80.90% -0.01%
==========================================
Files 1102 1102
Lines 376807 377041 +234
Branches 376807 377041 +234
==========================================
+ Hits 304874 305057 +183
- Misses 53775 53783 +8
- Partials 18158 18201 +43
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/24094?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] physical-plan: coerce UNION/INTERLEAVE schema mismatches at plan time [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #24094:
URL: https://github.com/apache/datafusion/pull/24094#discussion_r3719961769
##########
datafusion/physical-plan/src/union.rs:
##########
@@ -141,6 +135,179 @@ fn conform_stream_schema(
}
}
+/// Coerces a single child's declared output schema to `schema`, re-stamping
+/// every batch it produces to match. [`UnionExec::try_new`] and
+/// [`InterleaveExec::try_new`] insert this above any child whose own schema
+/// disagrees with the computed union schema, so that the coercion is visible
+/// in the plan tree (e.g. in `EXPLAIN`) instead of happening invisibly inside
+/// the union operator's own `execute()`.
+///
+/// A genuine data type mismatch (as opposed to a nullability-only one) is
+/// rejected eagerly, at construction time, by `EquivalenceProperties::
+/// with_new_schema` below -- unlike the old purely-runtime approach, a
+/// hand-built union/interleave with mismatched child types now fails in
+/// `try_new` rather than in `execute()`.
+///
+/// This node is a strict 1:1, order-preserving passthrough of `input` (it
+/// only ever changes a batch's declared schema, never its rows), so every
+/// `ExecutionPlan` method below that isn't about the schema itself just
+/// delegates straight to `input`.
+///
+/// See <https://github.com/apache/datafusion/issues/15394>.
+#[derive(Debug)]
+struct CoerceSchemaExec {
+ input: Arc<dyn ExecutionPlan>,
+ cache: Arc<PlanProperties>,
+ metrics: ExecutionPlanMetricsSet,
+}
+
+impl CoerceSchemaExec {
+ /// Wraps `input` in a [`CoerceSchemaExec`] targeting `schema` if its own
+ /// schema disagrees with `schema`, otherwise returns it unchanged.
+ fn wrap_if_needed(
+ input: Arc<dyn ExecutionPlan>,
+ schema: &SchemaRef,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ if &input.schema() == schema {
+ Ok(input)
+ } else {
+ Ok(Arc::new(Self::new(input, schema)?))
+ }
+ }
+
+ fn new(input: Arc<dyn ExecutionPlan>, schema: &SchemaRef) -> Result<Self> {
+ let eq_properties = input
+ .equivalence_properties()
+ .clone()
+ .with_new_schema(Arc::clone(schema))?;
+ let output_partitioning = input.output_partitioning().clone();
+ let cache = PlanProperties::new(
+ eq_properties,
+ output_partitioning,
+ emission_type_from_children(std::iter::once(&input)),
+ boundedness_from_children(std::iter::once(&input)),
+ );
+ Ok(Self {
+ input,
+ cache: Arc::new(cache),
+ metrics: ExecutionPlanMetricsSet::new(),
+ })
+ }
+}
+
+impl DisplayAs for CoerceSchemaExec {
+ fn fmt_as(
+ &self,
+ t: DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ match t {
+ DisplayFormatType::Default | DisplayFormatType::Verbose => {
+ write!(f, "CoerceSchemaExec")
+ }
+ DisplayFormatType::TreeRender => Ok(()),
+ }
+ }
+}
+
+impl ExecutionPlan for CoerceSchemaExec {
+ fn name(&self) -> &'static str {
+ "CoerceSchemaExec"
+ }
+
+ fn properties(&self) -> &Arc<PlanProperties> {
+ &self.cache
+ }
+
+ fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+ vec![&self.input]
+ }
+
+ fn maintains_input_order(&self) -> Vec<bool> {
+ vec![true]
+ }
+
+ // A 1:1 passthrough never combines partitions, so re-deriving the cache
+ // (rather than collapsing back to the raw child via `wrap_if_needed`) is
+ // always safe here -- it just keeps this node from vanishing and
+ // changing arity out from under a caller mid-rewrite.
+ fn with_new_children(
+ self: Arc<Self>,
+ mut children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ assert_or_internal_err!(
+ children.len() == 1,
+ "CoerceSchemaExec expects exactly one child"
+ );
+ Ok(Arc::new(Self::new(children.remove(0), &self.schema())?))
+ }
+
+ fn execute(
+ &self,
+ partition: usize,
+ context: Arc<TaskContext>,
+ ) -> Result<SendableRecordBatchStream> {
+ let baseline_metrics = BaselineMetrics::new(&self.metrics, partition);
+ let stream = self.input.execute(partition, context)?;
+ let stream = conform_stream_schema(self.schema(), stream);
+ Ok(Box::pin(ObservedStream::new(
+ stream,
+ baseline_metrics,
+ None,
+ )))
+ }
+
+ fn metrics(&self) -> Option<MetricsSet> {
+ Some(self.metrics.clone_inner())
+ }
+
+ fn benefits_from_input_partitioning(&self) -> Vec<bool> {
+ vec![false]
+ }
+
+ fn supports_limit_pushdown(&self) -> bool {
+ true
+ }
+
+ fn cardinality_effect(&self) -> CardinalityEffect {
+ CardinalityEffect::Equal
+ }
+
+ fn child_stats_requests(&self, partition: Option<usize>) -> Vec<ChildStats> {
+ vec![ChildStats::At(partition)]
+ }
+
+ fn statistics_from_inputs(
+ &self,
+ input_stats: &[Arc<Statistics>],
+ _args: &StatisticsArgs,
+ ) -> Result<Arc<Statistics>> {
+ Ok(Arc::clone(&input_stats[0]))
+ }
+
+ fn gather_filters_for_pushdown(
+ &self,
+ _phase: FilterPushdownPhase,
+ parent_filters: Vec<Arc<dyn PhysicalExpr>>,
+ _config: &ConfigOptions,
+ ) -> Result<FilterDescription> {
+ FilterDescription::from_children(parent_filters, &self.children())
+ }
+
+ #[cfg(feature = "proto")]
Review Comment:
Could we add a protobuf round-trip regression test for nullable and non-nullable `UNION` and `INTERLEAVE` inputs?
`CoerceSchemaExec::try_to_proto` intentionally leaves the wrapper out of the serialized plan, while `UnionExec::try_from_proto` and `InterleaveExec::try_from_proto` rebuild it through `try_new`. It would be helpful to verify that the decoded plans contain the expected coercion and that their emitted batches expose the final nullable schema.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org