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

Author: buraksenn

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()

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::*;