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 eec8b947c7 refactor(proto): migrate JsonSource serde (#24178)
eec8b947c7 is described below
commit eec8b947c71a916385a28a3723753f4ffc7cd8d0
Author: Burak Şen <[email protected]>
AuthorDate: Sat Aug 8 18:37:37 2026 +0300
refactor(proto): migrate JsonSource serde (#24178)
## Which issue does this PR close?
- Part of #23516.
## Rationale for this change
Part of epic #23494. Moves `JsonSource` protobuf serialization from the
central dispatch into the source implementation.
## What changes are included in this PR?
Add protobuf serialization and deserialization to `JsonSource`.
Repoint the live decode arm to `JsonSource::try_from_proto` and remove
the old central encode arm. Keep `try_into_json_scan_physical_plan` as a
deprecated compatibility wrapper that delegates to the new
implementation.
The protobuf wire format remains unchanged.
## Are these changes tested?
Yes. Added `roundtrip_json_scan`.
## Are there any user-facing changes?
The existing `PhysicalPlanNodeExt` method remains available as a
deprecated compatibility wrapper. There is no immediate API removal or
wire-format change.
---
datafusion/datasource-json/Cargo.toml | 1 +
datafusion/datasource-json/src/source.rs | 54 ++++++++++++++++++++++
datafusion/proto/src/physical_plan/mod.rs | 39 ++++++----------
.../proto/tests/cases/roundtrip_physical_plan.rs | 20 ++++++--
4 files changed, 85 insertions(+), 29 deletions(-)
diff --git a/datafusion/datasource-json/Cargo.toml
b/datafusion/datasource-json/Cargo.toml
index 7aefbb42c1..04192083f5 100644
--- a/datafusion/datasource-json/Cargo.toml
+++ b/datafusion/datasource-json/Cargo.toml
@@ -31,6 +31,7 @@ version.workspace = true
all-features = true
[features]
+# Enables protobuf serialization hooks for JSON sources and sinks.
proto = [
"dep:datafusion-proto-models",
"datafusion-datasource/proto",
diff --git a/datafusion/datasource-json/src/source.rs
b/datafusion/datasource-json/src/source.rs
index 8632d6b942..b7c2e5a45c 100644
--- a/datafusion/datasource-json/src/source.rs
+++ b/datafusion/datasource-json/src/source.rs
@@ -231,6 +231,60 @@ impl FileSource for JsonSource {
fn file_type(&self) -> &str {
"json"
}
+
+ /// Emit a `JsonScan` node wrapping the shared base config.
+ #[cfg(feature = "proto")]
+ fn try_to_proto(
+ &self,
+ base: &FileScanConfig,
+ ctx: &datafusion_physical_plan::proto::ExecutionPlanEncodeCtx<'_>,
+ ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalPlanNode>> {
+ use datafusion_proto_models::protobuf;
+ use protobuf::physical_plan_node::PhysicalPlanType;
+
+ let node = protobuf::JsonScanExecNode {
+ base_conf: Some(base.try_to_proto(ctx)?),
+ };
+ Ok(Some(protobuf::PhysicalPlanNode {
+ physical_plan_type: Some(PhysicalPlanType::JsonScan(node)),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+impl JsonSource {
+ /// Reconstructs a `DataSourceExec` from a protobuf `JsonScan`.
+ ///
+ /// Defaults to newline-delimited JSON because protobuf does not encode
the mode.
+ pub fn try_from_proto(
+ node: &datafusion_proto_models::protobuf::PhysicalPlanNode,
+ ctx: &datafusion_physical_plan::proto::ExecutionPlanDecodeCtx<'_>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ use datafusion_datasource::file_scan_config::FileScanConfig;
+ use datafusion_datasource::source::DataSourceExec;
+ use datafusion_proto_models::protobuf;
+
+ let scan = match &node.physical_plan_type {
+
Some(protobuf::physical_plan_node::PhysicalPlanType::JsonScan(scan)) => scan,
+ _ => {
+ return datafusion_common::internal_err!(
+ "PhysicalPlanNode is not a JsonScan"
+ );
+ }
+ };
+
+ let base_conf = scan.base_conf.as_ref().ok_or_else(|| {
+ datafusion_common::internal_datafusion_err!(
+ "JsonScanExecNode is missing required field 'base_conf'"
+ )
+ })?;
+
+ let table_schema =
FileScanConfig::parse_table_schema_from_proto(base_conf)?;
+ let source = Arc::new(JsonSource::new(table_schema));
+
+ let conf = FileScanConfig::try_from_proto(base_conf, ctx, source)?;
+ Ok(DataSourceExec::from_data_source(conf))
+ }
}
impl FileOpener for JsonOpener {
diff --git a/datafusion/proto/src/physical_plan/mod.rs
b/datafusion/proto/src/physical_plan/mod.rs
index 743cbfbf51..ee2d431a41 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -1079,8 +1079,8 @@ pub trait PhysicalPlanNodeExt: Sized {
PhysicalPlanType::CsvScan(_) => {
CsvSource::try_from_proto(self.node(), &decode_ctx)
}
- PhysicalPlanType::JsonScan(scan) => {
- self.try_into_json_scan_physical_plan(scan, ctx,
proto_converter)
+ PhysicalPlanType::JsonScan(_) => {
+ JsonSource::try_from_proto(self.node(), &decode_ctx)
}
PhysicalPlanType::ParquetScan(_) => {
#[cfg(feature = "parquet")]
@@ -1360,21 +1360,25 @@ pub trait PhysicalPlanNodeExt: Sized {
CsvSource::try_from_proto(&node, &decode_ctx)
}
+ #[deprecated(
+ since = "55.0.0",
+ note = "unused by DataFusion; `JsonSource` deserializes itself via
`JsonSource::try_from_proto`"
+ )]
fn try_into_json_scan_physical_plan(
&self,
scan: &protobuf::JsonScanExecNode,
ctx: &PhysicalPlanDecodeContext<'_>,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn ExecutionPlan>> {
- let base_conf = scan.base_conf.as_ref().unwrap();
- let table_schema = parse_table_schema_from_proto(base_conf)?;
- let scan_conf = parse_protobuf_file_scan_config(
- base_conf,
+ let node = protobuf::PhysicalPlanNode {
+ physical_plan_type: Some(PhysicalPlanType::JsonScan(scan.clone())),
+ };
+ let decoder = ConverterPlanDecoder {
ctx,
proto_converter,
- Arc::new(JsonSource::new(table_schema)),
- )?;
- Ok(DataSourceExec::from_data_source(scan_conf))
+ };
+ let decode_ctx = ExecutionPlanDecodeCtx::new(&decoder);
+ JsonSource::try_from_proto(&node, &decode_ctx)
}
fn try_into_arrow_scan_physical_plan(
@@ -2476,23 +2480,6 @@ pub trait PhysicalPlanNodeExt: Sized {
) -> Result<Option<protobuf::PhysicalPlanNode>> {
let data_source = data_source_exec.data_source();
- if let Some(scan_conf) = data_source.downcast_ref::<FileScanConfig>() {
- let source = scan_conf.file_source();
- if let Some(_json_source) = source.downcast_ref::<JsonSource>() {
- return Ok(Some(protobuf::PhysicalPlanNode {
- physical_plan_type: Some(PhysicalPlanType::JsonScan(
- protobuf::JsonScanExecNode {
- base_conf: Some(serialize_file_scan_config(
- scan_conf,
- codec,
- proto_converter,
- )?),
- },
- )),
- }));
- }
- }
-
if let Some(scan_conf) = data_source.downcast_ref::<FileScanConfig>() {
let source = scan_conf.file_source();
if let Some(_arrow_source) = source.downcast_ref::<ArrowSource>() {
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index 65a817b4cb..9efbd90f15 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -37,8 +37,8 @@ use datafusion::datasource::listing::{
use datafusion::datasource::object_store::ObjectStoreUrl;
use datafusion::datasource::physical_plan::{
ArrowSource, CsvSource, FileGroup, FileOutputMode, FileScanConfig,
- FileScanConfigBuilder, FileSinkConfig, ParquetSource,
wrap_partition_type_in_dict,
- wrap_partition_value_in_dict,
+ FileScanConfigBuilder, FileSinkConfig, JsonSource, ParquetSource,
+ wrap_partition_type_in_dict, wrap_partition_value_in_dict,
};
use datafusion::datasource::sink::{DataSink, DataSinkExec};
use datafusion::datasource::source::DataSourceExec;
@@ -1363,6 +1363,21 @@ fn roundtrip_arrow_scan() -> Result<()> {
roundtrip_test(DataSourceExec::from_data_source(scan_config))
}
+#[test]
+fn roundtrip_json_scan() -> Result<()> {
+ let file_schema =
+ Arc::new(Schema::new(vec![Field::new("col", DataType::Utf8, false)]));
+ let file_source =
Arc::new(JsonSource::new(TableSchema::from(&file_schema)));
+ let scan_config =
+ FileScanConfigBuilder::new(ObjectStoreUrl::local_filesystem(),
file_source)
+ .with_file_groups(vec![FileGroup::new(vec![PartitionedFile::new(
+ "/path/to/file.json".to_string(),
+ 1024,
+ )])])
+ .build();
+ roundtrip_test(DataSourceExec::from_data_source(scan_config))
+}
+
#[test]
fn roundtrip_csv_scan_preserves_format_options() -> Result<()> {
use datafusion::common::config::CsvOptions;
@@ -1418,7 +1433,6 @@ fn roundtrip_csv_scan_preserves_format_options() ->
Result<()> {
assert!(csv_source.truncate_rows());
Ok(())
}
-
#[tokio::test]
async fn roundtrip_parquet_exec_with_table_partition_cols() -> Result<()> {
let mut file_group =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
