preserve FilterExec batch size during ser/de

Author: askalt

datafusion/physical-plan/src/filter.rs

@@ -240,6 +240,11 @@ impl FilterExec {
         FilterExecBuilder::new(predicate, input).build()
     }
 
+    /// Get a batch size
+    pub fn batch_size(&self) -> usize {
+        self.batch_size
+    }
+
     /// Set the default selectivity
     pub fn with_default_selectivity(
         mut self,

datafusion/proto/proto/datafusion.proto

@@ -1007,6 +1007,7 @@ message FilterExecNode {
   PhysicalExprNode expr = 2;
   uint32 default_filter_selectivity = 3;
   repeated uint32 projection = 9;
+  uint32 batch_size = 10;
 }
 
 message FileGroup {

datafusion/proto/src/generated/pbjson.rs

@@ -590,6 +590,7 @@ impl protobuf::PhysicalPlanNode {
 
         let filter = FilterExecBuilder::new(predicate, input)
             .apply_projection(projection)?
+            .with_batch_size(filter.batch_size as usize)
             .build()?;
         match filter_selectivity {
             Ok(filter_selectivity) => Ok(Arc::new(
@@ -2121,6 +2122,7 @@ impl protobuf::PhysicalPlanNode {
                     projection: exec.projection().as_ref().map_or_else(Vec::new, |v| {
                         v.iter().map(|x| *x as u32).collect::<Vec<u32>>()
                     }),
+                    batch_size: exec.batch_size() as u32,
                 },
             ))),
         })