fix(proto): destructure JSON source and sink serde hooks

datafusion/datasource-json/src/file_format.rs

@@ -500,6 +500,13 @@ impl DataSink for JsonSink {
         use datafusion_proto_models::protobuf;
         use protobuf::physical_plan_node::PhysicalPlanType;
 
+        // Keep an exhaustive guard in the active hook while centralizing the
+        // field mapping in the exhaustive `TryFrom<&JsonSink>` below.
+        let Self {
+            config: _,
+            writer_options: _,
+        } = self;
+
         let input = ctx.encode_child(exec.input())?;
         let sort_order = exec.encode_sort_order(ctx)?;
         let sink = protobuf::JsonSink::try_from(self)?;
@@ -520,9 +527,14 @@ impl TryFrom<&JsonSink> for datafusion_proto_models::protobuf::JsonSink {
     type Error = datafusion_common::DataFusionError;
 
     fn try_from(value: &JsonSink) -> Result<Self> {
+        // Keep this public conversion exhaustive like the active serde hook.
+        let JsonSink {
+            config,
+            writer_options,
+        } = value;
         Ok(Self {
-            config: Some(value.config().try_into()?),
-            writer_options: Some(value.writer_options().try_into()?),
+            config: Some(config.try_into()?),
+            writer_options: Some(writer_options.try_into()?),
         })
     }
 }
@@ -532,14 +544,17 @@ impl TryFrom<&datafusion_proto_models::protobuf::JsonSink> for JsonSink {
     type Error = datafusion_common::DataFusionError;
 
     fn try_from(value: &datafusion_proto_models::protobuf::JsonSink) -> Result<Self> {
-        let config =
-            FileSinkConfig::try_from(value.config.as_ref().ok_or_else(|| {
-                datafusion_common::internal_datafusion_err!(
-                    "JsonSink is missing required field 'config'"
-                )
-            })?)?;
-        let writer_options = value
-            .writer_options
+        // Exhaustive destructure: new wire fields must be explicitly restored.
+        let datafusion_proto_models::protobuf::JsonSink {
+            config,
+            writer_options,
+        } = value;
+        let config = FileSinkConfig::try_from(config.as_ref().ok_or_else(|| {
+            datafusion_common::internal_datafusion_err!(
+                "JsonSink is missing required field 'config'"
+            )
+        })?)?;
+        let writer_options = writer_options
             .as_ref()
             .ok_or_else(|| {
                 datafusion_common::internal_datafusion_err!(
@@ -566,19 +581,26 @@ impl JsonSink {
             protobuf::physical_plan_node::PhysicalPlanType::JsonSink,
             "JsonSink",
         );
-        let input = ctx.decode_required_child(
-            sink_node.input.as_deref(),
-            "JsonSinkExecNode",
-            "input",
-        )?;
-        let proto_sink = sink_node.sink.as_ref().ok_or_else(|| {
+        // Exhaustive destructure: a new field on `JsonSinkExecNode` is a
+        // compile error here rather than a silently ignored wire field.
+        let protobuf::JsonSinkExecNode {
+            input,
+            sink,
+            // The output schema is recomputed by `DataSinkExec::new`.
+            sink_schema: _,
+            sort_order,
+        } = sink_node.as_ref();
+
+        let input =
+            ctx.decode_required_child(input.as_deref(), "JsonSinkExecNode", "input")?;
+        let proto_sink = sink.as_ref().ok_or_else(|| {
             datafusion_common::internal_datafusion_err!(
                 "JsonSinkExecNode is missing required field 'sink'"
             )
         })?;
         let data_sink = JsonSink::try_from(proto_sink)?;
         let sort_order = DataSinkExec::decode_sort_order(
-            sink_node.sort_order.as_ref(),
+            sort_order.as_ref(),
             ctx,
             input.schema().as_ref(),
         )?;

datafusion/datasource-json/src/source.rs

@@ -257,9 +257,24 @@ impl FileSource for JsonSource {
         use datafusion_proto_models::protobuf;
         use protobuf::physical_plan_node::PhysicalPlanType;
 
+        // Exhaustive destructure: adding a field to `JsonSource` without
+        // deciding how it is serialized is a compile error, not a silent
+        // round-trip gap.
+        let Self {
+            // Serialized in `base` and used to rebuild the source on decode.
+            table_schema: _,
+            // Set from `FileScanConfig` when the scan is opened.
+            batch_size: _,
+            // Runtime metrics, not part of the plan.
+            metrics: _,
+            // Serialized in `base` and reapplied on decode.
+            projection: _,
+            newline_delimited,
+        } = self;
+
         let node = protobuf::JsonScanExecNode {
             base_conf: Some(base.try_to_proto(ctx)?),
-            newline_delimited: if self.newline_delimited {
+            newline_delimited: if *newline_delimited {
                 None
             } else {
                 Some(false)
@@ -292,7 +307,14 @@ impl JsonSource {
             );
         };
 
-        let base_conf = scan.base_conf.as_ref().ok_or_else(|| {
+        // Exhaustive destructure: a new field on `JsonScanExecNode` is a
+        // compile error here rather than a silently ignored wire field.
+        let protobuf::JsonScanExecNode {
+            base_conf,
+            newline_delimited,
+        } = scan;
+
+        let base_conf = base_conf.as_ref().ok_or_else(|| {
             datafusion_common::internal_datafusion_err!(
                 "JsonScanExecNode is missing required field 'base_conf'"
             )
@@ -301,7 +323,7 @@ impl JsonSource {
         let table_schema = FileScanConfig::parse_table_schema_from_proto(base_conf)?;
         let source = Arc::new(
             JsonSource::new(table_schema)
-                .with_newline_delimited(scan.newline_delimited.unwrap_or(true)),
+                .with_newline_delimited(newline_delimited.unwrap_or(true)),
         );
 
         let conf = FileScanConfig::try_from_proto(base_conf, ctx, source)?;

datafusion/proto-common/proto/datafusion_common.proto

@@ -431,6 +431,7 @@ enum CompressionTypeVariant {
 
 message JsonWriterOptions {
   CompressionTypeVariant compression = 1;
+  optional uint32 compression_level = 2;
 }
 
 

datafusion/proto-common/src/from_proto/mod.rs

@@ -995,8 +995,19 @@ impl TryFrom<&protobuf::JsonWriterOptions> for JsonWriterOptions {
     fn try_from(
         opts: &protobuf::JsonWriterOptions,
     ) -> datafusion_common::Result<Self, Self::Error> {
-        let compression: CompressionTypeVariant = opts.compression().into();
-        Ok(JsonWriterOptions::new(compression))
+        // Exhaustive destructure: new wire fields must be explicitly restored.
+        let protobuf::JsonWriterOptions {
+            compression,
+            compression_level,
+        } = opts;
+        let compression: CompressionTypeVariant =
+            protobuf::CompressionTypeVariant::try_from(*compression)
+                .unwrap_or_default()
+                .into();
+        Ok(JsonWriterOptions {
+            compression,
+            compression_level: *compression_level,
+        })
     }
 }
 

datafusion/proto-common/src/generated/pbjson.rs

@@ -5164,12 +5164,18 @@ impl serde::Serialize for JsonWriterOptions {
         if self.compression != 0 {
             len += 1;
         }
+        if self.compression_level.is_some() {
+            len += 1;
+        }
         let mut struct_ser = serializer.serialize_struct("datafusion_common.JsonWriterOptions", len)?;
         if self.compression != 0 {
             let v = CompressionTypeVariant::try_from(self.compression)
                 .map_err(|_| serde::ser::Error::custom(format!("Invalid variant {}", self.compression)))?;
             struct_ser.serialize_field("compression", &v)?;
         }
+        if let Some(v) = self.compression_level.as_ref() {
+            struct_ser.serialize_field("compressionLevel", v)?;
+        }
         struct_ser.end()
     }
 }
@@ -5181,11 +5187,14 @@ impl<'de> serde::Deserialize<'de> for JsonWriterOptions {
     {
         const FIELDS: &[&str] = &[
             "compression",
+            "compression_level",
+            "compressionLevel",
         ];
 
         #[allow(clippy::enum_variant_names)]
         enum GeneratedField {
             Compression,
+            CompressionLevel,
         }
         impl<'de> serde::Deserialize<'de> for GeneratedField {
             fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -5208,6 +5217,7 @@ impl<'de> serde::Deserialize<'de> for JsonWriterOptions {
                     {
                         match value {
                             "compression" => Ok(GeneratedField::Compression),
+                            "compressionLevel" | "compression_level" => Ok(GeneratedField::CompressionLevel),
                             _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
                         }
                     }
@@ -5228,6 +5238,7 @@ impl<'de> serde::Deserialize<'de> for JsonWriterOptions {
                     V: serde::de::MapAccess<'de>,
             {
                 let mut compression__ = None;
+                let mut compression_level__ = None;
                 while let Some(k) = map_.next_key()? {
                     match k {
                         GeneratedField::Compression => {
@@ -5236,10 +5247,19 @@ impl<'de> serde::Deserialize<'de> for JsonWriterOptions {
                             }
                             compression__ = Some(map_.next_value::<CompressionTypeVariant>()? as i32);
                         }
+                        GeneratedField::CompressionLevel => {
+                            if compression_level__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("compressionLevel"));
+                            }
+                            compression_level__ = 
+                                map_.next_value::<::std::option::Option<::pbjson::private::NumberDeserialize<_>>>()?.map(|x| x.0)
+                            ;
+                        }
                     }
                 }
                 Ok(JsonWriterOptions {
                     compression: compression__.unwrap_or_default(),
+                    compression_level: compression_level__,
                 })
             }
         }

datafusion/proto-common/src/generated/prost.rs

@@ -583,6 +583,8 @@ pub struct EmptyMessage {}
 pub struct JsonWriterOptions {
     #[prost(enumeration = "CompressionTypeVariant", tag = "1")]
     pub compression: i32,
+    #[prost(uint32, optional, tag = "2")]
+    pub compression_level: ::core::option::Option<u32>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct CsvWriterOptions {

datafusion/proto-common/src/to_proto/mod.rs

@@ -885,9 +885,16 @@ impl TryFrom<&JsonWriterOptions> for protobuf::JsonWriterOptions {
     fn try_from(
         opts: &JsonWriterOptions,
     ) -> datafusion_common::Result<Self, Self::Error> {
-        let compression: protobuf::CompressionTypeVariant = opts.compression.into();
+        // Exhaustive destructure: new writer options must be explicitly
+        // included in the wire representation.
+        let JsonWriterOptions {
+            compression,
+            compression_level,
+        } = opts;
+        let compression: protobuf::CompressionTypeVariant = (*compression).into();
         Ok(protobuf::JsonWriterOptions {
             compression: compression.into(),
+            compression_level: *compression_level,
         })
     }
 }

datafusion/proto-models/src/generated/datafusion_proto_common.rs

@@ -583,6 +583,8 @@ pub struct EmptyMessage {}
 pub struct JsonWriterOptions {
     #[prost(enumeration = "CompressionTypeVariant", tag = "1")]
     pub compression: i32,
+    #[prost(uint32, optional, tag = "2")]
+    pub compression_level: ::core::option::Option<u32>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct CsvWriterOptions {

datafusion/proto/tests/cases/plans/sinks.rs

@@ -27,10 +27,12 @@ use datafusion::datasource::file_format::json::JsonSink;
 use datafusion::datasource::file_format::parquet::ParquetSink;
 use datafusion::datasource::listing::{ListingTableUrl, PartitionedFile};
 use datafusion::datasource::object_store::ObjectStoreUrl;
-use datafusion::datasource::physical_plan::{FileGroup, FileOutputMode, FileSinkConfig};
+use datafusion::datasource::physical_plan::{
+    FileGroup, FileOutputMode, FileSink, FileSinkConfig,
+};
 use datafusion::datasource::sink::{DataSink, DataSinkExec};
 use datafusion::execution::TaskContext;
-use datafusion::physical_expr::PhysicalSortRequirement;
+use datafusion::physical_expr::{LexRequirement, PhysicalSortRequirement};
 use datafusion::physical_plan::expressions::Column;
 use datafusion::physical_plan::placeholder_row::PlaceholderRowExec;
 use datafusion::physical_plan::proto::ExecutionPlanEncodeCtx;
@@ -218,9 +220,9 @@ fn roundtrip_json_sink() -> Result<()> {
     };
     let data_sink = Arc::new(JsonSink::new(
         file_sink_config,
-        JsonWriterOptions::new(CompressionTypeVariant::UNCOMPRESSED),
+        JsonWriterOptions::new_with_level(CompressionTypeVariant::ZSTD, 7),
     ));
-    let sort_order = [PhysicalSortRequirement::new(
+    let sort_order: LexRequirement = [PhysicalSortRequirement::new(
         Arc::new(Column::new("plan_type", 0)),
         Some(SortOptions {
             descending: true,
@@ -229,11 +231,45 @@ fn roundtrip_json_sink() -> Result<()> {
     )]
     .into();
 
-    roundtrip_test(Arc::new(DataSinkExec::new(
-        input,
-        data_sink,
-        Some(sort_order),
-    )))
+    let ctx = SessionContext::new();
+    let codec = DefaultPhysicalExtensionCodec {};
+    let proto_converter = DefaultPhysicalProtoConverter {};
+    let roundtrip_plan = roundtrip_test_and_return(
+        Arc::new(DataSinkExec::new(
+            input,
+            data_sink,
+            Some(sort_order.clone()),
+        )),
+        &ctx,
+        &codec,
+        &proto_converter,
+    )?;
+
+    let roundtrip_plan =
+        roundtrip_plan
+            .downcast_ref::<DataSinkExec>()
+            .ok_or_else(|| {
+                datafusion_common::internal_datafusion_err!("Expected DataSinkExec")
+            })?;
+    let json_sink = roundtrip_plan
+        .sink()
+        .downcast_ref::<JsonSink>()
+        .ok_or_else(|| {
+            datafusion_common::internal_datafusion_err!("Expected JsonSink")
+        })?;
+    assert_eq!(json_sink.config().insert_op, InsertOp::Overwrite);
+    assert!(json_sink.config().keep_partition_by_columns);
+    assert_eq!(
+        json_sink.config().file_output_mode,
+        FileOutputMode::SingleFile
+    );
+    assert_eq!(
+        json_sink.writer_options().compression,
+        CompressionTypeVariant::ZSTD
+    );
+    assert_eq!(json_sink.writer_options().compression_level, Some(7));
+    assert_eq!(roundtrip_plan.sort_order(), &Some(sort_order));
+    Ok(())
 }
 
 #[test]