serde: fix FilterExec with empty projection

Prior this patch FilterExec with empty projection was deserialized
as FilterExec with *-projection

Author: Egor Markov

datafusion/proto/src/physical_plan/mod.rs

@@ -684,16 +684,19 @@ impl protobuf::PhysicalPlanNode {
             })?;
 
         let filter_selectivity = filter.default_filter_selectivity.try_into();
-        let projection = if !filter.projection.is_empty() {
-            Some(
-                filter
-                    .projection
-                    .iter()
-                    .map(|i| *i as usize)
-                    .collect::<Vec<_>>(),
-            )
-        } else {
+        let mut projection = Vec::with_capacity(filter.projection.len());
+        let mut is_full_projection =
+            filter.projection.len() == input.schema().fields.len();
+        for (i, &idx) in filter.projection.iter().enumerate() {
+            let idx = idx as usize;
+            is_full_projection &= idx == i;
+            projection.push(idx);
+        }
+        let projection = if is_full_projection {
+            // Store None instead of continuous numbers vector.
             None
+        } else {
+            Some(projection)
         };
 
         let filter = FilterExecBuilder::new(predicate, input)
@@ -2364,9 +2367,10 @@ impl protobuf::PhysicalPlanNode {
                             .physical_expr_to_proto(exec.predicate(), codec)?,
                     ),
                     default_filter_selectivity: exec.default_selectivity() as u32,
-                    projection: exec.projection().as_ref().map_or_else(Vec::new, |v| {
-                        v.iter().map(|x| *x as u32).collect::<Vec<u32>>()
-                    }),
+                    projection: exec.projection().as_ref().map_or(
+                        (0..exec.schema().fields().len() as u32).collect::<Vec<u32>>(),
+                        |v| v.iter().map(|x| *x as u32).collect::<Vec<u32>>(),
+                    ),
                     batch_size: exec.batch_size() as u32,
                     fetch: exec.fetch().map(|f| f as u32),
                 },