You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/05/06 02:26:34 UTC
(datafusion) branch gh-readonly-queue/main/pr-21912-2c7af17529533cd2458b6b227a9128294e1fc209 created (now 739e1471b2)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-21912-2c7af17529533cd2458b6b227a9128294e1fc209
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 739e1471b2 Add reusable plan-time schema alignment helper and apply to RecursiveQueryExec (#21912)
This branch includes the following new commits:
new 739e1471b2 Add reusable plan-time schema alignment helper and apply to RecursiveQueryExec (#21912)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: Add reusable plan-time schema alignment helper and apply to RecursiveQueryExec (#21912)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-21912-2c7af17529533cd2458b6b227a9128294e1fc209
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 739e1471b262e9e3b2e26089d4678ad4c054c399
Author: kosiew <ko...@gmail.com>
AuthorDate: Wed May 6 10:26:10 2026 +0800
Add reusable plan-time schema alignment helper and apply to RecursiveQueryExec (#21912)
## Which issue does this PR close?
* Closes #21910.
---
## Rationale for this change
Physical plans in DataFusion can expose schemas that differ from their
declared output schema, particularly when combining independently
planned branches such as in recursive CTEs. This mismatch can lead to
inconsistencies observed by downstream operators or consumers that rely
on field names or schema metadata.
Previously, schema alignment for recursive queries was handled at the
`RecordBatch` level during execution, which can obscure contract
violations in the physical plan and make behavior harder to audit.
This change introduces a reusable execution-layer helper to align
schemas at plan construction time, ensuring that child plans conform to
the expected schema before execution.
---
## What changes are included in this PR?
* Introduce `project_plan_to_schema` in
`datafusion/physical-plan/src/common.rs`:
* Returns the input plan unchanged when schemas match.
* Applies a `ProjectionExec` to align field names when schemas are
positionally compatible.
* Validates column count, data types, nullability, and metadata before
applying projection.
* Produces clear errors when alignment is not possible.
* Update `RecursiveQueryExec`:
* Apply `project_plan_to_schema` to the recursive term during
construction.
* Remove batch-level schema rebinding logic from `RecursiveQueryStream`.
* Adjust tests and expected plans to reflect consistent field naming:
* Updated recursive CTE tests and explain output expectations.
---
## Are these changes tested?
Yes. The following tests are included:
* In `common.rs`:
* `project_plan_to_schema_returns_input_when_schema_matches`
* `project_plan_to_schema_aliases_field_names_with_projection_exec`
* `project_plan_to_schema_preserves_matching_metadata_while_renaming`
* `project_plan_to_schema_errors_on_column_count_mismatch`
* `project_plan_to_schema_errors_on_type_mismatch`
* `project_plan_to_schema_errors_on_nullability_mismatch`
* `project_plan_to_schema_errors_on_field_metadata_mismatch`
* `project_plan_to_schema_errors_on_schema_metadata_mismatch`
* In `recursive_query.rs`:
* `recursive_query_exec_projects_recursive_term_to_reconciled_schema`
* `recursive_query_exec_rejects_nullability_mismatch`
* Updates to existing sqllogictest cases in `cte.slt` and explain plan
expectations.
---
## Are there any user-facing changes?
No direct user-facing API changes are introduced.
However, physical plans for recursive queries now consistently expose
the declared schema at plan time, which may result in more consistent
field names in explain plans and downstream consumers.
---
## LLM-generated code disclosure
This PR includes LLM-generated code and comments. All LLM-generated
content has been manually reviewed and tested.
---------
Co-authored-by: Copilot <co...@github.com>
---
datafusion/core/tests/sql/explain_analyze.rs | 2 +-
datafusion/physical-plan/src/common.rs | 258 +++++++++++++++++++++++-
datafusion/physical-plan/src/recursive_query.rs | 101 ++++++++--
datafusion/sqllogictest/test_files/cte.slt | 6 +-
4 files changed, 340 insertions(+), 27 deletions(-)
diff --git a/datafusion/core/tests/sql/explain_analyze.rs b/datafusion/core/tests/sql/explain_analyze.rs
index 8ab0d150a7..b093563d9a 100644
--- a/datafusion/core/tests/sql/explain_analyze.rs
+++ b/datafusion/core/tests/sql/explain_analyze.rs
@@ -1019,7 +1019,7 @@ async fn parquet_recursive_projection_pushdown() -> Result<()> {
RepartitionExec: partitioning=RoundRobinBatch(NUM_CORES), input_partitions=1
DataSourceExec: file_groups={1 group: [[TMP_DIR/hierarchy.parquet]]}, projection=[id], file_type=parquet, predicate=id@0 = 1, pruning_predicate=id_null_count@2 != row_count@3 AND id_min@0 <= 1 AND 1 <= id_max@1, required_guarantees=[id in (1)]
CoalescePartitionsExec
- ProjectionExec: expr=[id@0 + 1 as ns.id + Int64(1), level@1 + 1 as ns.level + Int64(1)]
+ ProjectionExec: expr=[id@0 + 1 as id, level@1 + 1 as level]
FilterExec: id@0 < 10
RepartitionExec: partitioning=RoundRobinBatch(NUM_CORES), input_partitions=1
WorkTableExec: name=number_series
diff --git a/datafusion/physical-plan/src/common.rs b/datafusion/physical-plan/src/common.rs
index 5551685427..0dafcf6bd3 100644
--- a/datafusion/physical-plan/src/common.rs
+++ b/datafusion/physical-plan/src/common.rs
@@ -22,11 +22,13 @@ use std::fs::metadata;
use std::sync::Arc;
use super::SendableRecordBatchStream;
+use crate::expressions::{CastExpr, Column};
+use crate::projection::{ProjectionExec, ProjectionExpr};
use crate::stream::RecordBatchReceiverStream;
-use crate::{ColumnStatistics, Statistics};
+use crate::{ColumnStatistics, ExecutionPlan, Statistics};
use arrow::array::Array;
-use arrow::datatypes::Schema;
+use arrow::datatypes::{Schema, SchemaRef};
use arrow::record_batch::RecordBatch;
use datafusion_common::stats::Precision;
use datafusion_common::{Result, plan_err};
@@ -88,6 +90,96 @@ fn build_file_list_recurse(
Ok(())
}
+/// Align `input`'s physical plan schema with `expected_schema`.
+///
+/// This helper is intended for operators that combine independently planned children but
+/// expose a single declared output schema. It returns `input` unchanged when schemas already
+/// match exactly. Otherwise, it validates that projection can safely produce the expected
+/// schema, then wraps `input` in a [`ProjectionExec`] that keeps columns in their existing
+/// positional order and aliases them to `expected_schema`'s field names.
+///
+/// [`ProjectionExec`] can rename fields. When the expected field is nullable and the input
+/// field is not, this helper also widens nullability with a same-type [`CastExpr`]. It rejects
+/// differences that projection cannot safely normalize exactly, such as data type, metadata,
+/// schema metadata, and nullability narrowing.
+pub fn project_plan_to_schema(
+ input: Arc<dyn ExecutionPlan>,
+ expected_schema: &SchemaRef,
+) -> Result<Arc<dyn ExecutionPlan>> {
+ let input_schema = input.schema();
+ if input_schema.as_ref() == expected_schema.as_ref() {
+ return Ok(input);
+ }
+
+ if input_schema.fields().len() != expected_schema.fields().len() {
+ return plan_err!(
+ "Cannot project plan to expected schema: expected {} column(s), got {}",
+ expected_schema.fields().len(),
+ input_schema.fields().len()
+ );
+ }
+
+ if input_schema.metadata() != expected_schema.metadata() {
+ return plan_err!(
+ "Cannot project plan to expected schema: schema metadata differ"
+ );
+ }
+
+ if let Some((i, input_field, expected_field, mismatch)) = input_schema
+ .fields()
+ .iter()
+ .zip(expected_schema.fields().iter())
+ .enumerate()
+ .find_map(|(i, (input_field, expected_field))| {
+ if input_field.data_type() != expected_field.data_type() {
+ Some((i, input_field, expected_field, "data type"))
+ } else if input_field.is_nullable() && !expected_field.is_nullable() {
+ Some((i, input_field, expected_field, "nullability"))
+ } else if input_field.metadata() != expected_field.metadata() {
+ Some((i, input_field, expected_field, "metadata"))
+ } else {
+ None
+ }
+ })
+ {
+ return plan_err!(
+ "Cannot project plan column {i} ('{}') to expected output field '{}': \
+ field {mismatch} differs (input field: {:?}, expected field: {:?})",
+ input_field.name(),
+ expected_field.name(),
+ input_field,
+ expected_field
+ );
+ }
+
+ let projection_exprs = expected_schema
+ .fields()
+ .iter()
+ .enumerate()
+ .map(|(i, expected_field)| {
+ let input_field = input_schema.field(i);
+ let column = Arc::new(Column::new(input_field.name(), i));
+ let expr = if !input_field.is_nullable() && expected_field.is_nullable() {
+ Arc::new(CastExpr::new_with_target_field(
+ column,
+ Arc::clone(expected_field),
+ None,
+ )) as _
+ } else {
+ column as _
+ };
+ ProjectionExpr {
+ expr,
+ alias: expected_field.name().clone(),
+ }
+ })
+ .collect::<Vec<_>>();
+
+ let projection = ProjectionExec::try_new(projection_exprs, input)?;
+ debug_assert_eq!(projection.schema().as_ref(), expected_schema.as_ref());
+ Ok(Arc::new(projection))
+}
+
/// If running in a tokio context spawns the execution of `stream` to a separate task
/// allowing it to execute in parallel with an intermediate buffer of size `buffer`
pub fn spawn_buffered(
@@ -178,10 +270,7 @@ pub fn compute_record_batch_statistics(
}
/// Checks if the given projection is valid for the given schema.
-pub fn can_project(
- schema: &arrow::datatypes::SchemaRef,
- projection: Option<&[usize]>,
-) -> Result<()> {
+pub fn can_project(schema: &SchemaRef, projection: Option<&[usize]>) -> Result<()> {
match projection {
Some(columns) => {
if columns
@@ -206,12 +295,20 @@ pub fn can_project(
#[cfg(test)]
mod tests {
use super::*;
+ use crate::empty::EmptyExec;
+ use crate::projection::ProjectionExec;
+
+ use std::collections::HashMap;
use arrow::{
array::{Float32Array, Float64Array, UInt64Array},
- datatypes::{DataType, Field},
+ datatypes::{DataType, Field, Schema},
};
+ fn empty_exec(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(Schema::new(fields))))
+ }
+
#[test]
fn test_compute_record_batch_statistics_empty() -> Result<()> {
let schema = Arc::new(Schema::new(vec![
@@ -310,4 +407,151 @@ mod tests {
assert_eq!(actual, expected);
Ok(())
}
+
+ #[test]
+ fn project_plan_to_schema_returns_input_when_schema_matches() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "value",
+ DataType::Int32,
+ false,
+ )]));
+ let input: Arc<dyn ExecutionPlan> = Arc::new(EmptyExec::new(Arc::clone(&schema)));
+
+ let result = project_plan_to_schema(Arc::clone(&input), &schema)?;
+
+ assert!(Arc::ptr_eq(&input, &result));
+ Ok(())
+ }
+
+ #[test]
+ fn project_plan_to_schema_aliases_field_names_with_projection_exec() -> Result<()> {
+ let input = empty_exec(vec![
+ Field::new("recursive_a", DataType::Int32, false),
+ Field::new("recursive_b", DataType::Utf8, true),
+ ]);
+ let expected_schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, false),
+ Field::new("b", DataType::Utf8, true),
+ ]));
+
+ let result = project_plan_to_schema(Arc::clone(&input), &expected_schema)?;
+
+ let projection = result
+ .downcast_ref::<ProjectionExec>()
+ .expect("schema rename should use ProjectionExec");
+ assert!(Arc::ptr_eq(projection.input(), &input));
+ assert_eq!(projection.schema(), expected_schema);
+ assert_eq!(projection.expr()[0].alias, "a");
+ assert_eq!(projection.expr()[1].alias, "b");
+ Ok(())
+ }
+
+ #[test]
+ fn project_plan_to_schema_preserves_matching_metadata_while_renaming() -> Result<()> {
+ let field_metadata = HashMap::from([("key".to_string(), "value".to_string())]);
+ let schema_metadata =
+ HashMap::from([("schema-key".to_string(), "schema-value".to_string())]);
+ let input_schema = Arc::new(Schema::new_with_metadata(
+ vec![
+ Field::new("input", DataType::Int32, false)
+ .with_metadata(field_metadata.clone()),
+ ],
+ schema_metadata.clone(),
+ ));
+ let input: Arc<dyn ExecutionPlan> = Arc::new(EmptyExec::new(input_schema));
+ let expected_schema = Arc::new(Schema::new_with_metadata(
+ vec![
+ Field::new("expected", DataType::Int32, false)
+ .with_metadata(field_metadata),
+ ],
+ schema_metadata,
+ ));
+
+ let result = project_plan_to_schema(input, &expected_schema)?;
+
+ assert_eq!(result.schema(), expected_schema);
+ Ok(())
+ }
+
+ #[test]
+ fn project_plan_to_schema_errors_on_column_count_mismatch() {
+ let input = empty_exec(vec![Field::new("a", DataType::Int32, false)]);
+ let expected_schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, false),
+ Field::new("b", DataType::Int32, false),
+ ]));
+
+ let err = project_plan_to_schema(input, &expected_schema).unwrap_err();
+ assert!(err.to_string().contains("expected 2 column"));
+ }
+
+ #[test]
+ fn project_plan_to_schema_errors_on_type_mismatch() {
+ let input = empty_exec(vec![Field::new("a", DataType::Int32, false)]);
+ let expected_schema =
+ Arc::new(Schema::new(vec![Field::new("a", DataType::Float32, false)]));
+
+ let err = project_plan_to_schema(input, &expected_schema).unwrap_err();
+ assert!(err.to_string().contains("field data type differs"));
+ }
+
+ #[test]
+ fn project_plan_to_schema_widens_nullability() -> Result<()> {
+ let input = empty_exec(vec![Field::new("a", DataType::Int32, false)]);
+ let expected_schema = Arc::new(Schema::new(vec![Field::new(
+ "renamed",
+ DataType::Int32,
+ true,
+ )]));
+
+ let result = project_plan_to_schema(input, &expected_schema)?;
+
+ assert_eq!(result.schema(), expected_schema);
+ Ok(())
+ }
+
+ #[test]
+ fn project_plan_to_schema_errors_on_nullability_narrowing() {
+ let input = empty_exec(vec![Field::new("a", DataType::Int32, true)]);
+ let expected_schema = Arc::new(Schema::new(vec![Field::new(
+ "renamed",
+ DataType::Int32,
+ false,
+ )]));
+
+ let err = project_plan_to_schema(input, &expected_schema).unwrap_err();
+ assert!(err.to_string().contains("field nullability differs"));
+ }
+
+ #[test]
+ fn project_plan_to_schema_errors_on_field_metadata_mismatch() {
+ let input =
+ empty_exec(vec![Field::new("a", DataType::Int32, false).with_metadata(
+ HashMap::from([("source".to_string(), "input".to_string())]),
+ )]);
+ let expected_schema = Arc::new(Schema::new(vec![
+ Field::new("renamed", DataType::Int32, false).with_metadata(HashMap::from([
+ ("source".to_string(), "expected".to_string()),
+ ])),
+ ]));
+
+ let err = project_plan_to_schema(input, &expected_schema).unwrap_err();
+ assert!(err.to_string().contains("field metadata differs"));
+ }
+
+ #[test]
+ fn project_plan_to_schema_errors_on_schema_metadata_mismatch() {
+ let input_schema = Arc::new(Schema::new_with_metadata(
+ vec![Field::new("a", DataType::Int32, false)],
+ HashMap::from([("source".to_string(), "input".to_string())]),
+ ));
+ let input: Arc<dyn ExecutionPlan> = Arc::new(EmptyExec::new(input_schema));
+ let expected_schema = Arc::new(Schema::new_with_metadata(
+ vec![Field::new("renamed", DataType::Int32, false)],
+ HashMap::from([("source".to_string(), "expected".to_string())]),
+ ));
+
+ let err = project_plan_to_schema(input, &expected_schema).unwrap_err();
+ assert!(err.to_string().contains("schema metadata differ"));
+ }
}
diff --git a/datafusion/physical-plan/src/recursive_query.rs b/datafusion/physical-plan/src/recursive_query.rs
index 6eee5d1237..c160f9a0dc 100644
--- a/datafusion/physical-plan/src/recursive_query.rs
+++ b/datafusion/physical-plan/src/recursive_query.rs
@@ -24,6 +24,7 @@ use std::task::{Context, Poll};
use super::work_table::{ReservedBatches, WorkTable};
use crate::aggregates::group_values::{GroupValues, new_group_values};
use crate::aggregates::order::GroupOrdering;
+use crate::common::project_plan_to_schema;
use crate::execution_plan::{Boundedness, EmissionType, reset_plan_states};
use crate::metrics::{
BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet, RecordOutput,
@@ -34,7 +35,7 @@ use crate::{
};
use arrow::array::{BooleanArray, BooleanBuilder};
use arrow::compute::filter_record_batch;
-use arrow::datatypes::SchemaRef;
+use arrow::datatypes::{Field, Schema, SchemaRef};
use arrow::record_batch::RecordBatch;
use datafusion_common::tree_node::TreeNodeRecursion;
use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
@@ -90,8 +91,12 @@ impl RecursiveQueryExec {
// Each recursive query needs its own work table
let work_table = Arc::new(WorkTable::new(name.clone()));
// Use the same work table for both the WorkTableExec and the recursive term
+ let output_schema =
+ recursive_output_schema(&static_term.schema(), &recursive_term.schema());
+ let static_term = project_plan_to_schema(static_term, &output_schema)?;
let recursive_term = assign_work_table(recursive_term, &work_table)?;
- let cache = Self::compute_properties(static_term.schema());
+ let recursive_term = project_plan_to_schema(recursive_term, &output_schema)?;
+ let cache = Self::compute_properties(output_schema);
Ok(RecursiveQueryExec {
name,
static_term,
@@ -318,19 +323,6 @@ impl RecursiveQueryStream {
) -> Poll<Option<Result<RecordBatch>>> {
let baseline_metrics = self.baseline_metrics.clone();
- // Rebind to the declared output schema. The recursive term is planned
- // independently from the static term and its projection may leave
- // columns un-aliased (e.g. `upper(r.val)` vs the anchor's
- // `upper(val) AS val`); downstream consumers that key on
- // `batch.schema().field(i).name()` (TopK, CSV/JSON writers, custom
- // collectors) would otherwise see the recursive branch's names leak
- // through. Logical-plan coercion guarantees matching types, so this
- // is a zero-copy field rebind.
- if batch.schema() != self.schema {
- batch =
- RecordBatch::try_new(Arc::clone(&self.schema), batch.columns().to_vec())?;
- }
-
if let Some(deduplicator) = &mut self.distinct_deduplicator {
let _timer_guard = baseline_metrics.elapsed_compute().timer();
batch = deduplicator.deduplicate(&batch)?;
@@ -378,6 +370,30 @@ impl RecursiveQueryStream {
}
}
+fn recursive_output_schema(
+ static_schema: &SchemaRef,
+ recursive_schema: &SchemaRef,
+) -> SchemaRef {
+ let fields = static_schema
+ .fields()
+ .iter()
+ .zip(recursive_schema.fields())
+ .map(|(static_field, recursive_field)| {
+ Field::new(
+ static_field.name(),
+ static_field.data_type().clone(),
+ static_field.is_nullable() || recursive_field.is_nullable(),
+ )
+ .with_metadata(static_field.metadata().clone())
+ })
+ .collect::<Vec<_>>();
+
+ Arc::new(Schema::new_with_metadata(
+ fields,
+ static_schema.metadata().clone(),
+ ))
+}
+
fn assign_work_table(
plan: Arc<dyn ExecutionPlan>,
work_table: &Arc<WorkTable>,
@@ -502,4 +518,57 @@ fn new_groups_mask(
}
#[cfg(test)]
-mod tests {}
+mod tests {
+ use super::*;
+ use crate::empty::EmptyExec;
+ use crate::projection::ProjectionExec;
+
+ use arrow::datatypes::{DataType, Field, Schema};
+
+ fn empty_exec(fields: Vec<Field>) -> Arc<dyn ExecutionPlan> {
+ Arc::new(EmptyExec::new(Arc::new(Schema::new(fields))))
+ }
+
+ #[test]
+ fn recursive_query_exec_projects_recursive_term_to_reconciled_schema() -> Result<()> {
+ let static_term = empty_exec(vec![Field::new("value", DataType::Int32, false)]);
+ let recursive_term =
+ empty_exec(vec![Field::new("value + Int32(1)", DataType::Int32, false)]);
+
+ let exec = RecursiveQueryExec::try_new(
+ "numbers".to_string(),
+ Arc::clone(&static_term),
+ Arc::clone(&recursive_term),
+ false,
+ )?;
+
+ assert_eq!(exec.schema(), static_term.schema());
+ let projection = exec
+ .recursive_term()
+ .downcast_ref::<ProjectionExec>()
+ .expect("recursive term should be aligned with ProjectionExec");
+ assert!(Arc::ptr_eq(projection.input(), &recursive_term));
+ assert!(!projection.schema().field(0).is_nullable());
+ assert_eq!(projection.expr()[0].alias, "value");
+ Ok(())
+ }
+
+ #[test]
+ fn recursive_query_exec_reconciles_nullability() -> Result<()> {
+ let static_term = empty_exec(vec![Field::new("value", DataType::Int32, false)]);
+ let recursive_term =
+ empty_exec(vec![Field::new("value + Int32(1)", DataType::Int32, true)]);
+
+ let exec = RecursiveQueryExec::try_new(
+ "numbers".to_string(),
+ static_term,
+ recursive_term,
+ false,
+ )?;
+
+ assert!(exec.schema().field(0).is_nullable());
+ assert!(exec.static_term().schema().field(0).is_nullable());
+ assert!(exec.recursive_term().schema().field(0).is_nullable());
+ Ok(())
+ }
+}
diff --git a/datafusion/sqllogictest/test_files/cte.slt b/datafusion/sqllogictest/test_files/cte.slt
index e9c1c0245d..d13e0d4f08 100644
--- a/datafusion/sqllogictest/test_files/cte.slt
+++ b/datafusion/sqllogictest/test_files/cte.slt
@@ -699,7 +699,7 @@ WITH RECURSIVE region_sales AS (
SELECT
s.salesperson_id AS salesperson_id,
SUM(s.sale_amount) AS amount,
- 0 as level
+ SUM(0) as level
FROM
sales s
GROUP BY
@@ -1082,7 +1082,7 @@ physical_plan
02)--ProjectionExec: expr=[1 as n]
03)----PlaceholderRowExec
04)--CoalescePartitionsExec
-05)----ProjectionExec: expr=[n@0 + 1 as numbers.n + Int64(1)]
+05)----ProjectionExec: expr=[n@0 + 1 as n]
06)------FilterExec: n@0 < 10
07)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
08)----------WorkTableExec: name=numbers
@@ -1107,7 +1107,7 @@ physical_plan
02)--ProjectionExec: expr=[1 as n]
03)----PlaceholderRowExec
04)--CoalescePartitionsExec
-05)----ProjectionExec: expr=[n@0 + 1 as numbers.n + Int64(1)]
+05)----ProjectionExec: expr=[n@0 + 1 as n]
06)------FilterExec: n@0 < 10
07)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
08)----------WorkTableExec: name=numbers
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org