This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new ce2f153ebf fix(proto): preserve empty projection when ser/de
MemoryScanExec (#24087)
ce2f153ebf is described below
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]
