You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "timsaucer (via GitHub)" <gi...@apache.org> on 2025/12/21 13:58:08 UTC
[PR] Add Physical Extension Protobuf Codec [datafusion]
timsaucer opened a new pull request, #19437:
URL: https://github.com/apache/datafusion/pull/19437
## Which issue does this PR close?
- Closes https://github.com/apache/datafusion/issues/18477
## Rationale for this change
This PR adds a new trait for converting to and from Protobuf objects and Physical expressions and plans.
### DRAFT
This is in draft form at the time of this writing so we can compare approaches between this and https://github.com/apache/datafusion/pull/19267 and https://github.com/apache/datafusion/pull/19234
## What changes are included in this PR?
- Add `PhysicalExtensionProtoCodec` and default implementation.
- Update `from_proto` methods to use this trait.
## Are these changes tested?
Two examples are added, but additional testing still needs to be done
## Are there any user-facing changes?
Users will need to specific a protobuf codec, but a default implementation is provided.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3770144434
> It unfortunately has conflicts which blocks CI from running. I'd be happy to resolve them but don't want to force push to your branch. Could you clear them up so CI can run?
It's running now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3794658074
I'm AFK for the weekend but added this to my review queue for the coming week
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3735713827
<img width="262" height="62" alt="Screenshot 2026-01-11 at 3 28 50 PM" src="https://github.com/user-attachments/assets/d0eac98c-a0a4-494a-b13f-07f9085b7004" />
I see what you did there
<img width="300" height="169" alt="image" src="https://github.com/user-attachments/assets/c66439f5-2dce-4811-9a22-1551cdc4976b" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2725504109
##########
docs/source/library-user-guide/upgrading.md:
##########
@@ -154,6 +154,48 @@ The builder pattern is more efficient as it computes properties once during `bui
Note: `with_default_selectivity()` is not deprecated as it simply updates a field value and does not require the overhead of the builder pattern.
+### Protobuf conversion trait added
+
+A new trait, `PhysicalProtoConverterExtension`, has been added to the `datafusion-proto`
+crate. This is used for controlling the process of conversion of physical plans and
+expressions to and from their protobuf equivalents. The methods for conversion now
+require an additional parameter. Most users can simply use the `DefaultPhysicalProtoConverter`
+implementation.
+
+For conversion from protobuf types to their physical types:
+
+**Before:**
+
+```rust,ignore
+let plan = proto_plan.try_into_physical_plan(ctx, codec)?;
+let expr = parse_physical_expr(proto_expr, ctx, input_schema, codec)?;
+```
+
+**After:**
+
+```rust,ignore
+let converter = DefaultPhysicalProtoConverter {};
+let plan = proto_plan.try_into_physical_plan(ctx, codec, &converter)?;
Review Comment:
I've applied your suggestion and now the diff is smaller. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3699421831
> Sorry for late reply! Had a quick look, proposal does make sense to me.
>
> Introduction of `PhysicalExtensionProtoCodec` is a bit confusing to me, maybe we need better name. I believe average user does not need to know about it so `protobuf::PhysicalPlanNode::try_from_physical_plan` could stay as it was and we can add `protobuf::PhysicalPlanNode::try_from_physical_plan_with_custom_proto_codec(...)` (or similar)
Sound advice! I wanted to make sure the approach as a whole seems reasonable to you both.
I agree about the naming. It's not actually a codec. It doesn't actually do any serialization or deserialization. I tried working with a LLM to come up with a better name.
- `PhysicalExtensionProtoAdapter`
- `PhysicalExtensionProtoConverter`
- `PhysicalExtensionProtoTranslator`
And maybe the name ordering should actually be
- `PhysicalProtoTranslatorExtension` because the thing we're extending is the proto translation?
I'm on PTO through next week, but I think in principle we have a path we want to go down. I would suggest we give equal treatment to the logical side and make sure the current paths are not impacted/use the default.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2654880063
##########
datafusion-examples/examples/custom_data_source/adapter_serialization.rs:
##########
@@ -0,0 +1,527 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! See `main.rs` for how to run it.
+//!
+//! This example demonstrates how to use the `PhysicalExtensionCodec` trait's
+//! interception methods (`serialize_physical_plan` and `deserialize_physical_plan`)
+//! to implement custom serialization logic.
+//!
+//! The key insight is that `FileScanConfig::expr_adapter_factory` is NOT serialized by
+//! default. This example shows how to:
+//! 1. Detect plans with custom adapters during serialization
+//! 2. Wrap them as Extension nodes with JSON-serialized adapter metadata
+//! 3. Unwrap and restore the adapter during deserialization
+//!
+//! This demonstrates nested serialization (protobuf outer, JSON inner) and the power
+//! of the `PhysicalExtensionCodec` interception pattern. Both plan and expression
+//! serialization route through the codec, enabling interception at every node in the tree.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use arrow::array::record_batch;
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::assert_batches_eq;
+use datafusion::common::Result;
+use datafusion::common::not_impl_err;
+use datafusion::datasource::listing::{
+ ListingTable, ListingTableConfig, ListingTableConfigExt, ListingTableUrl,
+};
+use datafusion::datasource::physical_plan::{FileScanConfig, FileScanConfigBuilder};
+use datafusion::datasource::source::DataSourceExec;
+use datafusion::execution::TaskContext;
+use datafusion::execution::context::SessionContext;
+use datafusion::execution::object_store::ObjectStoreUrl;
+use datafusion::parquet::arrow::ArrowWriter;
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::ExecutionPlan;
+use datafusion::prelude::SessionConfig;
+use datafusion_physical_expr_adapter::{
+ DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
+};
+use datafusion_proto::bytes::{
+ physical_plan_from_bytes_with_extension_codec,
+ physical_plan_to_bytes_with_extension_codec,
+};
+use datafusion_proto::physical_plan::{
+ AsExecutionPlan, PhysicalExtensionCodec, PhysicalExtensionProtoCodec,
+};
+use datafusion_proto::protobuf::{
+ PhysicalExprNode, PhysicalExtensionNode, PhysicalPlanNode,
+ physical_plan_node::PhysicalPlanType,
+};
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, PutPayload};
+use prost::Message;
+use serde::{Deserialize, Serialize};
+
+/// Example showing how to preserve custom adapter information during plan serialization.
+///
+/// This demonstrates:
+/// 1. Creating a custom PhysicalExprAdapter with metadata
+/// 2. Using PhysicalExtensionCodec to intercept serialization
+/// 3. Wrapping adapter info as Extension nodes
+/// 4. Restoring adapters during deserialization
+pub async fn adapter_serialization() -> Result<()> {
+ println!("=== PhysicalExprAdapter Serialization Example ===\n");
+
+ // Step 1: Create sample Parquet data in memory
+ println!("Step 1: Creating sample Parquet data...");
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let batch = record_batch!(("id", Int32, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))?;
+ let path = Path::from("data.parquet");
+ write_parquet(&store, &path, &batch).await?;
+
+ // Step 2: Set up session with custom adapter
+ println!("Step 2: Setting up session with custom adapter...");
+ let logical_schema =
+ Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+
+ let mut cfg = SessionConfig::new();
+ cfg.options_mut().execution.parquet.pushdown_filters = true;
+ let ctx = SessionContext::new_with_config(cfg);
+ ctx.runtime_env().register_object_store(
+ ObjectStoreUrl::parse("memory://")?.as_ref(),
+ Arc::clone(&store),
+ );
+
+ // Create a table with our custom MetadataAdapterFactory
+ let adapter_factory = Arc::new(MetadataAdapterFactory::new("v1"));
+ let listing_config =
+ ListingTableConfig::new(ListingTableUrl::parse("memory:///data.parquet")?)
+ .infer_options(&ctx.state())
+ .await?
+ .with_schema(logical_schema)
+ .with_expr_adapter_factory(
+ Arc::clone(&adapter_factory) as Arc<dyn PhysicalExprAdapterFactory>
+ );
+ let table = ListingTable::try_new(listing_config)?;
+ ctx.register_table("my_table", Arc::new(table))?;
+
+ // Step 3: Create physical plan with filter
+ println!("Step 3: Creating physical plan with filter...");
+ let df = ctx.sql("SELECT * FROM my_table WHERE id > 5").await?;
+ let original_plan = df.create_physical_plan().await?;
+
+ // Verify adapter is present in original plan
+ let has_adapter_before = verify_adapter_in_plan(&original_plan, "original");
+ println!(" Original plan has adapter: {has_adapter_before}");
+
+ // Step 4: Serialize with our custom codec
+ println!("\nStep 4: Serializing plan with AdapterPreservingCodec...");
+ let codec = AdapterPreservingCodec;
+ let bytes = physical_plan_to_bytes_with_extension_codec(
+ Arc::clone(&original_plan),
+ &codec,
+ &codec,
+ )?;
+ println!(" Serialized {} bytes", bytes.len());
+ println!(" (DataSourceExec with adapter was wrapped as PhysicalExtensionNode)");
+
+ // Step 5: Deserialize with our custom codec
+ println!("\nStep 5: Deserializing plan with AdapterPreservingCodec...");
+ let task_ctx = ctx.task_ctx();
+ let restored_plan =
+ physical_plan_from_bytes_with_extension_codec(&bytes, &task_ctx, &codec, &codec)?;
+
+ // Verify adapter is restored
+ let has_adapter_after = verify_adapter_in_plan(&restored_plan, "restored");
+ println!(" Restored plan has adapter: {has_adapter_after}");
+
+ // Step 6: Execute and compare results
+ println!("\nStep 6: Executing plans and comparing results...");
+ let original_results =
+ datafusion::physical_plan::collect(Arc::clone(&original_plan), task_ctx.clone())
+ .await?;
+ let restored_results =
+ datafusion::physical_plan::collect(restored_plan, task_ctx).await?;
+
+ #[rustfmt::skip]
+ let expected = [
+ "+----+",
+ "| id |",
+ "+----+",
+ "| 6 |",
+ "| 7 |",
+ "| 8 |",
+ "| 9 |",
+ "| 10 |",
+ "+----+",
+ ];
+
+ println!("\n Original plan results:");
+ arrow::util::pretty::print_batches(&original_results)?;
+ assert_batches_eq!(expected, &original_results);
+
+ println!("\n Restored plan results:");
+ arrow::util::pretty::print_batches(&restored_results)?;
+ assert_batches_eq!(expected, &restored_results);
+
+ println!("\n=== Example Complete! ===");
+ println!("Key takeaways:");
+ println!(
+ " 1. PhysicalExtensionCodec provides serialize_physical_plan/deserialize_physical_plan hooks"
+ );
+ println!(" 2. Custom metadata can be wrapped as PhysicalExtensionNode");
+ println!(" 3. Nested serialization (protobuf + JSON) works seamlessly");
+ println!(
+ " 4. Both plans produce identical results despite serialization round-trip"
+ );
+ println!(" 5. Adapters are fully preserved through the serialization round-trip");
+
+ Ok(())
+}
+
+// ============================================================================
+// MetadataAdapter - A simple custom adapter with a tag
+// ============================================================================
+
+/// A custom PhysicalExprAdapter that wraps another adapter.
+/// The tag metadata is stored in the factory, not the adapter itself.
+#[derive(Debug)]
+struct MetadataAdapter {
+ inner: Arc<dyn PhysicalExprAdapter>,
+}
+
+impl PhysicalExprAdapter for MetadataAdapter {
+ fn rewrite(&self, expr: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
+ // Simply delegate to inner adapter
+ self.inner.rewrite(expr)
+ }
+}
+
+// ============================================================================
+// MetadataAdapterFactory - Factory for creating MetadataAdapter instances
+// ============================================================================
+
+/// Factory for creating MetadataAdapter instances.
+/// The tag is stored in the factory and extracted via Debug formatting in `extract_adapter_tag`.
+#[derive(Debug)]
+struct MetadataAdapterFactory {
+ // Note: This field is read via Debug formatting in `extract_adapter_tag`.
+ // Rust's dead code analysis doesn't recognize Debug-based field access.
+ // In PR #19234, this field is used by `with_partition_values`, but that method
+ // doesn't exist in upstream DataFusion's PhysicalExprAdapter trait.
+ #[allow(dead_code)]
+ tag: String,
+}
+
+impl MetadataAdapterFactory {
+ fn new(tag: impl Into<String>) -> Self {
+ Self { tag: tag.into() }
+ }
+}
+
+impl PhysicalExprAdapterFactory for MetadataAdapterFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> Arc<dyn PhysicalExprAdapter> {
+ let inner = DefaultPhysicalExprAdapterFactory
+ .create(logical_file_schema, physical_file_schema);
+ Arc::new(MetadataAdapter { inner })
+ }
+}
+
+// ============================================================================
+// AdapterPreservingCodec - Custom codec that preserves adapters
+// ============================================================================
+
+/// Extension payload structure for serializing adapter info
+#[derive(Serialize, Deserialize)]
+struct ExtensionPayload {
+ /// Marker to identify this is our custom extension
+ marker: String,
+ /// JSON-serialized adapter metadata
+ adapter_metadata: AdapterMetadata,
+ /// Protobuf-serialized inner DataSourceExec (without adapter)
+ inner_plan_bytes: Vec<u8>,
+}
+
+/// Metadata about the adapter to recreate it during deserialization
+#[derive(Serialize, Deserialize)]
+struct AdapterMetadata {
+ /// The adapter tag (e.g., "v1")
+ tag: String,
+}
+
+const EXTENSION_MARKER: &str = "adapter_preserving_extension_v1";
+
+/// A codec that intercepts serialization to preserve adapter information.
+#[derive(Debug)]
+struct AdapterPreservingCodec;
+
+impl PhysicalExtensionCodec for AdapterPreservingCodec {
+ // Required method: decode custom extension nodes
+ fn try_decode(
+ &self,
+ buf: &[u8],
+ _inputs: &[Arc<dyn ExecutionPlan>],
Review Comment:
It's just a nit but I was imaging that instead of storing the children in `inner_plan_bytes` we would recurse into calling the serialization machinery on the children and pass them into `PhysicalPlanNode` instead of `input: vec![]`. Then they'd come back as `inputs: &[Arc<dyn ExecutionPlan>]` in this function and we'd once again call the default deserializer / recurse (because it might call us back) to deserialize the children. Then we'd construct the new plan with the children.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb merged PR #19437:
URL: https://github.com/apache/datafusion/pull/19437
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2644641468
##########
datafusion-examples/examples/custom_data_source/adapter_serialization.rs:
##########
@@ -0,0 +1,527 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! See `main.rs` for how to run it.
+//!
+//! This example demonstrates how to use the `PhysicalExtensionCodec` trait's
+//! interception methods (`serialize_physical_plan` and `deserialize_physical_plan`)
+//! to implement custom serialization logic.
+//!
+//! The key insight is that `FileScanConfig::expr_adapter_factory` is NOT serialized by
+//! default. This example shows how to:
+//! 1. Detect plans with custom adapters during serialization
+//! 2. Wrap them as Extension nodes with JSON-serialized adapter metadata
+//! 3. Unwrap and restore the adapter during deserialization
+//!
+//! This demonstrates nested serialization (protobuf outer, JSON inner) and the power
+//! of the `PhysicalExtensionCodec` interception pattern. Both plan and expression
+//! serialization route through the codec, enabling interception at every node in the tree.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use arrow::array::record_batch;
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::assert_batches_eq;
+use datafusion::common::Result;
+use datafusion::common::not_impl_err;
+use datafusion::datasource::listing::{
+ ListingTable, ListingTableConfig, ListingTableConfigExt, ListingTableUrl,
+};
+use datafusion::datasource::physical_plan::{FileScanConfig, FileScanConfigBuilder};
+use datafusion::datasource::source::DataSourceExec;
+use datafusion::execution::TaskContext;
+use datafusion::execution::context::SessionContext;
+use datafusion::execution::object_store::ObjectStoreUrl;
+use datafusion::parquet::arrow::ArrowWriter;
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::ExecutionPlan;
+use datafusion::prelude::SessionConfig;
+use datafusion_physical_expr_adapter::{
+ DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
+};
+use datafusion_proto::bytes::{
+ physical_plan_from_bytes_with_extension_codec,
+ physical_plan_to_bytes_with_extension_codec,
+};
+use datafusion_proto::physical_plan::{
+ AsExecutionPlan, PhysicalExtensionCodec, PhysicalExtensionProtoCodec,
+};
+use datafusion_proto::protobuf::{
+ PhysicalExprNode, PhysicalExtensionNode, PhysicalPlanNode,
+ physical_plan_node::PhysicalPlanType,
+};
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, PutPayload};
+use prost::Message;
+use serde::{Deserialize, Serialize};
+
+/// Example showing how to preserve custom adapter information during plan serialization.
+///
+/// This demonstrates:
+/// 1. Creating a custom PhysicalExprAdapter with metadata
+/// 2. Using PhysicalExtensionCodec to intercept serialization
+/// 3. Wrapping adapter info as Extension nodes
+/// 4. Restoring adapters during deserialization
+pub async fn adapter_serialization() -> Result<()> {
+ println!("=== PhysicalExprAdapter Serialization Example ===\n");
+
+ // Step 1: Create sample Parquet data in memory
+ println!("Step 1: Creating sample Parquet data...");
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let batch = record_batch!(("id", Int32, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))?;
+ let path = Path::from("data.parquet");
+ write_parquet(&store, &path, &batch).await?;
+
+ // Step 2: Set up session with custom adapter
+ println!("Step 2: Setting up session with custom adapter...");
+ let logical_schema =
+ Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+
+ let mut cfg = SessionConfig::new();
+ cfg.options_mut().execution.parquet.pushdown_filters = true;
+ let ctx = SessionContext::new_with_config(cfg);
+ ctx.runtime_env().register_object_store(
+ ObjectStoreUrl::parse("memory://")?.as_ref(),
+ Arc::clone(&store),
+ );
+
+ // Create a table with our custom MetadataAdapterFactory
+ let adapter_factory = Arc::new(MetadataAdapterFactory::new("v1"));
+ let listing_config =
+ ListingTableConfig::new(ListingTableUrl::parse("memory:///data.parquet")?)
+ .infer_options(&ctx.state())
+ .await?
+ .with_schema(logical_schema)
+ .with_expr_adapter_factory(
+ Arc::clone(&adapter_factory) as Arc<dyn PhysicalExprAdapterFactory>
+ );
+ let table = ListingTable::try_new(listing_config)?;
+ ctx.register_table("my_table", Arc::new(table))?;
+
+ // Step 3: Create physical plan with filter
+ println!("Step 3: Creating physical plan with filter...");
+ let df = ctx.sql("SELECT * FROM my_table WHERE id > 5").await?;
+ let original_plan = df.create_physical_plan().await?;
+
+ // Verify adapter is present in original plan
+ let has_adapter_before = verify_adapter_in_plan(&original_plan, "original");
+ println!(" Original plan has adapter: {has_adapter_before}");
+
+ // Step 4: Serialize with our custom codec
+ println!("\nStep 4: Serializing plan with AdapterPreservingCodec...");
+ let codec = AdapterPreservingCodec;
+ let bytes = physical_plan_to_bytes_with_extension_codec(
+ Arc::clone(&original_plan),
+ &codec,
+ &codec,
+ )?;
+ println!(" Serialized {} bytes", bytes.len());
+ println!(" (DataSourceExec with adapter was wrapped as PhysicalExtensionNode)");
+
+ // Step 5: Deserialize with our custom codec
+ println!("\nStep 5: Deserializing plan with AdapterPreservingCodec...");
+ let task_ctx = ctx.task_ctx();
+ let restored_plan =
+ physical_plan_from_bytes_with_extension_codec(&bytes, &task_ctx, &codec, &codec)?;
+
+ // Verify adapter is restored
+ let has_adapter_after = verify_adapter_in_plan(&restored_plan, "restored");
+ println!(" Restored plan has adapter: {has_adapter_after}");
+
+ // Step 6: Execute and compare results
+ println!("\nStep 6: Executing plans and comparing results...");
+ let original_results =
+ datafusion::physical_plan::collect(Arc::clone(&original_plan), task_ctx.clone())
+ .await?;
+ let restored_results =
+ datafusion::physical_plan::collect(restored_plan, task_ctx).await?;
+
+ #[rustfmt::skip]
+ let expected = [
+ "+----+",
+ "| id |",
+ "+----+",
+ "| 6 |",
+ "| 7 |",
+ "| 8 |",
+ "| 9 |",
+ "| 10 |",
+ "+----+",
+ ];
+
+ println!("\n Original plan results:");
+ arrow::util::pretty::print_batches(&original_results)?;
+ assert_batches_eq!(expected, &original_results);
+
+ println!("\n Restored plan results:");
+ arrow::util::pretty::print_batches(&restored_results)?;
+ assert_batches_eq!(expected, &restored_results);
+
+ println!("\n=== Example Complete! ===");
+ println!("Key takeaways:");
+ println!(
+ " 1. PhysicalExtensionCodec provides serialize_physical_plan/deserialize_physical_plan hooks"
+ );
+ println!(" 2. Custom metadata can be wrapped as PhysicalExtensionNode");
+ println!(" 3. Nested serialization (protobuf + JSON) works seamlessly");
+ println!(
+ " 4. Both plans produce identical results despite serialization round-trip"
+ );
+ println!(" 5. Adapters are fully preserved through the serialization round-trip");
+
+ Ok(())
+}
+
+// ============================================================================
+// MetadataAdapter - A simple custom adapter with a tag
+// ============================================================================
+
+/// A custom PhysicalExprAdapter that wraps another adapter.
+/// The tag metadata is stored in the factory, not the adapter itself.
+#[derive(Debug)]
+struct MetadataAdapter {
+ inner: Arc<dyn PhysicalExprAdapter>,
+}
+
+impl PhysicalExprAdapter for MetadataAdapter {
+ fn rewrite(&self, expr: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
+ // Simply delegate to inner adapter
+ self.inner.rewrite(expr)
+ }
+}
+
+// ============================================================================
+// MetadataAdapterFactory - Factory for creating MetadataAdapter instances
+// ============================================================================
+
+/// Factory for creating MetadataAdapter instances.
+/// The tag is stored in the factory and extracted via Debug formatting in `extract_adapter_tag`.
+#[derive(Debug)]
+struct MetadataAdapterFactory {
+ // Note: This field is read via Debug formatting in `extract_adapter_tag`.
+ // Rust's dead code analysis doesn't recognize Debug-based field access.
+ // In PR #19234, this field is used by `with_partition_values`, but that method
+ // doesn't exist in upstream DataFusion's PhysicalExprAdapter trait.
+ #[allow(dead_code)]
+ tag: String,
+}
+
+impl MetadataAdapterFactory {
+ fn new(tag: impl Into<String>) -> Self {
+ Self { tag: tag.into() }
+ }
+}
+
+impl PhysicalExprAdapterFactory for MetadataAdapterFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> Arc<dyn PhysicalExprAdapter> {
+ let inner = DefaultPhysicalExprAdapterFactory
+ .create(logical_file_schema, physical_file_schema);
+ Arc::new(MetadataAdapter { inner })
+ }
+}
+
+// ============================================================================
+// AdapterPreservingCodec - Custom codec that preserves adapters
+// ============================================================================
+
+/// Extension payload structure for serializing adapter info
+#[derive(Serialize, Deserialize)]
+struct ExtensionPayload {
+ /// Marker to identify this is our custom extension
+ marker: String,
+ /// JSON-serialized adapter metadata
+ adapter_metadata: AdapterMetadata,
+ /// Protobuf-serialized inner DataSourceExec (without adapter)
+ inner_plan_bytes: Vec<u8>,
+}
+
+/// Metadata about the adapter to recreate it during deserialization
+#[derive(Serialize, Deserialize)]
+struct AdapterMetadata {
+ /// The adapter tag (e.g., "v1")
+ tag: String,
+}
+
+const EXTENSION_MARKER: &str = "adapter_preserving_extension_v1";
+
+/// A codec that intercepts serialization to preserve adapter information.
+#[derive(Debug)]
+struct AdapterPreservingCodec;
+
+impl PhysicalExtensionCodec for AdapterPreservingCodec {
+ // Required method: decode custom extension nodes
+ fn try_decode(
+ &self,
+ buf: &[u8],
+ _inputs: &[Arc<dyn ExecutionPlan>],
Review Comment:
Curious if we could use `inputs` here for instead of making that go out of band and hence "breaking" the serialization tree? I.e. have a custom node but have everything below it be "standard".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3770472953
It turns out datafusion-python is using `PhysicalPlanNode::try_from_physical_plan()` and others so we do need an entry in the upgrade guide.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2724225496
##########
docs/source/library-user-guide/upgrading.md:
##########
@@ -154,6 +154,48 @@ The builder pattern is more efficient as it computes properties once during `bui
Note: `with_default_selectivity()` is not deprecated as it simply updates a field value and does not require the overhead of the builder pattern.
+### Protobuf conversion trait added
+
+A new trait, `PhysicalProtoConverterExtension`, has been added to the `datafusion-proto`
+crate. This is used for controlling the process of conversion of physical plans and
+expressions to and from their protobuf equivalents. The methods for conversion now
+require an additional parameter. Most users can simply use the `DefaultPhysicalProtoConverter`
+implementation.
+
+For conversion from protobuf types to their physical types:
+
+**Before:**
+
+```rust,ignore
+let plan = proto_plan.try_into_physical_plan(ctx, codec)?;
+let expr = parse_physical_expr(proto_expr, ctx, input_schema, codec)?;
+```
+
+**After:**
+
+```rust,ignore
+let converter = DefaultPhysicalProtoConverter {};
+let plan = proto_plan.try_into_physical_plan(ctx, codec, &converter)?;
Review Comment:
@timsaucer if i recall correctly we could add
```rust
let plan = proto_plan.proto_plan.try_into_physical_plan_with_converter(ctx, codec, &converter)?;
```
and keep
```rust
proto_plan.try_into_physical_plan(ctx, codec)
```
where:
```
fn try_into_physical_plan_with_converter(&self, ctx, codec) {
let converter = DefaultPhysicalProtoConverter {};
self.proto_plan.try_into_physical_plan_with_converter_with_converter(ctx, codec, &converter)
}
```
That way we could keep change backward compatible and API simple as most of users will never need `Converter`
wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3806467978
I plan to merge this once CI passes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3735415585
Due to the size of this PR we could punt on the Logical side since we do not currently have any users asking for it and simply open an issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3770118424
> Ok @adriangb @milenkovicm @LiaCastaneda I think this is ready for review now. I'm really glad I added in a round trip unit test to verify the number of times we hit each method because it pointed out the same flaw @LiaCastaneda described.
Thanks so much for working on this Tim.
It unfortunately has conflicts which blocks CI from running. I'd be happy to resolve them but don't want to force push to your branch. Could you clear them up so CI can run?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3770108861
Ok @adriangb @milenkovicm @LiaCastaneda I think this is ready for review now. I'm really glad I added in a round trip unit test to verify the number of times we hit each method because it pointed out the same flaw @LiaCastaneda described.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3681956008
@milenkovicm @adriangb this is a proof of concept of an alternative, which I think is actually fairly clean. Basically the idea is that we have two portions of the proto work. There's one codec that turns bytes into physical expressions, plans, etc and vice versa. There a different codec that turns protobuf messages into physical expressions, plans, etc and vice versa. Here we just have two traits so the user can intersect at either point. Then these traits are used throughout the proto code. What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2706045697
##########
datafusion-examples/examples/proto/expression_deduplication.rs:
##########
@@ -0,0 +1,269 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! See `main.rs` for how to run it.
+//!
+//! This example demonstrates how to use the `PhysicalExtensionCodec` trait's
+//! interception methods to implement expression deduplication during deserialization.
+//!
+//! This pattern is inspired by PR #18192, which introduces expression caching
+//! to reduce memory usage when deserializing plans with duplicate expressions.
+//!
+//! The key insight is that identical expressions serialize to identical protobuf bytes.
+//! By caching deserialized expressions keyed by their protobuf bytes, we can:
+//! 1. Return the same Arc for duplicate expressions
+//! 2. Reduce memory allocation during deserialization
+//! 3. Enable downstream optimizations that rely on Arc pointer equality
+//!
+//! This demonstrates the decorator pattern enabled by the `PhysicalExtensionCodec` trait,
+//! where all expression serialization/deserialization routes through the codec methods.
+
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::sync::{Arc, RwLock};
+
+use arrow::datatypes::{DataType, Field, Schema};
+use datafusion::common::Result;
+use datafusion::execution::TaskContext;
+use datafusion::logical_expr::Operator;
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::ExecutionPlan;
+use datafusion::physical_plan::expressions::{BinaryExpr, col};
+use datafusion::physical_plan::filter::FilterExec;
+use datafusion::physical_plan::placeholder_row::PlaceholderRowExec;
+use datafusion::prelude::SessionContext;
+use datafusion_proto::physical_plan::from_proto::parse_physical_expr;
+use datafusion_proto::physical_plan::to_proto::serialize_physical_expr;
+use datafusion_proto::physical_plan::{
+ AsExecutionPlan, DefaultPhysicalExtensionCodec, PhysicalExtensionCodec,
+ PhysicalProtoConverterExtension,
+};
+use datafusion_proto::protobuf::{PhysicalExprNode, PhysicalPlanNode};
+use prost::Message;
+
+/// Example showing how to implement expression deduplication using the codec decorator pattern.
+///
+/// This demonstrates:
+/// 1. Creating a CachingCodec that caches expressions by their protobuf bytes
+/// 2. Intercepting deserialization to return cached Arcs for duplicate expressions
+/// 3. Verifying that duplicate expressions share the same Arc after deserialization
+///
+/// Deduplication is keyed by the protobuf bytes representing the expression,
+/// in reality deduplication could be done based on e.g. the pointer address of the
+/// serialized expression in memory, but this is simpler to demonstrate.
+///
+/// In this case our expression is trivial and just for demonstration purposes.
+/// In real scenarios, expressions can be much more complex, e.g. a large InList
+/// expression could be megabytes in size, so deduplication can save significant memory
+/// in addition to more correctly representing the original plan structure.
+pub async fn expression_deduplication() -> Result<()> {
+ println!("=== Expression Deduplication Example ===\n");
+
+ // Create a schema for our test expressions
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Boolean, false)]));
+
+ // Step 1: Create expressions with duplicates
+ println!("Step 1: Creating expressions with duplicates...");
+
+ // Create expression: col("a")
+ let a = col("a", &schema)?;
+
+ // Create a clone to show duplicates
+ let a_clone = Arc::clone(&a);
+
+ // Combine: a OR a_clone
+ let combined_expr =
+ Arc::new(BinaryExpr::new(a, Operator::Or, a_clone)) as Arc<dyn PhysicalExpr>;
+ println!(" Created expression: a OR a with duplicates");
+ println!(" Note: a appears twice in the expression tree\n");
+ // Step 2: Create a filter plan with this expression
+ println!("Step 2: Creating physical plan with the expression...");
+
+ let input = Arc::new(PlaceholderRowExec::new(Arc::clone(&schema)));
+ let filter_plan: Arc<dyn ExecutionPlan> =
+ Arc::new(FilterExec::try_new(combined_expr, input)?);
+
+ println!(" Created FilterExec with duplicate sub-expressions\n");
+
+ // Step 3: Serialize with the caching codec
+ println!("Step 3: Serializing plan...");
+
+ let extension_codec = DefaultPhysicalExtensionCodec {};
+ let caching_codec = CachingCodec::new();
+ let proto = caching_codec.execution_plan_to_proto(&filter_plan, &extension_codec)?;
+
+ // Serialize to bytes
+ let mut bytes = Vec::new();
+ proto.encode(&mut bytes).unwrap();
+ println!(" Serialized plan to {} bytes\n", bytes.len());
+
+ // Step 4: Deserialize with the caching codec
+ println!("Step 4: Deserializing plan with CachingCodec...");
+
+ let ctx = SessionContext::new();
+ let deserialized_plan = proto.try_into_physical_plan(
+ &ctx.task_ctx(),
+ &extension_codec,
+ &caching_codec,
+ )?;
+
+ // Step 5: check that we deduplicated expressions
+ println!("Step 5: Checking for deduplicated expressions...");
+ let Some(filter_exec) = deserialized_plan.as_any().downcast_ref::<FilterExec>()
+ else {
+ panic!("Deserialized plan is not a FilterExec");
+ };
+ let predicate = Arc::clone(filter_exec.predicate());
+ let binary_expr = predicate
+ .as_any()
+ .downcast_ref::<BinaryExpr>()
+ .expect("Predicate is not a BinaryExpr");
+ let left = &binary_expr.left();
+ let right = &binary_expr.right();
+ // Check if left and right point to the same Arc
+ let deduplicated = Arc::ptr_eq(left, right);
+ if deduplicated {
+ println!(" Success: Duplicate expressions were deduplicated!");
+ println!(
+ " Cache Stats: hits={}, misses={}",
+ caching_codec.stats.read().unwrap().cache_hits,
+ caching_codec.stats.read().unwrap().cache_misses,
+ );
+ } else {
+ println!(" Failure: Duplicate expressions were NOT deduplicated.");
+ }
+
+ Ok(())
+}
+
+// ============================================================================
+// CachingCodec - Implements expression deduplication
+// ============================================================================
+
+/// Statistics for cache performance monitoring
+#[derive(Debug, Default)]
+struct CacheStats {
+ cache_hits: usize,
+ cache_misses: usize,
+}
+
+/// A codec that caches deserialized expressions to enable deduplication.
+///
+/// When deserializing, if we've already seen the same protobuf bytes,
+/// we return the cached Arc instead of creating a new allocation.
+#[derive(Debug, Default)]
+struct CachingCodec {
+ /// Cache mapping protobuf bytes -> deserialized expression
+ expr_cache: RwLock<HashMap<Vec<u8>, Arc<dyn PhysicalExpr>>>,
+ /// Statistics for demonstration
+ stats: RwLock<CacheStats>,
+}
+
+impl CachingCodec {
+ fn new() -> Self {
+ Self::default()
+ }
+}
+
+impl PhysicalExtensionCodec for CachingCodec {
+ // Required: decode custom extension nodes
+ fn try_decode(
+ &self,
+ _buf: &[u8],
+ _inputs: &[Arc<dyn ExecutionPlan>],
+ _ctx: &TaskContext,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ datafusion::common::not_impl_err!("No custom extension nodes")
+ }
+
+ // Required: encode custom execution plans
+ fn try_encode(
+ &self,
+ _node: Arc<dyn ExecutionPlan>,
+ _buf: &mut Vec<u8>,
+ ) -> Result<()> {
+ datafusion::common::not_impl_err!("No custom extension nodes")
+ }
+}
+
+impl PhysicalProtoConverterExtension for CachingCodec {
+ fn proto_to_execution_plan(
+ &self,
+ _ctx: &TaskContext,
+ _extension_codec: &dyn PhysicalExtensionCodec,
+ _proto: &PhysicalPlanNode,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ todo!()
+ }
+
+ fn execution_plan_to_proto(
+ &self,
+ plan: &Arc<dyn ExecutionPlan>,
+ extension_codec: &dyn PhysicalExtensionCodec,
+ ) -> Result<PhysicalPlanNode> {
Review Comment:
Actually, you've correctly identified a bug that I'm correcting right now as I'm writing unit tests. All of the places we are calling `proto_plan.try_into_physical_plan` should instead be calling the proto converter's `proto_to_execution_plan` or else we only get one entry point.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2653048980
##########
datafusion-examples/examples/custom_data_source/adapter_serialization.rs:
##########
@@ -0,0 +1,527 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! See `main.rs` for how to run it.
+//!
+//! This example demonstrates how to use the `PhysicalExtensionCodec` trait's
+//! interception methods (`serialize_physical_plan` and `deserialize_physical_plan`)
+//! to implement custom serialization logic.
+//!
+//! The key insight is that `FileScanConfig::expr_adapter_factory` is NOT serialized by
+//! default. This example shows how to:
+//! 1. Detect plans with custom adapters during serialization
+//! 2. Wrap them as Extension nodes with JSON-serialized adapter metadata
+//! 3. Unwrap and restore the adapter during deserialization
+//!
+//! This demonstrates nested serialization (protobuf outer, JSON inner) and the power
+//! of the `PhysicalExtensionCodec` interception pattern. Both plan and expression
+//! serialization route through the codec, enabling interception at every node in the tree.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use arrow::array::record_batch;
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::assert_batches_eq;
+use datafusion::common::Result;
+use datafusion::common::not_impl_err;
+use datafusion::datasource::listing::{
+ ListingTable, ListingTableConfig, ListingTableConfigExt, ListingTableUrl,
+};
+use datafusion::datasource::physical_plan::{FileScanConfig, FileScanConfigBuilder};
+use datafusion::datasource::source::DataSourceExec;
+use datafusion::execution::TaskContext;
+use datafusion::execution::context::SessionContext;
+use datafusion::execution::object_store::ObjectStoreUrl;
+use datafusion::parquet::arrow::ArrowWriter;
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::ExecutionPlan;
+use datafusion::prelude::SessionConfig;
+use datafusion_physical_expr_adapter::{
+ DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
+};
+use datafusion_proto::bytes::{
+ physical_plan_from_bytes_with_extension_codec,
+ physical_plan_to_bytes_with_extension_codec,
+};
+use datafusion_proto::physical_plan::{
+ AsExecutionPlan, PhysicalExtensionCodec, PhysicalExtensionProtoCodec,
+};
+use datafusion_proto::protobuf::{
+ PhysicalExprNode, PhysicalExtensionNode, PhysicalPlanNode,
+ physical_plan_node::PhysicalPlanType,
+};
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, PutPayload};
+use prost::Message;
+use serde::{Deserialize, Serialize};
+
+/// Example showing how to preserve custom adapter information during plan serialization.
+///
+/// This demonstrates:
+/// 1. Creating a custom PhysicalExprAdapter with metadata
+/// 2. Using PhysicalExtensionCodec to intercept serialization
+/// 3. Wrapping adapter info as Extension nodes
+/// 4. Restoring adapters during deserialization
+pub async fn adapter_serialization() -> Result<()> {
+ println!("=== PhysicalExprAdapter Serialization Example ===\n");
+
+ // Step 1: Create sample Parquet data in memory
+ println!("Step 1: Creating sample Parquet data...");
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let batch = record_batch!(("id", Int32, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))?;
+ let path = Path::from("data.parquet");
+ write_parquet(&store, &path, &batch).await?;
+
+ // Step 2: Set up session with custom adapter
+ println!("Step 2: Setting up session with custom adapter...");
+ let logical_schema =
+ Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+
+ let mut cfg = SessionConfig::new();
+ cfg.options_mut().execution.parquet.pushdown_filters = true;
+ let ctx = SessionContext::new_with_config(cfg);
+ ctx.runtime_env().register_object_store(
+ ObjectStoreUrl::parse("memory://")?.as_ref(),
+ Arc::clone(&store),
+ );
+
+ // Create a table with our custom MetadataAdapterFactory
+ let adapter_factory = Arc::new(MetadataAdapterFactory::new("v1"));
+ let listing_config =
+ ListingTableConfig::new(ListingTableUrl::parse("memory:///data.parquet")?)
+ .infer_options(&ctx.state())
+ .await?
+ .with_schema(logical_schema)
+ .with_expr_adapter_factory(
+ Arc::clone(&adapter_factory) as Arc<dyn PhysicalExprAdapterFactory>
+ );
+ let table = ListingTable::try_new(listing_config)?;
+ ctx.register_table("my_table", Arc::new(table))?;
+
+ // Step 3: Create physical plan with filter
+ println!("Step 3: Creating physical plan with filter...");
+ let df = ctx.sql("SELECT * FROM my_table WHERE id > 5").await?;
+ let original_plan = df.create_physical_plan().await?;
+
+ // Verify adapter is present in original plan
+ let has_adapter_before = verify_adapter_in_plan(&original_plan, "original");
+ println!(" Original plan has adapter: {has_adapter_before}");
+
+ // Step 4: Serialize with our custom codec
+ println!("\nStep 4: Serializing plan with AdapterPreservingCodec...");
+ let codec = AdapterPreservingCodec;
+ let bytes = physical_plan_to_bytes_with_extension_codec(
+ Arc::clone(&original_plan),
+ &codec,
+ &codec,
+ )?;
+ println!(" Serialized {} bytes", bytes.len());
+ println!(" (DataSourceExec with adapter was wrapped as PhysicalExtensionNode)");
+
+ // Step 5: Deserialize with our custom codec
+ println!("\nStep 5: Deserializing plan with AdapterPreservingCodec...");
+ let task_ctx = ctx.task_ctx();
+ let restored_plan =
+ physical_plan_from_bytes_with_extension_codec(&bytes, &task_ctx, &codec, &codec)?;
+
+ // Verify adapter is restored
+ let has_adapter_after = verify_adapter_in_plan(&restored_plan, "restored");
+ println!(" Restored plan has adapter: {has_adapter_after}");
+
+ // Step 6: Execute and compare results
+ println!("\nStep 6: Executing plans and comparing results...");
+ let original_results =
+ datafusion::physical_plan::collect(Arc::clone(&original_plan), task_ctx.clone())
+ .await?;
+ let restored_results =
+ datafusion::physical_plan::collect(restored_plan, task_ctx).await?;
+
+ #[rustfmt::skip]
+ let expected = [
+ "+----+",
+ "| id |",
+ "+----+",
+ "| 6 |",
+ "| 7 |",
+ "| 8 |",
+ "| 9 |",
+ "| 10 |",
+ "+----+",
+ ];
+
+ println!("\n Original plan results:");
+ arrow::util::pretty::print_batches(&original_results)?;
+ assert_batches_eq!(expected, &original_results);
+
+ println!("\n Restored plan results:");
+ arrow::util::pretty::print_batches(&restored_results)?;
+ assert_batches_eq!(expected, &restored_results);
+
+ println!("\n=== Example Complete! ===");
+ println!("Key takeaways:");
+ println!(
+ " 1. PhysicalExtensionCodec provides serialize_physical_plan/deserialize_physical_plan hooks"
+ );
+ println!(" 2. Custom metadata can be wrapped as PhysicalExtensionNode");
+ println!(" 3. Nested serialization (protobuf + JSON) works seamlessly");
+ println!(
+ " 4. Both plans produce identical results despite serialization round-trip"
+ );
+ println!(" 5. Adapters are fully preserved through the serialization round-trip");
+
+ Ok(())
+}
+
+// ============================================================================
+// MetadataAdapter - A simple custom adapter with a tag
+// ============================================================================
+
+/// A custom PhysicalExprAdapter that wraps another adapter.
+/// The tag metadata is stored in the factory, not the adapter itself.
+#[derive(Debug)]
+struct MetadataAdapter {
+ inner: Arc<dyn PhysicalExprAdapter>,
+}
+
+impl PhysicalExprAdapter for MetadataAdapter {
+ fn rewrite(&self, expr: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
+ // Simply delegate to inner adapter
+ self.inner.rewrite(expr)
+ }
+}
+
+// ============================================================================
+// MetadataAdapterFactory - Factory for creating MetadataAdapter instances
+// ============================================================================
+
+/// Factory for creating MetadataAdapter instances.
+/// The tag is stored in the factory and extracted via Debug formatting in `extract_adapter_tag`.
+#[derive(Debug)]
+struct MetadataAdapterFactory {
+ // Note: This field is read via Debug formatting in `extract_adapter_tag`.
+ // Rust's dead code analysis doesn't recognize Debug-based field access.
+ // In PR #19234, this field is used by `with_partition_values`, but that method
+ // doesn't exist in upstream DataFusion's PhysicalExprAdapter trait.
+ #[allow(dead_code)]
+ tag: String,
+}
+
+impl MetadataAdapterFactory {
+ fn new(tag: impl Into<String>) -> Self {
+ Self { tag: tag.into() }
+ }
+}
+
+impl PhysicalExprAdapterFactory for MetadataAdapterFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> Arc<dyn PhysicalExprAdapter> {
+ let inner = DefaultPhysicalExprAdapterFactory
+ .create(logical_file_schema, physical_file_schema);
+ Arc::new(MetadataAdapter { inner })
+ }
+}
+
+// ============================================================================
+// AdapterPreservingCodec - Custom codec that preserves adapters
+// ============================================================================
+
+/// Extension payload structure for serializing adapter info
+#[derive(Serialize, Deserialize)]
+struct ExtensionPayload {
+ /// Marker to identify this is our custom extension
+ marker: String,
+ /// JSON-serialized adapter metadata
+ adapter_metadata: AdapterMetadata,
+ /// Protobuf-serialized inner DataSourceExec (without adapter)
+ inner_plan_bytes: Vec<u8>,
+}
+
+/// Metadata about the adapter to recreate it during deserialization
+#[derive(Serialize, Deserialize)]
+struct AdapterMetadata {
+ /// The adapter tag (e.g., "v1")
+ tag: String,
+}
+
+const EXTENSION_MARKER: &str = "adapter_preserving_extension_v1";
+
+/// A codec that intercepts serialization to preserve adapter information.
+#[derive(Debug)]
+struct AdapterPreservingCodec;
+
+impl PhysicalExtensionCodec for AdapterPreservingCodec {
+ // Required method: decode custom extension nodes
+ fn try_decode(
+ &self,
+ buf: &[u8],
+ _inputs: &[Arc<dyn ExecutionPlan>],
Review Comment:
I don't understand this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3681989677
I'm away this week, will have a look ASAP
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "LiaCastaneda (via GitHub)" <gi...@apache.org>.
LiaCastaneda commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2690810007
##########
datafusion-examples/examples/custom_data_source/adapter_serialization.rs:
##########
@@ -0,0 +1,536 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! See `main.rs` for how to run it.
+//!
+//! This example demonstrates how to use the `PhysicalExtensionCodec` trait's
+//! interception methods (`serialize_physical_plan` and `deserialize_physical_plan`)
+//! to implement custom serialization logic.
+//!
+//! The key insight is that `FileScanConfig::expr_adapter_factory` is NOT serialized by
+//! default. This example shows how to:
+//! 1. Detect plans with custom adapters during serialization
+//! 2. Wrap them as Extension nodes with JSON-serialized adapter metadata
+//! 3. Unwrap and restore the adapter during deserialization
+//!
+//! This demonstrates nested serialization (protobuf outer, JSON inner) and the power
+//! of the `PhysicalExtensionCodec` interception pattern. Both plan and expression
+//! serialization route through the codec, enabling interception at every node in the tree.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use arrow::array::record_batch;
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::assert_batches_eq;
+use datafusion::common::{Result, not_impl_err};
+use datafusion::datasource::listing::{
+ ListingTable, ListingTableConfig, ListingTableConfigExt, ListingTableUrl,
+};
+use datafusion::datasource::physical_plan::{FileScanConfig, FileScanConfigBuilder};
+use datafusion::datasource::source::DataSourceExec;
+use datafusion::execution::TaskContext;
+use datafusion::execution::context::SessionContext;
+use datafusion::execution::object_store::ObjectStoreUrl;
+use datafusion::parquet::arrow::ArrowWriter;
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::ExecutionPlan;
+use datafusion::prelude::SessionConfig;
+use datafusion_physical_expr_adapter::{
+ DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
+};
+use datafusion_proto::bytes::{
+ physical_plan_from_bytes_with_proto_converter,
+ physical_plan_to_bytes_with_proto_converter,
+};
+use datafusion_proto::physical_plan::from_proto::parse_physical_expr;
+use datafusion_proto::physical_plan::to_proto::serialize_physical_expr;
+use datafusion_proto::physical_plan::{
+ AsExecutionPlan, PhysicalExtensionCodec, PhysicalProtoConverterExtension,
+};
+use datafusion_proto::protobuf::physical_plan_node::PhysicalPlanType;
+use datafusion_proto::protobuf::{
+ PhysicalExprNode, PhysicalExtensionNode, PhysicalPlanNode,
+};
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, PutPayload};
+use prost::Message;
+use serde::{Deserialize, Serialize};
+
+/// Example showing how to preserve custom adapter information during plan serialization.
+///
+/// This demonstrates:
+/// 1. Creating a custom PhysicalExprAdapter with metadata
+/// 2. Using PhysicalExtensionCodec to intercept serialization
+/// 3. Wrapping adapter info as Extension nodes
+/// 4. Restoring adapters during deserialization
+pub async fn adapter_serialization() -> Result<()> {
+ println!("=== PhysicalExprAdapter Serialization Example ===\n");
+
+ // Step 1: Create sample Parquet data in memory
+ println!("Step 1: Creating sample Parquet data...");
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let batch = record_batch!(("id", Int32, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))?;
+ let path = Path::from("data.parquet");
+ write_parquet(&store, &path, &batch).await?;
+
+ // Step 2: Set up session with custom adapter
+ println!("Step 2: Setting up session with custom adapter...");
+ let logical_schema =
+ Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+
+ let mut cfg = SessionConfig::new();
+ cfg.options_mut().execution.parquet.pushdown_filters = true;
+ let ctx = SessionContext::new_with_config(cfg);
+ ctx.runtime_env().register_object_store(
+ ObjectStoreUrl::parse("memory://")?.as_ref(),
+ Arc::clone(&store),
+ );
+
+ // Create a table with our custom MetadataAdapterFactory
+ let adapter_factory = Arc::new(MetadataAdapterFactory::new("v1"));
+ let listing_config =
+ ListingTableConfig::new(ListingTableUrl::parse("memory:///data.parquet")?)
+ .infer_options(&ctx.state())
+ .await?
+ .with_schema(logical_schema)
+ .with_expr_adapter_factory(
+ Arc::clone(&adapter_factory) as Arc<dyn PhysicalExprAdapterFactory>
+ );
+ let table = ListingTable::try_new(listing_config)?;
+ ctx.register_table("my_table", Arc::new(table))?;
+
+ // Step 3: Create physical plan with filter
+ println!("Step 3: Creating physical plan with filter...");
+ let df = ctx.sql("SELECT * FROM my_table WHERE id > 5").await?;
+ let original_plan = df.create_physical_plan().await?;
+
+ // Verify adapter is present in original plan
+ let has_adapter_before = verify_adapter_in_plan(&original_plan, "original");
+ println!(" Original plan has adapter: {has_adapter_before}");
+
+ // Step 4: Serialize with our custom codec
+ println!("\nStep 4: Serializing plan with AdapterPreservingCodec...");
+ let codec = AdapterPreservingCodec;
+ let bytes = physical_plan_to_bytes_with_proto_converter(
+ Arc::clone(&original_plan),
+ &codec,
+ &codec,
+ )?;
+ println!(" Serialized {} bytes", bytes.len());
+ println!(" (DataSourceExec with adapter was wrapped as PhysicalExtensionNode)");
+
+ // Step 5: Deserialize with our custom codec
+ println!("\nStep 5: Deserializing plan with AdapterPreservingCodec...");
+ let task_ctx = ctx.task_ctx();
+ let restored_plan =
+ physical_plan_from_bytes_with_proto_converter(&bytes, &task_ctx, &codec, &codec)?;
+
+ // Verify adapter is restored
+ let has_adapter_after = verify_adapter_in_plan(&restored_plan, "restored");
+ println!(" Restored plan has adapter: {has_adapter_after}");
+
+ // Step 6: Execute and compare results
+ println!("\nStep 6: Executing plans and comparing results...");
+ let original_results =
+ datafusion::physical_plan::collect(Arc::clone(&original_plan), task_ctx.clone())
+ .await?;
+ let restored_results =
+ datafusion::physical_plan::collect(restored_plan, task_ctx).await?;
+
+ #[rustfmt::skip]
+ let expected = [
+ "+----+",
+ "| id |",
+ "+----+",
+ "| 6 |",
+ "| 7 |",
+ "| 8 |",
+ "| 9 |",
+ "| 10 |",
+ "+----+",
+ ];
+
+ println!("\n Original plan results:");
+ arrow::util::pretty::print_batches(&original_results)?;
+ assert_batches_eq!(expected, &original_results);
+
+ println!("\n Restored plan results:");
+ arrow::util::pretty::print_batches(&restored_results)?;
+ assert_batches_eq!(expected, &restored_results);
+
+ println!("\n=== Example Complete! ===");
+ println!("Key takeaways:");
+ println!(
+ " 1. PhysicalExtensionCodec provides serialize_physical_plan/deserialize_physical_plan hooks"
+ );
+ println!(" 2. Custom metadata can be wrapped as PhysicalExtensionNode");
+ println!(" 3. Nested serialization (protobuf + JSON) works seamlessly");
+ println!(
+ " 4. Both plans produce identical results despite serialization round-trip"
+ );
+ println!(" 5. Adapters are fully preserved through the serialization round-trip");
+
+ Ok(())
+}
+
+// ============================================================================
+// MetadataAdapter - A simple custom adapter with a tag
+// ============================================================================
+
+/// A custom PhysicalExprAdapter that wraps another adapter.
+/// The tag metadata is stored in the factory, not the adapter itself.
+#[derive(Debug)]
+struct MetadataAdapter {
+ inner: Arc<dyn PhysicalExprAdapter>,
+}
+
+impl PhysicalExprAdapter for MetadataAdapter {
+ fn rewrite(&self, expr: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
+ // Simply delegate to inner adapter
+ self.inner.rewrite(expr)
+ }
+}
+
+// ============================================================================
+// MetadataAdapterFactory - Factory for creating MetadataAdapter instances
+// ============================================================================
+
+/// Factory for creating MetadataAdapter instances.
+/// The tag is stored in the factory and extracted via Debug formatting in `extract_adapter_tag`.
+#[derive(Debug)]
+struct MetadataAdapterFactory {
+ // Note: This field is read via Debug formatting in `extract_adapter_tag`.
+ // Rust's dead code analysis doesn't recognize Debug-based field access.
+ // In PR #19234, this field is used by `with_partition_values`, but that method
+ // doesn't exist in upstream DataFusion's PhysicalExprAdapter trait.
+ #[allow(dead_code)]
+ tag: String,
+}
+
+impl MetadataAdapterFactory {
+ fn new(tag: impl Into<String>) -> Self {
+ Self { tag: tag.into() }
+ }
+}
+
+impl PhysicalExprAdapterFactory for MetadataAdapterFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> Arc<dyn PhysicalExprAdapter> {
+ let inner = DefaultPhysicalExprAdapterFactory
+ .create(logical_file_schema, physical_file_schema);
+ Arc::new(MetadataAdapter { inner })
+ }
+}
+
+// ============================================================================
+// AdapterPreservingCodec - Custom codec that preserves adapters
+// ============================================================================
+
+/// Extension payload structure for serializing adapter info
+#[derive(Serialize, Deserialize)]
+struct ExtensionPayload {
+ /// Marker to identify this is our custom extension
+ marker: String,
+ /// JSON-serialized adapter metadata
+ adapter_metadata: AdapterMetadata,
+ /// Protobuf-serialized inner DataSourceExec (without adapter)
+ inner_plan_bytes: Vec<u8>,
+}
+
+/// Metadata about the adapter to recreate it during deserialization
+#[derive(Serialize, Deserialize)]
+struct AdapterMetadata {
+ /// The adapter tag (e.g., "v1")
+ tag: String,
+}
+
+const EXTENSION_MARKER: &str = "adapter_preserving_extension_v1";
+
+/// A codec that intercepts serialization to preserve adapter information.
+#[derive(Debug)]
+struct AdapterPreservingCodec;
+
+impl PhysicalExtensionCodec for AdapterPreservingCodec {
+ // Required method: decode custom extension nodes
+ fn try_decode(
+ &self,
+ buf: &[u8],
+ _inputs: &[Arc<dyn ExecutionPlan>],
+ ctx: &TaskContext,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Try to parse as our extension payload
+ if let Ok(payload) = serde_json::from_slice::<ExtensionPayload>(buf)
+ && payload.marker == EXTENSION_MARKER
+ {
+ // Decode the inner plan
+ let inner_proto = PhysicalPlanNode::decode(&payload.inner_plan_bytes[..])
+ .map_err(|e| {
+ datafusion::error::DataFusionError::Plan(format!(
+ "Failed to decode inner plan: {e}"
+ ))
+ })?;
+
+ // Deserialize the inner plan using default implementation
+ let inner_plan = inner_proto.try_into_physical_plan(ctx, self, self)?;
+
+ // Recreate the adapter factory
+ let adapter_factory = create_adapter_factory(&payload.adapter_metadata.tag);
+
+ // Inject adapter into the plan
+ return inject_adapter_into_plan(inner_plan, adapter_factory);
+ }
+
+ not_impl_err!("Unknown extension type")
+ }
+
+ // Required method: encode custom execution plans
+ fn try_encode(
+ &self,
+ _node: Arc<dyn ExecutionPlan>,
+ _buf: &mut Vec<u8>,
+ ) -> Result<()> {
+ // We don't need this for the example - we use serialize_physical_plan instead
+ not_impl_err!(
+ "try_encode not used - adapter wrapping happens in serialize_physical_plan"
+ )
+ }
+}
+
+impl PhysicalProtoConverterExtension for AdapterPreservingCodec {
+ fn execution_plan_to_proto(
+ &self,
+ plan: &Arc<dyn ExecutionPlan>,
+ extension_codec: &dyn PhysicalExtensionCodec,
+ ) -> Result<PhysicalPlanNode> {
+ // Check if this is a DataSourceExec with adapter
+ if let Some(exec) = plan.as_any().downcast_ref::<DataSourceExec>()
+ && let Some(config) =
+ exec.data_source().as_any().downcast_ref::<FileScanConfig>()
+ && let Some(adapter_factory) = &config.expr_adapter_factory
+ && let Some(tag) = extract_adapter_tag(adapter_factory.as_ref())
+ {
+ // Try to extract our MetadataAdapterFactory's tag
+ println!(" [Serialize] Found DataSourceExec with adapter tag: {tag}");
+
+ // 1. Create adapter metadata
+ let adapter_metadata = AdapterMetadata { tag };
+
+ // 2. Create a copy of the config without the adapter
+ let config_without_adapter = rebuild_config_without_adapter(config);
Review Comment:
If the user skip this step, would it be possible to enter an infitite recursive loop since we are passing the same `AdapterPreservingCodec` when calling `try_from_physical_plan` below?
try_from_physical_plan (wth adapter) -> execution_plan_to_proto (with adapter) -> try_from_physical_plan (with adapter) -> ....
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2724225496
##########
docs/source/library-user-guide/upgrading.md:
##########
@@ -154,6 +154,48 @@ The builder pattern is more efficient as it computes properties once during `bui
Note: `with_default_selectivity()` is not deprecated as it simply updates a field value and does not require the overhead of the builder pattern.
+### Protobuf conversion trait added
+
+A new trait, `PhysicalProtoConverterExtension`, has been added to the `datafusion-proto`
+crate. This is used for controlling the process of conversion of physical plans and
+expressions to and from their protobuf equivalents. The methods for conversion now
+require an additional parameter. Most users can simply use the `DefaultPhysicalProtoConverter`
+implementation.
+
+For conversion from protobuf types to their physical types:
+
+**Before:**
+
+```rust,ignore
+let plan = proto_plan.try_into_physical_plan(ctx, codec)?;
+let expr = parse_physical_expr(proto_expr, ctx, input_schema, codec)?;
+```
+
+**After:**
+
+```rust,ignore
+let converter = DefaultPhysicalProtoConverter {};
+let plan = proto_plan.try_into_physical_plan(ctx, codec, &converter)?;
Review Comment:
@timsaucer if i recall correctly we could add
```rust
let plan = proto_plan.proto_plan.try_into_physical_plan_with_converter(ctx, codec, &converter)?;
```
and keep
```rust
proto_plan.try_into_physical_plan(ctx, codec)
```
where:
```
fn try_into_physical_plan_with_converter(&self, ctx, codec) {
let converter = DefaultPhysicalProtoConverter {};
self.try_into_physical_plan(ctx, codec, &converter)
}
```
That way we could keep change backward compatible and API simple as most of users will never need `Converter`
wdyt?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2724262264
##########
docs/source/library-user-guide/upgrading.md:
##########
@@ -154,6 +154,48 @@ The builder pattern is more efficient as it computes properties once during `bui
Note: `with_default_selectivity()` is not deprecated as it simply updates a field value and does not require the overhead of the builder pattern.
+### Protobuf conversion trait added
+
+A new trait, `PhysicalProtoConverterExtension`, has been added to the `datafusion-proto`
+crate. This is used for controlling the process of conversion of physical plans and
+expressions to and from their protobuf equivalents. The methods for conversion now
+require an additional parameter. Most users can simply use the `DefaultPhysicalProtoConverter`
+implementation.
+
+For conversion from protobuf types to their physical types:
+
+**Before:**
+
+```rust,ignore
+let plan = proto_plan.try_into_physical_plan(ctx, codec)?;
+let expr = parse_physical_expr(proto_expr, ctx, input_schema, codec)?;
+```
+
+**After:**
+
+```rust,ignore
+let converter = DefaultPhysicalProtoConverter {};
+let plan = proto_plan.try_into_physical_plan(ctx, codec, &converter)?;
Review Comment:
Good idea.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3735395815
@adriangb I put in a little time this morning on tightening this up. I still want to apply a similar approach on the logical side, but I think this is now achieving what you wanted. I hope to get some time to do the migration guide and unit tests, but it's not my top priority right now so it may take a bit. If you have someone who could step in, I'd be happy to hand it off. Otherwise I'll keep working on it when I can find some time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3794644755
Upgrade instructions added. Ready for review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2651911765
##########
datafusion/proto/src/physical_plan/mod.rs:
##########
@@ -3418,6 +3721,44 @@ struct DataEncoderTuple {
pub blob: Vec<u8>,
}
+pub struct DefaultPhysicalExtensionProtoCodec;
+impl PhysicalExtensionProtoCodec for DefaultPhysicalExtensionProtoCodec {
+ fn proto_to_execution_plan(
+ &self,
+ ctx: &TaskContext,
+ extension_codec: &dyn PhysicalExtensionCodec,
+ proto: &protobuf::PhysicalPlanNode,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ proto.try_into_physical_plan(ctx, extension_codec, self)
+ }
+
+ // TODO(tsaucer) original variant had these as &mut self instead of &self
Review Comment:
I think `&self` is fine - we can always use interior mutability if needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2655569181
##########
datafusion-examples/examples/custom_data_source/adapter_serialization.rs:
##########
@@ -0,0 +1,527 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! See `main.rs` for how to run it.
+//!
+//! This example demonstrates how to use the `PhysicalExtensionCodec` trait's
+//! interception methods (`serialize_physical_plan` and `deserialize_physical_plan`)
+//! to implement custom serialization logic.
+//!
+//! The key insight is that `FileScanConfig::expr_adapter_factory` is NOT serialized by
+//! default. This example shows how to:
+//! 1. Detect plans with custom adapters during serialization
+//! 2. Wrap them as Extension nodes with JSON-serialized adapter metadata
+//! 3. Unwrap and restore the adapter during deserialization
+//!
+//! This demonstrates nested serialization (protobuf outer, JSON inner) and the power
+//! of the `PhysicalExtensionCodec` interception pattern. Both plan and expression
+//! serialization route through the codec, enabling interception at every node in the tree.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use arrow::array::record_batch;
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::assert_batches_eq;
+use datafusion::common::Result;
+use datafusion::common::not_impl_err;
+use datafusion::datasource::listing::{
+ ListingTable, ListingTableConfig, ListingTableConfigExt, ListingTableUrl,
+};
+use datafusion::datasource::physical_plan::{FileScanConfig, FileScanConfigBuilder};
+use datafusion::datasource::source::DataSourceExec;
+use datafusion::execution::TaskContext;
+use datafusion::execution::context::SessionContext;
+use datafusion::execution::object_store::ObjectStoreUrl;
+use datafusion::parquet::arrow::ArrowWriter;
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::ExecutionPlan;
+use datafusion::prelude::SessionConfig;
+use datafusion_physical_expr_adapter::{
+ DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
+};
+use datafusion_proto::bytes::{
+ physical_plan_from_bytes_with_extension_codec,
+ physical_plan_to_bytes_with_extension_codec,
+};
+use datafusion_proto::physical_plan::{
+ AsExecutionPlan, PhysicalExtensionCodec, PhysicalExtensionProtoCodec,
+};
+use datafusion_proto::protobuf::{
+ PhysicalExprNode, PhysicalExtensionNode, PhysicalPlanNode,
+ physical_plan_node::PhysicalPlanType,
+};
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, PutPayload};
+use prost::Message;
+use serde::{Deserialize, Serialize};
+
+/// Example showing how to preserve custom adapter information during plan serialization.
+///
+/// This demonstrates:
+/// 1. Creating a custom PhysicalExprAdapter with metadata
+/// 2. Using PhysicalExtensionCodec to intercept serialization
+/// 3. Wrapping adapter info as Extension nodes
+/// 4. Restoring adapters during deserialization
+pub async fn adapter_serialization() -> Result<()> {
+ println!("=== PhysicalExprAdapter Serialization Example ===\n");
+
+ // Step 1: Create sample Parquet data in memory
+ println!("Step 1: Creating sample Parquet data...");
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let batch = record_batch!(("id", Int32, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))?;
+ let path = Path::from("data.parquet");
+ write_parquet(&store, &path, &batch).await?;
+
+ // Step 2: Set up session with custom adapter
+ println!("Step 2: Setting up session with custom adapter...");
+ let logical_schema =
+ Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+
+ let mut cfg = SessionConfig::new();
+ cfg.options_mut().execution.parquet.pushdown_filters = true;
+ let ctx = SessionContext::new_with_config(cfg);
+ ctx.runtime_env().register_object_store(
+ ObjectStoreUrl::parse("memory://")?.as_ref(),
+ Arc::clone(&store),
+ );
+
+ // Create a table with our custom MetadataAdapterFactory
+ let adapter_factory = Arc::new(MetadataAdapterFactory::new("v1"));
+ let listing_config =
+ ListingTableConfig::new(ListingTableUrl::parse("memory:///data.parquet")?)
+ .infer_options(&ctx.state())
+ .await?
+ .with_schema(logical_schema)
+ .with_expr_adapter_factory(
+ Arc::clone(&adapter_factory) as Arc<dyn PhysicalExprAdapterFactory>
+ );
+ let table = ListingTable::try_new(listing_config)?;
+ ctx.register_table("my_table", Arc::new(table))?;
+
+ // Step 3: Create physical plan with filter
+ println!("Step 3: Creating physical plan with filter...");
+ let df = ctx.sql("SELECT * FROM my_table WHERE id > 5").await?;
+ let original_plan = df.create_physical_plan().await?;
+
+ // Verify adapter is present in original plan
+ let has_adapter_before = verify_adapter_in_plan(&original_plan, "original");
+ println!(" Original plan has adapter: {has_adapter_before}");
+
+ // Step 4: Serialize with our custom codec
+ println!("\nStep 4: Serializing plan with AdapterPreservingCodec...");
+ let codec = AdapterPreservingCodec;
+ let bytes = physical_plan_to_bytes_with_extension_codec(
+ Arc::clone(&original_plan),
+ &codec,
+ &codec,
+ )?;
+ println!(" Serialized {} bytes", bytes.len());
+ println!(" (DataSourceExec with adapter was wrapped as PhysicalExtensionNode)");
+
+ // Step 5: Deserialize with our custom codec
+ println!("\nStep 5: Deserializing plan with AdapterPreservingCodec...");
+ let task_ctx = ctx.task_ctx();
+ let restored_plan =
+ physical_plan_from_bytes_with_extension_codec(&bytes, &task_ctx, &codec, &codec)?;
+
+ // Verify adapter is restored
+ let has_adapter_after = verify_adapter_in_plan(&restored_plan, "restored");
+ println!(" Restored plan has adapter: {has_adapter_after}");
+
+ // Step 6: Execute and compare results
+ println!("\nStep 6: Executing plans and comparing results...");
+ let original_results =
+ datafusion::physical_plan::collect(Arc::clone(&original_plan), task_ctx.clone())
+ .await?;
+ let restored_results =
+ datafusion::physical_plan::collect(restored_plan, task_ctx).await?;
+
+ #[rustfmt::skip]
+ let expected = [
+ "+----+",
+ "| id |",
+ "+----+",
+ "| 6 |",
+ "| 7 |",
+ "| 8 |",
+ "| 9 |",
+ "| 10 |",
+ "+----+",
+ ];
+
+ println!("\n Original plan results:");
+ arrow::util::pretty::print_batches(&original_results)?;
+ assert_batches_eq!(expected, &original_results);
+
+ println!("\n Restored plan results:");
+ arrow::util::pretty::print_batches(&restored_results)?;
+ assert_batches_eq!(expected, &restored_results);
+
+ println!("\n=== Example Complete! ===");
+ println!("Key takeaways:");
+ println!(
+ " 1. PhysicalExtensionCodec provides serialize_physical_plan/deserialize_physical_plan hooks"
+ );
+ println!(" 2. Custom metadata can be wrapped as PhysicalExtensionNode");
+ println!(" 3. Nested serialization (protobuf + JSON) works seamlessly");
+ println!(
+ " 4. Both plans produce identical results despite serialization round-trip"
+ );
+ println!(" 5. Adapters are fully preserved through the serialization round-trip");
+
+ Ok(())
+}
+
+// ============================================================================
+// MetadataAdapter - A simple custom adapter with a tag
+// ============================================================================
+
+/// A custom PhysicalExprAdapter that wraps another adapter.
+/// The tag metadata is stored in the factory, not the adapter itself.
+#[derive(Debug)]
+struct MetadataAdapter {
+ inner: Arc<dyn PhysicalExprAdapter>,
+}
+
+impl PhysicalExprAdapter for MetadataAdapter {
+ fn rewrite(&self, expr: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
+ // Simply delegate to inner adapter
+ self.inner.rewrite(expr)
+ }
+}
+
+// ============================================================================
+// MetadataAdapterFactory - Factory for creating MetadataAdapter instances
+// ============================================================================
+
+/// Factory for creating MetadataAdapter instances.
+/// The tag is stored in the factory and extracted via Debug formatting in `extract_adapter_tag`.
+#[derive(Debug)]
+struct MetadataAdapterFactory {
+ // Note: This field is read via Debug formatting in `extract_adapter_tag`.
+ // Rust's dead code analysis doesn't recognize Debug-based field access.
+ // In PR #19234, this field is used by `with_partition_values`, but that method
+ // doesn't exist in upstream DataFusion's PhysicalExprAdapter trait.
+ #[allow(dead_code)]
+ tag: String,
+}
+
+impl MetadataAdapterFactory {
+ fn new(tag: impl Into<String>) -> Self {
+ Self { tag: tag.into() }
+ }
+}
+
+impl PhysicalExprAdapterFactory for MetadataAdapterFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> Arc<dyn PhysicalExprAdapter> {
+ let inner = DefaultPhysicalExprAdapterFactory
+ .create(logical_file_schema, physical_file_schema);
+ Arc::new(MetadataAdapter { inner })
+ }
+}
+
+// ============================================================================
+// AdapterPreservingCodec - Custom codec that preserves adapters
+// ============================================================================
+
+/// Extension payload structure for serializing adapter info
+#[derive(Serialize, Deserialize)]
+struct ExtensionPayload {
+ /// Marker to identify this is our custom extension
+ marker: String,
+ /// JSON-serialized adapter metadata
+ adapter_metadata: AdapterMetadata,
+ /// Protobuf-serialized inner DataSourceExec (without adapter)
+ inner_plan_bytes: Vec<u8>,
+}
+
+/// Metadata about the adapter to recreate it during deserialization
+#[derive(Serialize, Deserialize)]
+struct AdapterMetadata {
+ /// The adapter tag (e.g., "v1")
+ tag: String,
+}
+
+const EXTENSION_MARKER: &str = "adapter_preserving_extension_v1";
+
+/// A codec that intercepts serialization to preserve adapter information.
+#[derive(Debug)]
+struct AdapterPreservingCodec;
+
+impl PhysicalExtensionCodec for AdapterPreservingCodec {
+ // Required method: decode custom extension nodes
+ fn try_decode(
+ &self,
+ buf: &[u8],
+ _inputs: &[Arc<dyn ExecutionPlan>],
Review Comment:
Oh, ok. I hadn't thought about your specific example but rather the interfaces. This specific part uses the existing codec work, I think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "LiaCastaneda (via GitHub)" <gi...@apache.org>.
LiaCastaneda commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2690810007
##########
datafusion-examples/examples/custom_data_source/adapter_serialization.rs:
##########
@@ -0,0 +1,536 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! See `main.rs` for how to run it.
+//!
+//! This example demonstrates how to use the `PhysicalExtensionCodec` trait's
+//! interception methods (`serialize_physical_plan` and `deserialize_physical_plan`)
+//! to implement custom serialization logic.
+//!
+//! The key insight is that `FileScanConfig::expr_adapter_factory` is NOT serialized by
+//! default. This example shows how to:
+//! 1. Detect plans with custom adapters during serialization
+//! 2. Wrap them as Extension nodes with JSON-serialized adapter metadata
+//! 3. Unwrap and restore the adapter during deserialization
+//!
+//! This demonstrates nested serialization (protobuf outer, JSON inner) and the power
+//! of the `PhysicalExtensionCodec` interception pattern. Both plan and expression
+//! serialization route through the codec, enabling interception at every node in the tree.
+
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use arrow::array::record_batch;
+use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::assert_batches_eq;
+use datafusion::common::{Result, not_impl_err};
+use datafusion::datasource::listing::{
+ ListingTable, ListingTableConfig, ListingTableConfigExt, ListingTableUrl,
+};
+use datafusion::datasource::physical_plan::{FileScanConfig, FileScanConfigBuilder};
+use datafusion::datasource::source::DataSourceExec;
+use datafusion::execution::TaskContext;
+use datafusion::execution::context::SessionContext;
+use datafusion::execution::object_store::ObjectStoreUrl;
+use datafusion::parquet::arrow::ArrowWriter;
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::ExecutionPlan;
+use datafusion::prelude::SessionConfig;
+use datafusion_physical_expr_adapter::{
+ DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
+};
+use datafusion_proto::bytes::{
+ physical_plan_from_bytes_with_proto_converter,
+ physical_plan_to_bytes_with_proto_converter,
+};
+use datafusion_proto::physical_plan::from_proto::parse_physical_expr;
+use datafusion_proto::physical_plan::to_proto::serialize_physical_expr;
+use datafusion_proto::physical_plan::{
+ AsExecutionPlan, PhysicalExtensionCodec, PhysicalProtoConverterExtension,
+};
+use datafusion_proto::protobuf::physical_plan_node::PhysicalPlanType;
+use datafusion_proto::protobuf::{
+ PhysicalExprNode, PhysicalExtensionNode, PhysicalPlanNode,
+};
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, PutPayload};
+use prost::Message;
+use serde::{Deserialize, Serialize};
+
+/// Example showing how to preserve custom adapter information during plan serialization.
+///
+/// This demonstrates:
+/// 1. Creating a custom PhysicalExprAdapter with metadata
+/// 2. Using PhysicalExtensionCodec to intercept serialization
+/// 3. Wrapping adapter info as Extension nodes
+/// 4. Restoring adapters during deserialization
+pub async fn adapter_serialization() -> Result<()> {
+ println!("=== PhysicalExprAdapter Serialization Example ===\n");
+
+ // Step 1: Create sample Parquet data in memory
+ println!("Step 1: Creating sample Parquet data...");
+ let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+ let batch = record_batch!(("id", Int32, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))?;
+ let path = Path::from("data.parquet");
+ write_parquet(&store, &path, &batch).await?;
+
+ // Step 2: Set up session with custom adapter
+ println!("Step 2: Setting up session with custom adapter...");
+ let logical_schema =
+ Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+
+ let mut cfg = SessionConfig::new();
+ cfg.options_mut().execution.parquet.pushdown_filters = true;
+ let ctx = SessionContext::new_with_config(cfg);
+ ctx.runtime_env().register_object_store(
+ ObjectStoreUrl::parse("memory://")?.as_ref(),
+ Arc::clone(&store),
+ );
+
+ // Create a table with our custom MetadataAdapterFactory
+ let adapter_factory = Arc::new(MetadataAdapterFactory::new("v1"));
+ let listing_config =
+ ListingTableConfig::new(ListingTableUrl::parse("memory:///data.parquet")?)
+ .infer_options(&ctx.state())
+ .await?
+ .with_schema(logical_schema)
+ .with_expr_adapter_factory(
+ Arc::clone(&adapter_factory) as Arc<dyn PhysicalExprAdapterFactory>
+ );
+ let table = ListingTable::try_new(listing_config)?;
+ ctx.register_table("my_table", Arc::new(table))?;
+
+ // Step 3: Create physical plan with filter
+ println!("Step 3: Creating physical plan with filter...");
+ let df = ctx.sql("SELECT * FROM my_table WHERE id > 5").await?;
+ let original_plan = df.create_physical_plan().await?;
+
+ // Verify adapter is present in original plan
+ let has_adapter_before = verify_adapter_in_plan(&original_plan, "original");
+ println!(" Original plan has adapter: {has_adapter_before}");
+
+ // Step 4: Serialize with our custom codec
+ println!("\nStep 4: Serializing plan with AdapterPreservingCodec...");
+ let codec = AdapterPreservingCodec;
+ let bytes = physical_plan_to_bytes_with_proto_converter(
+ Arc::clone(&original_plan),
+ &codec,
+ &codec,
+ )?;
+ println!(" Serialized {} bytes", bytes.len());
+ println!(" (DataSourceExec with adapter was wrapped as PhysicalExtensionNode)");
+
+ // Step 5: Deserialize with our custom codec
+ println!("\nStep 5: Deserializing plan with AdapterPreservingCodec...");
+ let task_ctx = ctx.task_ctx();
+ let restored_plan =
+ physical_plan_from_bytes_with_proto_converter(&bytes, &task_ctx, &codec, &codec)?;
+
+ // Verify adapter is restored
+ let has_adapter_after = verify_adapter_in_plan(&restored_plan, "restored");
+ println!(" Restored plan has adapter: {has_adapter_after}");
+
+ // Step 6: Execute and compare results
+ println!("\nStep 6: Executing plans and comparing results...");
+ let original_results =
+ datafusion::physical_plan::collect(Arc::clone(&original_plan), task_ctx.clone())
+ .await?;
+ let restored_results =
+ datafusion::physical_plan::collect(restored_plan, task_ctx).await?;
+
+ #[rustfmt::skip]
+ let expected = [
+ "+----+",
+ "| id |",
+ "+----+",
+ "| 6 |",
+ "| 7 |",
+ "| 8 |",
+ "| 9 |",
+ "| 10 |",
+ "+----+",
+ ];
+
+ println!("\n Original plan results:");
+ arrow::util::pretty::print_batches(&original_results)?;
+ assert_batches_eq!(expected, &original_results);
+
+ println!("\n Restored plan results:");
+ arrow::util::pretty::print_batches(&restored_results)?;
+ assert_batches_eq!(expected, &restored_results);
+
+ println!("\n=== Example Complete! ===");
+ println!("Key takeaways:");
+ println!(
+ " 1. PhysicalExtensionCodec provides serialize_physical_plan/deserialize_physical_plan hooks"
+ );
+ println!(" 2. Custom metadata can be wrapped as PhysicalExtensionNode");
+ println!(" 3. Nested serialization (protobuf + JSON) works seamlessly");
+ println!(
+ " 4. Both plans produce identical results despite serialization round-trip"
+ );
+ println!(" 5. Adapters are fully preserved through the serialization round-trip");
+
+ Ok(())
+}
+
+// ============================================================================
+// MetadataAdapter - A simple custom adapter with a tag
+// ============================================================================
+
+/// A custom PhysicalExprAdapter that wraps another adapter.
+/// The tag metadata is stored in the factory, not the adapter itself.
+#[derive(Debug)]
+struct MetadataAdapter {
+ inner: Arc<dyn PhysicalExprAdapter>,
+}
+
+impl PhysicalExprAdapter for MetadataAdapter {
+ fn rewrite(&self, expr: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
+ // Simply delegate to inner adapter
+ self.inner.rewrite(expr)
+ }
+}
+
+// ============================================================================
+// MetadataAdapterFactory - Factory for creating MetadataAdapter instances
+// ============================================================================
+
+/// Factory for creating MetadataAdapter instances.
+/// The tag is stored in the factory and extracted via Debug formatting in `extract_adapter_tag`.
+#[derive(Debug)]
+struct MetadataAdapterFactory {
+ // Note: This field is read via Debug formatting in `extract_adapter_tag`.
+ // Rust's dead code analysis doesn't recognize Debug-based field access.
+ // In PR #19234, this field is used by `with_partition_values`, but that method
+ // doesn't exist in upstream DataFusion's PhysicalExprAdapter trait.
+ #[allow(dead_code)]
+ tag: String,
+}
+
+impl MetadataAdapterFactory {
+ fn new(tag: impl Into<String>) -> Self {
+ Self { tag: tag.into() }
+ }
+}
+
+impl PhysicalExprAdapterFactory for MetadataAdapterFactory {
+ fn create(
+ &self,
+ logical_file_schema: SchemaRef,
+ physical_file_schema: SchemaRef,
+ ) -> Arc<dyn PhysicalExprAdapter> {
+ let inner = DefaultPhysicalExprAdapterFactory
+ .create(logical_file_schema, physical_file_schema);
+ Arc::new(MetadataAdapter { inner })
+ }
+}
+
+// ============================================================================
+// AdapterPreservingCodec - Custom codec that preserves adapters
+// ============================================================================
+
+/// Extension payload structure for serializing adapter info
+#[derive(Serialize, Deserialize)]
+struct ExtensionPayload {
+ /// Marker to identify this is our custom extension
+ marker: String,
+ /// JSON-serialized adapter metadata
+ adapter_metadata: AdapterMetadata,
+ /// Protobuf-serialized inner DataSourceExec (without adapter)
+ inner_plan_bytes: Vec<u8>,
+}
+
+/// Metadata about the adapter to recreate it during deserialization
+#[derive(Serialize, Deserialize)]
+struct AdapterMetadata {
+ /// The adapter tag (e.g., "v1")
+ tag: String,
+}
+
+const EXTENSION_MARKER: &str = "adapter_preserving_extension_v1";
+
+/// A codec that intercepts serialization to preserve adapter information.
+#[derive(Debug)]
+struct AdapterPreservingCodec;
+
+impl PhysicalExtensionCodec for AdapterPreservingCodec {
+ // Required method: decode custom extension nodes
+ fn try_decode(
+ &self,
+ buf: &[u8],
+ _inputs: &[Arc<dyn ExecutionPlan>],
+ ctx: &TaskContext,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ // Try to parse as our extension payload
+ if let Ok(payload) = serde_json::from_slice::<ExtensionPayload>(buf)
+ && payload.marker == EXTENSION_MARKER
+ {
+ // Decode the inner plan
+ let inner_proto = PhysicalPlanNode::decode(&payload.inner_plan_bytes[..])
+ .map_err(|e| {
+ datafusion::error::DataFusionError::Plan(format!(
+ "Failed to decode inner plan: {e}"
+ ))
+ })?;
+
+ // Deserialize the inner plan using default implementation
+ let inner_plan = inner_proto.try_into_physical_plan(ctx, self, self)?;
+
+ // Recreate the adapter factory
+ let adapter_factory = create_adapter_factory(&payload.adapter_metadata.tag);
+
+ // Inject adapter into the plan
+ return inject_adapter_into_plan(inner_plan, adapter_factory);
+ }
+
+ not_impl_err!("Unknown extension type")
+ }
+
+ // Required method: encode custom execution plans
+ fn try_encode(
+ &self,
+ _node: Arc<dyn ExecutionPlan>,
+ _buf: &mut Vec<u8>,
+ ) -> Result<()> {
+ // We don't need this for the example - we use serialize_physical_plan instead
+ not_impl_err!(
+ "try_encode not used - adapter wrapping happens in serialize_physical_plan"
+ )
+ }
+}
+
+impl PhysicalProtoConverterExtension for AdapterPreservingCodec {
+ fn execution_plan_to_proto(
+ &self,
+ plan: &Arc<dyn ExecutionPlan>,
+ extension_codec: &dyn PhysicalExtensionCodec,
+ ) -> Result<PhysicalPlanNode> {
+ // Check if this is a DataSourceExec with adapter
+ if let Some(exec) = plan.as_any().downcast_ref::<DataSourceExec>()
+ && let Some(config) =
+ exec.data_source().as_any().downcast_ref::<FileScanConfig>()
+ && let Some(adapter_factory) = &config.expr_adapter_factory
+ && let Some(tag) = extract_adapter_tag(adapter_factory.as_ref())
+ {
+ // Try to extract our MetadataAdapterFactory's tag
+ println!(" [Serialize] Found DataSourceExec with adapter tag: {tag}");
+
+ // 1. Create adapter metadata
+ let adapter_metadata = AdapterMetadata { tag };
+
+ // 2. Create a copy of the config without the adapter
+ let config_without_adapter = rebuild_config_without_adapter(config);
Review Comment:
If the user skip this step, would it be possible to enter an infitite recursive loop since we are passing the same `AdapterPreservingCodec` when calling `try_from_physical_plan` below?
try_from_physical_plan (wth adapter) -> execution_plan_to_proto (with adapter) -> try_from_physical_plan (with adapter) -> ....
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "LiaCastaneda (via GitHub)" <gi...@apache.org>.
LiaCastaneda commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2691053647
##########
datafusion-examples/examples/proto/expression_deduplication.rs:
##########
@@ -0,0 +1,269 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! See `main.rs` for how to run it.
+//!
+//! This example demonstrates how to use the `PhysicalExtensionCodec` trait's
+//! interception methods to implement expression deduplication during deserialization.
+//!
+//! This pattern is inspired by PR #18192, which introduces expression caching
+//! to reduce memory usage when deserializing plans with duplicate expressions.
+//!
+//! The key insight is that identical expressions serialize to identical protobuf bytes.
+//! By caching deserialized expressions keyed by their protobuf bytes, we can:
+//! 1. Return the same Arc for duplicate expressions
+//! 2. Reduce memory allocation during deserialization
+//! 3. Enable downstream optimizations that rely on Arc pointer equality
+//!
+//! This demonstrates the decorator pattern enabled by the `PhysicalExtensionCodec` trait,
+//! where all expression serialization/deserialization routes through the codec methods.
+
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::sync::{Arc, RwLock};
+
+use arrow::datatypes::{DataType, Field, Schema};
+use datafusion::common::Result;
+use datafusion::execution::TaskContext;
+use datafusion::logical_expr::Operator;
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion::physical_plan::ExecutionPlan;
+use datafusion::physical_plan::expressions::{BinaryExpr, col};
+use datafusion::physical_plan::filter::FilterExec;
+use datafusion::physical_plan::placeholder_row::PlaceholderRowExec;
+use datafusion::prelude::SessionContext;
+use datafusion_proto::physical_plan::from_proto::parse_physical_expr;
+use datafusion_proto::physical_plan::to_proto::serialize_physical_expr;
+use datafusion_proto::physical_plan::{
+ AsExecutionPlan, DefaultPhysicalExtensionCodec, PhysicalExtensionCodec,
+ PhysicalProtoConverterExtension,
+};
+use datafusion_proto::protobuf::{PhysicalExprNode, PhysicalPlanNode};
+use prost::Message;
+
+/// Example showing how to implement expression deduplication using the codec decorator pattern.
+///
+/// This demonstrates:
+/// 1. Creating a CachingCodec that caches expressions by their protobuf bytes
+/// 2. Intercepting deserialization to return cached Arcs for duplicate expressions
+/// 3. Verifying that duplicate expressions share the same Arc after deserialization
+///
+/// Deduplication is keyed by the protobuf bytes representing the expression,
+/// in reality deduplication could be done based on e.g. the pointer address of the
+/// serialized expression in memory, but this is simpler to demonstrate.
+///
+/// In this case our expression is trivial and just for demonstration purposes.
+/// In real scenarios, expressions can be much more complex, e.g. a large InList
+/// expression could be megabytes in size, so deduplication can save significant memory
+/// in addition to more correctly representing the original plan structure.
+pub async fn expression_deduplication() -> Result<()> {
+ println!("=== Expression Deduplication Example ===\n");
+
+ // Create a schema for our test expressions
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Boolean, false)]));
+
+ // Step 1: Create expressions with duplicates
+ println!("Step 1: Creating expressions with duplicates...");
+
+ // Create expression: col("a")
+ let a = col("a", &schema)?;
+
+ // Create a clone to show duplicates
+ let a_clone = Arc::clone(&a);
+
+ // Combine: a OR a_clone
+ let combined_expr =
+ Arc::new(BinaryExpr::new(a, Operator::Or, a_clone)) as Arc<dyn PhysicalExpr>;
+ println!(" Created expression: a OR a with duplicates");
+ println!(" Note: a appears twice in the expression tree\n");
+ // Step 2: Create a filter plan with this expression
+ println!("Step 2: Creating physical plan with the expression...");
+
+ let input = Arc::new(PlaceholderRowExec::new(Arc::clone(&schema)));
+ let filter_plan: Arc<dyn ExecutionPlan> =
+ Arc::new(FilterExec::try_new(combined_expr, input)?);
+
+ println!(" Created FilterExec with duplicate sub-expressions\n");
+
+ // Step 3: Serialize with the caching codec
+ println!("Step 3: Serializing plan...");
+
+ let extension_codec = DefaultPhysicalExtensionCodec {};
+ let caching_codec = CachingCodec::new();
+ let proto = caching_codec.execution_plan_to_proto(&filter_plan, &extension_codec)?;
+
+ // Serialize to bytes
+ let mut bytes = Vec::new();
+ proto.encode(&mut bytes).unwrap();
+ println!(" Serialized plan to {} bytes\n", bytes.len());
+
+ // Step 4: Deserialize with the caching codec
+ println!("Step 4: Deserializing plan with CachingCodec...");
+
+ let ctx = SessionContext::new();
+ let deserialized_plan = proto.try_into_physical_plan(
+ &ctx.task_ctx(),
+ &extension_codec,
+ &caching_codec,
+ )?;
+
+ // Step 5: check that we deduplicated expressions
+ println!("Step 5: Checking for deduplicated expressions...");
+ let Some(filter_exec) = deserialized_plan.as_any().downcast_ref::<FilterExec>()
+ else {
+ panic!("Deserialized plan is not a FilterExec");
+ };
+ let predicate = Arc::clone(filter_exec.predicate());
+ let binary_expr = predicate
+ .as_any()
+ .downcast_ref::<BinaryExpr>()
+ .expect("Predicate is not a BinaryExpr");
+ let left = &binary_expr.left();
+ let right = &binary_expr.right();
+ // Check if left and right point to the same Arc
+ let deduplicated = Arc::ptr_eq(left, right);
+ if deduplicated {
+ println!(" Success: Duplicate expressions were deduplicated!");
+ println!(
+ " Cache Stats: hits={}, misses={}",
+ caching_codec.stats.read().unwrap().cache_hits,
+ caching_codec.stats.read().unwrap().cache_misses,
+ );
+ } else {
+ println!(" Failure: Duplicate expressions were NOT deduplicated.");
+ }
+
+ Ok(())
+}
+
+// ============================================================================
+// CachingCodec - Implements expression deduplication
+// ============================================================================
+
+/// Statistics for cache performance monitoring
+#[derive(Debug, Default)]
+struct CacheStats {
+ cache_hits: usize,
+ cache_misses: usize,
+}
+
+/// A codec that caches deserialized expressions to enable deduplication.
+///
+/// When deserializing, if we've already seen the same protobuf bytes,
+/// we return the cached Arc instead of creating a new allocation.
+#[derive(Debug, Default)]
+struct CachingCodec {
+ /// Cache mapping protobuf bytes -> deserialized expression
+ expr_cache: RwLock<HashMap<Vec<u8>, Arc<dyn PhysicalExpr>>>,
+ /// Statistics for demonstration
+ stats: RwLock<CacheStats>,
+}
+
+impl CachingCodec {
+ fn new() -> Self {
+ Self::default()
+ }
+}
+
+impl PhysicalExtensionCodec for CachingCodec {
+ // Required: decode custom extension nodes
+ fn try_decode(
+ &self,
+ _buf: &[u8],
+ _inputs: &[Arc<dyn ExecutionPlan>],
+ _ctx: &TaskContext,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ datafusion::common::not_impl_err!("No custom extension nodes")
+ }
+
+ // Required: encode custom execution plans
+ fn try_encode(
+ &self,
+ _node: Arc<dyn ExecutionPlan>,
+ _buf: &mut Vec<u8>,
+ ) -> Result<()> {
+ datafusion::common::not_impl_err!("No custom extension nodes")
+ }
+}
+
+impl PhysicalProtoConverterExtension for CachingCodec {
+ fn proto_to_execution_plan(
+ &self,
+ _ctx: &TaskContext,
+ _extension_codec: &dyn PhysicalExtensionCodec,
+ _proto: &PhysicalPlanNode,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ todo!()
+ }
+
+ fn execution_plan_to_proto(
+ &self,
+ plan: &Arc<dyn ExecutionPlan>,
+ extension_codec: &dyn PhysicalExtensionCodec,
+ ) -> Result<PhysicalPlanNode> {
Review Comment:
IIUC (feel free to correct me if I'm understanding this wrongly), unlike `proto_to_physical_expr`, this method will be called only once at the top of the ExecutionPlan. Therefore, if users want to add some kind of interception for a node in the middle of the plan, they would need to downcast to that node inside this method right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: implement protobuf converter trait to allow control over serialization and deserialization processes [datafusion]
Posted by "adriangb (via GitHub)" <gi...@apache.org>.
adriangb commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3807269365
It looks like this conflicted with https://github.com/apache/datafusion/pull/20017 and [broke CI on main](https://github.com/apache/datafusion/actions/runs/21408662031/job/61639470906#step:5:1052). The fix is https://github.com/apache/datafusion/pull/20035/changes/e8cc2719108c5ed57edbd0eed7b62f555943b7a3, I can make a separate PR but maybe we could consider https://github.com/apache/datafusion/pull/20035 since it makes changes to that same example?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3678826952
I have another branch where I followed the original design more closely and the trait had all of the byte serialization/deserialization methods, but in the end I think this cleans it up nicely. We have a default implementation that users can lean on / copy if they need to. There is a question of whether the methods need to be `&mut dyn` or if `&dyn` is sufficient.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#issuecomment-3682043847
> I'm away this week, will have a look ASAP
No rush at all on my side.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Physical Extension Protobuf Codec [datafusion]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #19437:
URL: https://github.com/apache/datafusion/pull/19437#discussion_r2650840361
##########
datafusion-examples/examples/proto/composed_extension_codec.rs:
##########
@@ -62,17 +63,19 @@ pub async fn composed_extension_codec() -> Result<()> {
Arc::new(ParentPhysicalExtensionCodec {}),
Arc::new(ChildPhysicalExtensionCodec {}),
]);
+ let proto_codec = DefaultPhysicalExtensionProtoCodec {};
Review Comment:
could we keep original methods which would create DPEPC{} internally to keep backward compatibility and add new methods which would accept DPEPC as a parameter?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org