Serialize parquet_options in datafusion-proto

datafusion/proto/proto/datafusion.proto

@@ -1005,6 +1005,8 @@ message ParquetScanExecNode {
   reserved 2;
 
   PhysicalExprNode predicate = 3;
+
+  datafusion_common.TableParquetOptions parquet_options = 4;
 }
 
 message CsvScanExecNode {

datafusion/proto/src/physical_plan/mod.rs

@@ -258,6 +258,11 @@ impl AsExecutionPlan for protobuf::PhysicalPlanNode {
                         })
                         .transpose()?;
                     let mut builder = ParquetExec::builder(base_config);
+
+                    if let Some(options) = scan.parquet_options.as_ref() {
+                        builder = builder.with_table_parquet_options(options.try_into()?)
+                    }
+
                     if let Some(predicate) = predicate {
                         builder = builder.with_predicate(predicate)
                     }
@@ -1654,6 +1659,7 @@ impl AsExecutionPlan for protobuf::PhysicalPlanNode {
                             extension_codec,
                         )?),
                         predicate,
+                        parquet_options: Some(exec.table_parquet_options().try_into()?),
                     },
                 )),
             });

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -730,9 +730,14 @@ fn roundtrip_parquet_exec_with_pruning_predicate() -> Result<()> {
         Operator::Eq,
         lit("1"),
     ));
+
+    let mut options = TableParquetOptions::new();
+    options.global.pushdown_filters = true;
+
     roundtrip_test(
         ParquetExec::builder(scan_config)
             .with_predicate(predicate)
+            .with_table_parquet_options(options)
             .build_arc(),
     )
 }