This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new e440932b08 refactor(proto): migrate ParquetSource serde (#24169)
e440932b08 is described below
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]
