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,17 +684,13 @@ 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 {
-            None
-        };
+        let projection = Some(
+            filter
+                .projection
+                .iter()
+                .map(|i| *i as usize)
+                .collect::<Vec<_>>(),
+        );
 
         let filter = FilterExecBuilder::new(predicate, input)
             .apply_projection(projection)?
@@ -2364,9 +2360,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),
                 },