Make DefaultLogicalExtensionCodec support serialisation of build in file formats #19071

peterxcli · 2025-12-03T17:20:06Z

Which issue does this PR close?

Closes make DefaultLogicalExtensionCodec support serialisation of build in file formats #16944

Rationale for this change

Logical plans can include file sources. To serialize/deserialize logical plans (e.g., to ship between processes, store/replay), DataFusion needs to encode the file format configuration (CSV/JSON/Parquet/Arrow).
Previously, the default codec didn’t implement try_encode_file_format / try_decode_file_format (it returned “not implemented”). This change makes the default codec able to handle the common formats out-of-the-box.

What changes are included in this PR?

New try_decode_file_format(&self, buf: &[u8], ctx: &SessionContext) -> Result<Arc<dyn FileFormatFactory>> in DefaultLogicalExtensionCodec
New try_encode_file_format(&self, buf: &mut Vec<u8>, node: Arc<dyn FileFormatFactory>) -> Result<()> in DefaultLogicalExtensionCodec

Are these changes tested?

direct test:
- tests added inside datafusion/proto/tests/cases/roundtrip_logical_plan.rs
indirect test:
- roundtrip_arrow_scan — Uses logical_plan_to_bytes() and logical_plan_from_bytes(), which use DefaultLogicalExtensionCodec, so it exercises Arrow file format encoding/decoding.
- roundtrip_custom_listing_tables — Uses the default codec and may involve file formats.

Are there any user-facing changes?

Yes, makes the default codec able to handle the common formats out-of-the-box.

…ExtensionCodec * Implemented `try_decode_file_format` and `try_encode_file_format` methods to handle various file formats (CSV, JSON, Parquet, Arrow). * Introduced internal structures for file format handling, including `FileFormatKind`, `FileFormatProtoWrapper`, and `FileFormatWrapper`. * Enhanced error handling for unsupported file formats.

…ExtensionCodec-support-serialisation-of-build-in-file-formats

peterxcli · 2025-12-04T06:57:08Z

cc @milenkovicm Please take a look. Thanks!

milenkovicm

thanks @peterxcli it looks good on a quick glance, will have a better look later.
I have two comments to start with

datafusion/wasmtest/datafusion-wasm-app/package-lock.json

milenkovicm · 2025-12-04T09:24:36Z

datafusion_contributor_guide.md

why is this file added? perhaps its part of different PR ?

oh sorry i accidently committed these file, actually this file is just for myself to read.
thanks for catching, will revert

milenkovicm

Thanks for your contribution @peterxcli
I have some follow up questions, please have a look when you get chance.
Also, I'd like to ask you to which extend you used coding assistant(s) for this PR?

milenkovicm · 2025-12-06T13:13:20Z

datafusion/proto/src/logical_plan/mod.rs

+        fn try_decode_roundtrip(
+            ctx: &TaskContext,
+            buf: &[u8],
+        ) -> Option<Arc<dyn FileFormatFactory>> {
+            #[cfg(feature = "parquet")]
+            let candidates: &[&dyn LogicalExtensionCodec] = &[
+                &ParquetLogicalExtensionCodec {},
+                &CsvLogicalExtensionCodec {},
+                &JsonLogicalExtensionCodec {},
+            ];
+            #[cfg(not(feature = "parquet"))]
+            let candidates: &[&dyn LogicalExtensionCodec] =
+                &[&CsvLogicalExtensionCodec {}, &JsonLogicalExtensionCodec {}];
+
+            for codec in candidates {
+                if let Ok(ff) = codec.try_decode_file_format(buf, ctx) {
+                    let mut re = Vec::new();
+                    if codec
+                        .try_encode_file_format(
+                            &mut re,
+                            Arc::<dyn FileFormatFactory>::clone(&ff),
+                        )
+                        .is_ok()
+                        && re == buf
+                    {
+                        return Some(ff);
+                    }
+                }
+            }
+            None
+        }
+
+        if let Some(ff) = try_decode_roundtrip(ctx, buf) {
+            return Ok(ff);
+        }
+
+        // If nothing matched, return a clear error rather than guessing
+        exec_err!(
+            "Unsupported FileFormatFactory bytes for DefaultLogicalExtensionCodec ({} bytes)",
+            buf.len()
+        )


whats the purpose of this code?

milenkovicm · 2025-12-06T13:14:47Z

datafusion/proto/src/logical_plan/mod.rs

+#[derive(Clone, PartialEq, ::prost::Message)]
+struct FileFormatProtoWrapper {
+    #[prost(int32, tag = "1")]
+    kind: i32,
+    #[prost(bytes, tag = "2")]
+    blob: Vec<u8>,
+}


whould it make sense to have this as part of proto file ?

milenkovicm · 2025-12-06T13:15:42Z

datafusion/proto/src/logical_plan/mod.rs

+struct FileFormatWrapper {
+    kind: FileFormatKind,
+    blob: Vec<u8>,
+}
+
+impl FileFormatWrapper {
+    fn new(kind: FileFormatKind, blob: Vec<u8>) -> Self {
+        Self { kind, blob }
+    }
+
+    fn encode_into(self, buf: &mut Vec<u8>) -> Result<()> {
+        let proto = FileFormatProtoWrapper {
+            kind: self.kind as i32,
+            blob: self.blob,
+        };
+        proto
+            .encode(buf)
+            .map_err(|e| DataFusionError::Internal(e.to_string()))
+    }
+
+    fn try_decode(buf: &[u8]) -> Result<Self> {
+        let proto = FileFormatProtoWrapper::decode(buf)
+            .map_err(|e| DataFusionError::Internal(e.to_string()))?;
+        let kind = match proto.kind {
+            0 => FileFormatKind::Csv,
+            1 => FileFormatKind::Json,
+            2 => FileFormatKind::Parquet,
+            3 => FileFormatKind::Arrow,
+            _ => {
+                return Err(DataFusionError::Internal(
+                    "Unknown file format kind".to_string(),
+                ))
+            }
+        };
+        Ok(Self {
+            kind,
+            blob: proto.blob,
+        })
+    }


do we need structure additional to actual decode function

milenkovicm · 2025-12-06T13:17:57Z