This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24169-4b48cb0899ddfdf6d48532e18991df78aadacd62 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit e440932b084e8e68dabc0f5eb1e1a4b8e5054353 Author: Burak Şen <[email protected]> AuthorDate: Sat Aug 8 18:37:01 2026 +0300 refactor(proto): migrate ParquetSource serde (#24169) ## Which issue does this PR close? - Closes #23517. ## Rationale for this change Part of epic #23494. Moves `ParquetSource` protobuf serialization from central dispatch into the source implementation. ## What changes are included in this PR? Add protobuf serialization and deserialization to `ParquetSource`, including the pushdown predicate and `TableParquetOptions`. Rebuild the cached Parquet reader factory from the runtime environment during deserialization. ## Are these changes tested? Yes. Existing Parquet round-trip tests cover this change. ## Are there any user-facing changes? no --- datafusion/datasource-parquet/src/source.rs | 129 +++++++++++++++++++++++ datafusion/proto/src/physical_plan/from_proto.rs | 4 + datafusion/proto/src/physical_plan/mod.rs | 115 ++++---------------- 3 files changed, 153 insertions(+), 95 deletions(-) diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index fb8506f741..b11436812c 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -1057,6 +1057,135 @@ impl FileSource for ParquetSource { inner: Arc::new(new_source) as Arc<dyn FileSource>, }) } + + /// Emit a `ParquetScan` node wrapping the shared base config plus the + /// Parquet-specific pushdown predicate and `TableParquetOptions`. + #[cfg(feature = "proto")] + fn try_to_proto( + &self, + base: &FileScanConfig, + ctx: &datafusion_physical_plan::proto::ExecutionPlanEncodeCtx<'_>, + ) -> datafusion_common::Result< + Option<datafusion_proto_models::protobuf::PhysicalPlanNode>, + > { + use datafusion_proto_models::protobuf; + use protobuf::physical_plan_node::PhysicalPlanType; + + let predicate = self + .filter() + .map(|pred| ctx.encode_expr(&pred)) + .transpose()?; + + let node = protobuf::ParquetScanExecNode { + base_conf: Some(base.try_to_proto(ctx)?), + predicate, + parquet_options: Some(self.table_parquet_options().try_into()?), + }; + Ok(Some(protobuf::PhysicalPlanNode { + physical_plan_type: Some(PhysicalPlanType::ParquetScan(node)), + })) + } +} + +#[cfg(feature = "proto")] +impl ParquetSource { + /// Reconstructs a `DataSourceExec` from a protobuf `ParquetScan`. + /// + /// Rebuilds the reader factory from the decode context because it is not serialized. + pub fn try_from_proto( + node: &datafusion_proto_models::protobuf::PhysicalPlanNode, + ctx: &datafusion_physical_plan::proto::ExecutionPlanDecodeCtx<'_>, + ) -> datafusion_common::Result<Arc<dyn datafusion_physical_plan::ExecutionPlan>> { + use crate::CachedParquetFileReaderFactory; + use arrow::datatypes::Schema; + use datafusion_common::config::TableParquetOptions; + use datafusion_datasource::file_scan_config::FileScanConfig; + use datafusion_datasource::source::DataSourceExec; + use datafusion_execution::object_store::ObjectStoreUrl; + use datafusion_proto_models::protobuf; + + let scan = match &node.physical_plan_type { + Some(protobuf::physical_plan_node::PhysicalPlanType::ParquetScan(scan)) => { + scan + } + _ => { + return datafusion_common::internal_err!( + "PhysicalPlanNode is not a ParquetScan" + ); + } + }; + + let base_conf = scan.base_conf.as_ref().ok_or_else(|| { + datafusion_common::internal_datafusion_err!( + "ParquetScanExecNode is missing required field 'base_conf'" + ) + })?; + + let schema: Arc<Schema> = Arc::new( + base_conf + .schema + .as_ref() + .ok_or_else(|| { + datafusion_common::internal_datafusion_err!( + "FileScanExecConf is missing required field 'schema'" + ) + })? + .try_into()?, + ); + + // The predicate was serialized against the scan's output schema, so it + // must be decoded against the projected schema when a projection is + // present. + let predicate_schema = if !base_conf.projection.is_empty() { + let projected_fields: Vec<_> = base_conf + .projection + .iter() + .map(|&i| schema.field(i as usize).clone()) + .collect(); + Arc::new(Schema::new(projected_fields)) + } else { + schema + }; + + let predicate = scan + .predicate + .as_ref() + .map(|expr| ctx.decode_expr(expr, predicate_schema.as_ref())) + .transpose()?; + + let mut options = TableParquetOptions::default(); + if let Some(table_options) = scan.parquet_options.as_ref() { + options = table_options.try_into()?; + } + + let table_schema = FileScanConfig::parse_table_schema_from_proto(base_conf)?; + let object_store_url = match base_conf.object_store_url.is_empty() { + false => ObjectStoreUrl::parse(&base_conf.object_store_url)?, + true => ObjectStoreUrl::local_filesystem(), + }; + let store = ctx + .task_ctx() + .runtime_env() + .object_store(object_store_url)?; + let metadata_cache = ctx + .task_ctx() + .runtime_env() + .cache_manager + .get_file_metadata_cache(); + let reader_factory = + Arc::new(CachedParquetFileReaderFactory::new(store, metadata_cache)); + + let mut source = ParquetSource::new(table_schema) + .with_parquet_file_reader_factory(reader_factory) + .with_table_parquet_options(options); + + if let Some(predicate) = predicate { + source = source.with_predicate(predicate); + } + let base_config = + FileScanConfig::try_from_proto(base_conf, ctx, Arc::new(source))?; + Ok(DataSourceExec::from_data_source(base_config)) + } } /// Returns the a [`TableSchema`] containing a [`RowNumber`] virtual column and a [`Column`] expression referencing its row index column. diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs index 645854295b..9a3a845d7c 100644 --- a/datafusion/proto/src/physical_plan/from_proto.rs +++ b/datafusion/proto/src/physical_plan/from_proto.rs @@ -425,6 +425,10 @@ pub fn parse_protobuf_partitioning( .transpose() .map(Option::flatten) } +#[deprecated( + since = "55.0.0", + note = "unused by DataFusion; use `FileScanConfig::parse_table_schema_from_proto` to reconstruct the full table schema" +)] pub fn parse_protobuf_file_scan_schema( proto: &protobuf::FileScanExecConf, ) -> Result<Arc<Schema>> { diff --git a/datafusion/proto/src/physical_plan/mod.rs b/datafusion/proto/src/physical_plan/mod.rs index 22f623c8aa..743cbfbf51 100644 --- a/datafusion/proto/src/physical_plan/mod.rs +++ b/datafusion/proto/src/physical_plan/mod.rs @@ -26,8 +26,6 @@ use datafusion_catalog::memory::MemorySourceConfig; use datafusion_common::{ DataFusionError, Result, internal_datafusion_err, internal_err, not_impl_err, }; -#[cfg(feature = "parquet")] -use datafusion_datasource::file::FileSource; use datafusion_datasource::file_scan_config::FileScanConfig; use datafusion_datasource::sink::DataSinkExec; use datafusion_datasource::source::{DataSource, DataSourceExec}; @@ -39,13 +37,9 @@ use datafusion_datasource_csv::source::CsvSource; use datafusion_datasource_json::file_format::JsonSink; use datafusion_datasource_json::source::JsonSource; #[cfg(feature = "parquet")] -use datafusion_datasource_parquet::CachedParquetFileReaderFactory; -#[cfg(feature = "parquet")] use datafusion_datasource_parquet::file_format::ParquetSink; #[cfg(feature = "parquet")] use datafusion_datasource_parquet::source::ParquetSource; -#[cfg(feature = "parquet")] -use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_execution::{FunctionRegistry, TaskContext}; use datafusion_expr::physical_planning_context::ScalarSubqueryResults; use datafusion_expr::{AggregateUDF, HigherOrderUDF, ScalarUDF, WindowUDF}; @@ -1088,8 +1082,15 @@ pub trait PhysicalPlanNodeExt: Sized { PhysicalPlanType::JsonScan(scan) => { self.try_into_json_scan_physical_plan(scan, ctx, proto_converter) } - PhysicalPlanType::ParquetScan(scan) => { - self.try_into_parquet_scan_physical_plan(scan, ctx, proto_converter) + PhysicalPlanType::ParquetScan(_) => { + #[cfg(feature = "parquet")] + { + ParquetSource::try_from_proto(self.node(), &decode_ctx) + } + #[cfg(not(feature = "parquet"))] + panic!( + "Unable to process a Parquet PhysicalPlan when `parquet` feature is not enabled" + ) } PhysicalPlanType::AvroScan(scan) => { self.try_into_avro_scan_physical_plan(scan, ctx, proto_converter) @@ -1396,6 +1397,10 @@ pub trait PhysicalPlanNodeExt: Sized { } #[cfg_attr(not(feature = "parquet"), expect(unused_variables))] + #[deprecated( + since = "55.0.0", + note = "unused by DataFusion; `ParquetSource` deserializes itself via `ParquetSource::try_from_proto`" + )] fn try_into_parquet_scan_physical_plan( &self, scan: &protobuf::ParquetScanExecNode, @@ -1404,74 +1409,17 @@ pub trait PhysicalPlanNodeExt: Sized { ) -> Result<Arc<dyn ExecutionPlan>> { #[cfg(feature = "parquet")] { - let schema = from_proto::parse_protobuf_file_scan_schema( - scan.base_conf.as_ref().unwrap(), - )?; - - // Check if there's a projection and use projected schema for predicate parsing - let base_conf = scan.base_conf.as_ref().unwrap(); - let predicate_schema = if !base_conf.projection.is_empty() { - // Create projected schema for parsing the predicate - let projected_fields: Vec<_> = base_conf - .projection - .iter() - .map(|&i| schema.field(i as usize).clone()) - .collect(); - Arc::new(Schema::new(projected_fields)) - } else { - schema - }; - - let predicate = scan - .predicate - .as_ref() - .map(|expr| { - proto_converter.proto_to_physical_expr( - expr, - predicate_schema.as_ref(), - ctx, - ) - }) - .transpose()?; - let mut options = datafusion_common::config::TableParquetOptions::default(); - - if let Some(table_options) = scan.parquet_options.as_ref() { - options = table_options.try_into()?; - } - - // Parse table schema with partition columns - let table_schema = parse_table_schema_from_proto(base_conf)?; - let object_store_url = match base_conf.object_store_url.is_empty() { - false => ObjectStoreUrl::parse(&base_conf.object_store_url)?, - true => ObjectStoreUrl::local_filesystem(), + let node = protobuf::PhysicalPlanNode { + physical_plan_type: Some(PhysicalPlanType::ParquetScan(scan.clone())), }; - let store = ctx - .task_ctx() - .runtime_env() - .object_store(object_store_url)?; - let metadata_cache = ctx - .task_ctx() - .runtime_env() - .cache_manager - .get_file_metadata_cache(); - let reader_factory = - Arc::new(CachedParquetFileReaderFactory::new(store, metadata_cache)); - - let mut source = ParquetSource::new(table_schema) - .with_parquet_file_reader_factory(reader_factory) - .with_table_parquet_options(options); - - if let Some(predicate) = predicate { - source = source.with_predicate(predicate); - } - let base_config = parse_protobuf_file_scan_config( - base_conf, + let decoder = ConverterPlanDecoder { ctx, proto_converter, - Arc::new(source), - )?; - Ok(DataSourceExec::from_data_source(base_config)) + }; + let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder); + ParquetSource::try_from_proto(&node, &decode_ctx) } + #[cfg(not(feature = "parquet"))] panic!( "Unable to process a Parquet PhysicalPlan when `parquet` feature is not enabled" @@ -2562,29 +2510,6 @@ pub trait PhysicalPlanNodeExt: Sized { } } - #[cfg(feature = "parquet")] - if let Some((maybe_parquet, conf)) = - data_source_exec.downcast_to_file_source::<ParquetSource>() - { - let predicate = conf - .filter() - .map(|pred| proto_converter.physical_expr_to_proto(&pred, codec)) - .transpose()?; - return Ok(Some(protobuf::PhysicalPlanNode { - physical_plan_type: Some(PhysicalPlanType::ParquetScan( - protobuf::ParquetScanExecNode { - base_conf: Some(serialize_file_scan_config( - maybe_parquet, - codec, - proto_converter, - )?), - predicate, - parquet_options: Some(conf.table_parquet_options().try_into()?), - }, - )), - })); - } - #[cfg(feature = "avro")] if let Some(maybe_avro) = data_source.downcast_ref::<FileScanConfig>() { let source = maybe_avro.file_source(); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
