fix empty projection proto roundtrip (apache#15449)

Author: XiangpengHao

datafusion/proto/src/physical_plan/from_proto.rs

@@ -489,11 +489,6 @@ pub fn parse_protobuf_file_scan_config(
         .iter()
         .map(|i| *i as usize)
         .collect::<Vec<_>>();
-    let projection = if projection.is_empty() {
-        None
-    } else {
-        Some(projection)
-    };
 
     let constraints = convert_required!(proto.constraints)?;
     let statistics = convert_required!(proto.statistics)?;
@@ -543,7 +538,7 @@ pub fn parse_protobuf_file_scan_config(
         .with_file_groups(file_groups)
         .with_constraints(constraints)
         .with_statistics(statistics)
-        .with_projection(projection)
+        .with_projection(Some(projection))
         .with_limit(proto.limit.as_ref().map(|sl| sl.limit as usize))
         .with_table_partition_cols(table_partition_cols)
         .with_output_ordering(output_ordering)

datafusion/proto/src/physical_plan/to_proto.rs

@@ -512,7 +512,7 @@ pub fn serialize_file_scan_config(
         projection: conf
             .projection
             .as_ref()
-            .unwrap_or(&vec![])
+            .unwrap_or(&(0..schema.fields().len()).collect::<Vec<_>>())
             .iter()
             .map(|n| *n as u32)
             .collect(),

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -1655,3 +1655,10 @@ async fn roundtrip_parquet_select_projection_predicate() -> Result<()> {
     let sql = "select string_col, timestamp_col from alltypes_plain where id > 4";
     roundtrip_test_sql_with_context(sql, &ctx).await
 }
+
+#[tokio::test]
+async fn roundtrip_empty_projection() -> Result<()> {
+    let ctx = all_types_context().await?;
+    let sql = "select 1 from alltypes_plain";
+    roundtrip_test_sql_with_context(sql, &ctx).await
+}