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-24087-373fab7bd5fb48ee3e9957de4860bb98d72c3890 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit ce2f153ebffabc659f79fbace6d9f42959470b40 Author: Burak Şen <[email protected]> AuthorDate: Wed Aug 5 21:53:05 2026 +0300 fix(proto): preserve empty projection when ser/de MemoryScanExec (#24087) ## Which issue does this PR close? - Closes #24085 ## Rationale for this change Empty projection is not ser/de correctly in MemoryScanExec ## What changes are included in this PR? Use same approach as `FilterExec`, `HashJoinExec` etc for projection encode/decode ## Are these changes tested? Yes, new roundtrip test `roundtrip_memory_source_empty_projection`, which fails on main. ## Are there any user-facing changes? No --- datafusion/proto/src/physical_plan/mod.rs | 27 ++++++++++------------ .../proto/tests/cases/roundtrip_physical_plan.rs | 19 +++++++++++++++ 2 files changed, 31 insertions(+), 15 deletions(-) diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index 79c6394933..c7d5bc9c4f 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/datafusion/proto/src/physical_plan/mod.rs @@ -1179,15 +1179,11 @@ pub trait PhysicalPlanNodeExt: Sized { })?; let schema: SchemaRef = SchemaRef::new(proto_schema.try_into()?); - let projection = if !scan.projection.is_empty() { - Some( - scan.projection - .iter() - .map(|i| *i as usize) - .collect::<Vec<_>>(), - ) - } else { - None + // Preserve the empty-projection sentinel written by `try_from_data_source_exec`. + let projection = match scan.projection.as_slice() { + [] => None, + [u32::MAX] => Some(Vec::new()), + indices => Some(indices.iter().map(|i| *i as usize).collect()), }; let mut sort_information = vec![]; @@ -2364,12 +2360,13 @@ pub trait PhysicalPlanNodeExt: Sized { let proto_schema: protobuf::Schema = source_conf.original_schema().as_ref().try_into()?; - let proto_projection = source_conf - .projection() - .as_ref() - .map_or_else(Vec::new, |v| { - v.iter().map(|x| *x as u32).collect::<Vec<u32>>() - }); + // Proto3 can't tell `None` from `Some(vec![])`; encode the latter + // as the `[u32::MAX]` sentinel, matching the join/filter nodes. + let proto_projection = match source_conf.projection().as_ref() { + None => Vec::new(), + Some(v) if v.is_empty() => vec![u32::MAX], + Some(v) => v.iter().map(|x| *x as u32).collect(), + }; let proto_sort_information = source_conf .sort_information() diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs index b22cfd7764..cbc50a96e9 100644 --- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs @@ -2678,6 +2678,25 @@ async fn roundtrip_empty_projection() -> Result<()> { roundtrip_test_sql_with_context(sql, &ctx).await } +#[tokio::test] +async fn roundtrip_memory_source_empty_projection() -> Result<()> { + // Memory scan: `Some(vec![])` must not decode back as `None` + let ctx = SessionContext::new(); + let batch = RecordBatch::try_new( + Arc::new(Schema::new(vec![ + Field::new("a", DataType::Utf8, false), + Field::new("b", DataType::Int64, false), + ])), + vec![ + Arc::new(arrow::array::StringArray::from(vec!["Tom"])), + Arc::new(arrow::array::Int64Array::from(vec![18i64])), + ], + )?; + ctx.register_batch("tmem", batch)?; + let sql = "select 1 from tmem"; + roundtrip_test_sql_with_context(sql, &ctx).await +} + #[tokio::test] async fn roundtrip_physical_plan_node() { use datafusion::prelude::*; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
