From 2f98a97fc7c08e13dd66d97d69665055388e85f2 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sun, 9 Aug 2026 09:38:03 -0400 Subject: [PATCH 1/7] refactor(proto): drop TryFromProto shims for datasource and sink types The real `TryFrom` impls for `PartitionedFile`, `FileRange`, `FileGroup` (#24006) and for `JsonSink` / `CsvSink` / `ParquetSink` / `FileSinkConfig` (#23781) now live next to the types, so the `TryFromProto` copies in `datafusion-proto` were pure delegation. `TryFrom<&[PartitionedFile]> for protobuf::FileGroup` goes away here and comes back in the `datafusion-proto-models` commit, which is the one crate that can express it. The `PartitionedFile` tests move to `datafusion-datasource` alongside the logic they cover; the two that duplicated existing coverage there are dropped. Part of #24019. Co-Authored-By: Claude Opus 5 --- Cargo.lock | 1 - datafusion/datasource/src/proto.rs | 54 +++++- datafusion/proto/Cargo.toml | 1 - .../proto/src/physical_plan/from_proto.rs | 165 +----------------- .../proto/src/physical_plan/to_proto.rs | 83 +-------- 5 files changed, 54 insertions(+), 250 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3ddb32f60ffd5..a62e3d8c76892 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2520,7 +2520,6 @@ version = "54.1.0" dependencies = [ "arrow", "async-trait", - "chrono", "datafusion", "datafusion-catalog", "datafusion-catalog-listing", diff --git a/datafusion/datasource/src/proto.rs b/datafusion/datasource/src/proto.rs index cf48a461655c7..0f183e23a1ca5 100644 --- a/datafusion/datasource/src/proto.rs +++ b/datafusion/datasource/src/proto.rs @@ -18,10 +18,9 @@ //! Protobuf conversions for the file-scan leaf types owned by this crate: //! [`FileRange`], [`PartitionedFile`] and [`FileGroup`]. //! -//! These are the single copy of that wire logic. `datafusion-proto`'s -//! `TryFromProto` implementations for the same types are thin shims that -//! delegate here, so the format cannot drift between the central serializer and -//! the per-source `try_to_proto` hooks. +//! These are the single copy of that wire logic, used both by the central +//! serializer in `datafusion-proto` and by the per-source `try_to_proto` hooks, +//! so the format cannot drift between them. //! //! None of these conversions need a codec or an encode/decode context: every //! field is plain data or goes through `datafusion-proto-common`. That is why @@ -204,6 +203,53 @@ mod tests { Ok(()) } + #[test] + fn partitioned_file_path_roundtrip_percent_encoded() -> Result<()> { + // The wire format carries the *encoded* path, so a location that already + // contains percent escapes must survive without a second round of + // encoding or decoding. + let path_str = "foo/foo%2Fbar/baz%252Fqux"; + let pf = PartitionedFile::new_from_meta(ObjectMeta { + location: Path::parse(path_str)?, + last_modified: Utc.timestamp_nanos(1_000), + size: 42, + e_tag: None, + version: None, + }); + + let encoded = protobuf::PartitionedFile::try_from(&pf)?; + assert_eq!(encoded.path, path_str); + + let decoded = PartitionedFile::try_from(&encoded)?; + assert_eq!(decoded.object_meta.location.as_ref(), path_str); + assert_eq!(decoded.object_meta.location, pf.object_meta.location); + Ok(()) + } + + #[test] + fn partitioned_file_arrow_schema_roundtrip_preserves_metadata() -> Result<()> { + use std::collections::HashMap; + + let arrow_schema = Arc::new(Schema::new_with_metadata( + vec![ + Field::new("id", DataType::Int64, false), + Field::new("value", DataType::Utf8, true).with_metadata(HashMap::from([ + ("field_meta".to_string(), "field_value".to_string()), + ])), + ], + HashMap::from([("schema_meta".to_string(), "schema_value".to_string())]), + )); + let pf = PartitionedFile::new("foo/bar.parquet", 10) + .with_arrow_schema(Arc::clone(&arrow_schema)); + + let encoded = protobuf::PartitionedFile::try_from(&pf)?; + assert!(encoded.arrow_schema.is_some()); + + let decoded = PartitionedFile::try_from(&encoded)?; + assert_eq!(decoded.arrow_schema.as_deref(), Some(arrow_schema.as_ref())); + Ok(()) + } + #[test] fn partitioned_file_from_proto_rejects_invalid_path() { let proto = protobuf::PartitionedFile { diff --git a/datafusion/proto/Cargo.toml b/datafusion/proto/Cargo.toml index 314480937940f..6a78e795933c1 100644 --- a/datafusion/proto/Cargo.toml +++ b/datafusion/proto/Cargo.toml @@ -56,7 +56,6 @@ avro = ["datafusion-datasource-avro"] [dependencies] arrow = { workspace = true } -chrono = { workspace = true } datafusion-catalog = { workspace = true } datafusion-catalog-listing = { workspace = true } datafusion-common = { workspace = true } diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs index 645854295bc00..476cd8aba4bbd 100644 --- a/datafusion/proto/src/physical_plan/from_proto.rs +++ b/datafusion/proto/src/physical_plan/from_proto.rs @@ -23,16 +23,10 @@ use arrow::array::RecordBatch; use arrow::compute::SortOptions; use arrow::datatypes::{Field, Schema}; use arrow::ipc::reader::StreamReader; -use datafusion_common::{DataFusionError, Result, internal_datafusion_err, not_impl_err}; +use datafusion_common::{Result, internal_datafusion_err, not_impl_err}; +use datafusion_datasource::TableSchema; use datafusion_datasource::file::FileSource; -use datafusion_datasource::file_groups::FileGroup; use datafusion_datasource::file_scan_config::FileScanConfig; -use datafusion_datasource::file_sink_config::FileSinkConfig; -use datafusion_datasource::{FileRange, PartitionedFile, TableSchema}; -use datafusion_datasource_csv::file_format::CsvSink; -use datafusion_datasource_json::file_format::JsonSink; -#[cfg(feature = "parquet")] -use datafusion_datasource_parquet::file_format::ParquetSink; use datafusion_execution::{FunctionRegistry, TaskContext}; use datafusion_expr::WindowFunctionDefinition; use datafusion_physical_expr::expressions::{LambdaExpr, LambdaVariable}; @@ -467,66 +461,6 @@ pub fn parse_record_batches(buf: &[u8]) -> Result> { Ok(batches) } -/// Thin shim over `TryFrom<&protobuf::PartitionedFile>`, which owns the wire logic. -impl TryFromProto<&protobuf::PartitionedFile> for PartitionedFile { - type Error = DataFusionError; - - fn try_from_proto(val: &protobuf::PartitionedFile) -> Result { - PartitionedFile::try_from(val) - } -} - -/// Thin shim over `TryFrom<&protobuf::FileRange>`, which owns the wire logic. -impl TryFromProto<&protobuf::FileRange> for FileRange { - type Error = DataFusionError; - - fn try_from_proto(value: &protobuf::FileRange) -> Result { - FileRange::try_from(value) - } -} - -/// Thin shim over `TryFrom<&protobuf::FileGroup>`, which owns the wire logic. -impl TryFromProto<&protobuf::FileGroup> for FileGroup { - type Error = DataFusionError; - - fn try_from_proto(val: &protobuf::FileGroup) -> Result { - FileGroup::try_from(val) - } -} - -impl TryFromProto<&protobuf::JsonSink> for JsonSink { - type Error = DataFusionError; - - fn try_from_proto(value: &protobuf::JsonSink) -> Result { - Self::try_from(value) - } -} - -#[cfg(feature = "parquet")] -impl TryFromProto<&protobuf::ParquetSink> for ParquetSink { - type Error = DataFusionError; - - fn try_from_proto(value: &protobuf::ParquetSink) -> Result { - Self::try_from(value) - } -} - -impl TryFromProto<&protobuf::CsvSink> for CsvSink { - type Error = DataFusionError; - - fn try_from_proto(value: &protobuf::CsvSink) -> Result { - Self::try_from(value) - } -} - -impl TryFromProto<&protobuf::FileSinkConfig> for FileSinkConfig { - type Error = DataFusionError; - - fn try_from_proto(conf: &protobuf::FileSinkConfig) -> Result { - conf.try_into() - } -} - /// Concrete [`PhysicalExprDecode`] driver that backs /// [`PhysicalExprDecodeCtx`] inside `parse_physical_expr_with_converter`. /// @@ -553,98 +487,3 @@ impl datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprD .proto_to_physical_expr(node, schema, self.ctx) } } - -#[cfg(test)] -mod tests { - use super::*; - use arrow::datatypes::{DataType, Field, Schema}; - use chrono::{TimeZone, Utc}; - use datafusion_common::ScalarValue; - use object_store::ObjectMeta; - use object_store::path::Path; - - #[test] - fn partitioned_file_path_roundtrip_percent_encoded() { - let path_str = "foo/foo%2Fbar/baz%252Fqux"; - let pf = PartitionedFile::new_from_meta(ObjectMeta { - location: Path::parse(path_str).unwrap(), - last_modified: Utc.timestamp_nanos(1_000), - size: 42, - e_tag: None, - version: None, - }); - - let proto = protobuf::PartitionedFile::try_from_proto(&pf).unwrap(); - assert_eq!(proto.path, path_str); - - let pf2 = PartitionedFile::try_from_proto(&proto).unwrap(); - assert_eq!(pf2.object_meta.location.as_ref(), path_str); - assert_eq!(pf2.object_meta.location, pf.object_meta.location); - assert_eq!(pf2.object_meta.size, pf.object_meta.size); - assert_eq!(pf2.object_meta.last_modified, pf.object_meta.last_modified); - } - - #[test] - fn partitioned_file_arrow_schema_roundtrip() { - use std::collections::HashMap; - - let arrow_schema = Arc::new(Schema::new_with_metadata( - vec![ - Field::new("id", DataType::Int64, false), - Field::new("value", DataType::Utf8, true).with_metadata(HashMap::from([ - ("field_meta".to_string(), "field_value".to_string()), - ])), - ], - HashMap::from([("schema_meta".to_string(), "schema_value".to_string())]), - )); - let pf = PartitionedFile::new("foo/bar.parquet", 10) - .with_arrow_schema(Arc::clone(&arrow_schema)); - - let proto = protobuf::PartitionedFile::try_from_proto(&pf).unwrap(); - assert!(proto.arrow_schema.is_some()); - - let decoded = PartitionedFile::try_from_proto(&proto).unwrap(); - assert_eq!( - decoded.arrow_schema.as_ref().map(|s| s.as_ref()), - Some(arrow_schema.as_ref()) - ); - } - - #[test] - fn partitioned_file_statistics_roundtrip_with_partition_values() { - use datafusion_common::Statistics; - let file_schema = Schema::new(vec![Field::new("a", DataType::Int32, true)]); - let pf = PartitionedFile::new("foo/bar.parquet", 1234) - .with_partition_values(vec![ScalarValue::from("2024-01-01")]) - .with_statistics(Arc::new(Statistics::new_unknown(&file_schema))); - - // `statistics` covers the full table schema: file columns followed by one - // entry per partition column. - let expected_len = file_schema.fields().len() + pf.partition_values.len(); - assert_eq!( - pf.statistics.as_ref().unwrap().column_statistics.len(), - expected_len - ); - - let proto = protobuf::PartitionedFile::try_from_proto(&pf).unwrap(); - let decoded = PartitionedFile::try_from_proto(&proto).unwrap(); - - assert_eq!(decoded.statistics, pf.statistics); - } - - #[test] - fn partitioned_file_from_proto_invalid_path() { - let proto = protobuf::PartitionedFile { - arrow_schema: None, - path: "foo//bar".to_string(), - size: 1, - last_modified_ns: 0, - partition_values: vec![], - range: None, - statistics: None, - }; - - let err = PartitionedFile::try_from_proto(&proto).unwrap_err(); - assert!(err.to_string().contains("Invalid object_store path")); - } -} diff --git a/datafusion/proto/src/physical_plan/to_proto.rs b/datafusion/proto/src/physical_plan/to_proto.rs index c8a7ea383a69f..8d8224b491e96 100644 --- a/datafusion/proto/src/physical_plan/to_proto.rs +++ b/datafusion/proto/src/physical_plan/to_proto.rs @@ -20,16 +20,8 @@ use std::sync::Arc; use arrow::array::RecordBatch; use arrow::datatypes::Schema; use arrow::ipc::writer::StreamWriter; -use datafusion_common::{ - DataFusionError, Result, internal_datafusion_err, internal_err, not_impl_err, -}; +use datafusion_common::{Result, internal_datafusion_err, internal_err, not_impl_err}; use datafusion_datasource::file_scan_config::FileScanConfig; -use datafusion_datasource::file_sink_config::FileSinkConfig; -use datafusion_datasource::{FileRange, PartitionedFile}; -use datafusion_datasource_csv::file_format::CsvSink; -use datafusion_datasource_json::file_format::JsonSink; -#[cfg(feature = "parquet")] -use datafusion_datasource_parquet::file_format::ParquetSink; use datafusion_expr::WindowFrame; use datafusion_physical_expr::window::{SlidingAggregateWindowExpr, StandardWindowExpr}; use datafusion_physical_expr::{HigherOrderFunctionExpr, ScalarFunctionExpr}; @@ -43,7 +35,6 @@ use super::{ ConverterPlanEncoder, DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalProtoConverterExtension, encode_human_display_alias, }; -use crate::convert::TryFromProto; use crate::protobuf::{ self, PhysicalSortExprNode, physical_aggregate_expr_node, physical_window_expr_node, }; @@ -172,7 +163,7 @@ pub fn serialize_physical_window_expr( codec, proto_converter, )?; - let window_frame = protobuf::WindowFrame::try_from_proto(window_frame.as_ref()) + let window_frame = protobuf::WindowFrame::try_from(window_frame.as_ref()) .map_err(|e| internal_datafusion_err!("{e}"))?; Ok(protobuf::PhysicalWindowExprNode { @@ -365,43 +356,6 @@ pub fn serialize_partitioning( ) } -/// Thin shim over `TryFrom<&PartitionedFile>`, which owns the wire logic. -impl TryFromProto<&PartitionedFile> for protobuf::PartitionedFile { - type Error = DataFusionError; - - fn try_from_proto(pf: &PartitionedFile) -> Result { - pf.try_into() - } -} - -/// Thin shim over `TryFrom<&FileRange>`, which owns the wire logic. -impl TryFromProto<&FileRange> for protobuf::FileRange { - type Error = DataFusionError; - - fn try_from_proto(value: &FileRange) -> Result { - value.try_into() - } -} - -/// Thin shim over `TryFrom<&PartitionedFile>`, which owns the wire logic. -/// -/// The slice form cannot be a `TryFrom` impl: the orphan rule only accepts a -/// type this crate owns, and `&[PartitionedFile]` is not one (`&FileGroup` is, -/// hence the impl next to the type). Callers inside DataFusion go through -/// `FileGroup`; this stays for downstream users of the published signature. -impl TryFromProto<&[PartitionedFile]> for protobuf::FileGroup { - type Error = DataFusionError; - - fn try_from_proto(gr: &[PartitionedFile]) -> Result { - Ok(protobuf::FileGroup { - files: gr - .iter() - .map(TryInto::try_into) - .collect::>>()?, - }) - } -} - pub fn serialize_file_scan_config( conf: &FileScanConfig, codec: &dyn PhysicalExtensionCodec, @@ -440,36 +394,3 @@ pub fn serialize_record_batches(batches: &[RecordBatch]) -> Result> { writer.finish()?; Ok(buf) } - -impl TryFromProto<&JsonSink> for protobuf::JsonSink { - type Error = DataFusionError; - - fn try_from_proto(value: &JsonSink) -> Result { - Self::try_from(value) - } -} - -impl TryFromProto<&CsvSink> for protobuf::CsvSink { - type Error = DataFusionError; - - fn try_from_proto(value: &CsvSink) -> Result { - Self::try_from(value) - } -} - -#[cfg(feature = "parquet")] -impl TryFromProto<&ParquetSink> for protobuf::ParquetSink { - type Error = DataFusionError; - - fn try_from_proto(value: &ParquetSink) -> Result { - Self::try_from(value) - } -} - -impl TryFromProto<&FileSinkConfig> for protobuf::FileSinkConfig { - type Error = DataFusionError; - - fn try_from_proto(conf: &FileSinkConfig) -> Result { - conf.try_into() - } -} From add634f10fdce817ad8a9ee5cf26b148d783d597 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sun, 9 Aug 2026 09:46:42 -0400 Subject: [PATCH 2/7] refactor(proto): restore window-frame conversions as From/TryFrom in datafusion-expr MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `WindowFrame`, `WindowFrameBound`, `WindowFrameUnits`, `MergeIntoClauseKind` and `NullTreatment` are owned by `datafusion-expr`, so the orphan rule lets their proto conversions live next to the types as standard `From` / `TryFrom` impls — the shape they had in 54.1.0 — instead of the `FromProto` / `TryFromProto` workaround. `datafusion-expr` gains a `proto` feature (optional `datafusion-proto-common` and `datafusion-proto-models` deps), matching `datafusion-datasource`. The error types are unchanged: `FromProtoError` decoding, `ToProtoError` encoding. Part of #24019. Co-Authored-By: Claude Opus 5 --- Cargo.lock | 2 + datafusion/expr/Cargo.toml | 6 + datafusion/expr/src/lib.rs | 5 + datafusion/expr/src/proto.rs | 246 ++++++++++++++++++ datafusion/proto/Cargo.toml | 2 +- .../proto/src/logical_plan/from_proto.rs | 99 +------ datafusion/proto/src/logical_plan/to_proto.rs | 100 +------ .../proto/src/physical_plan/from_proto.rs | 3 +- 8 files changed, 276 insertions(+), 187 deletions(-) create mode 100644 datafusion/expr/src/proto.rs diff --git a/Cargo.lock b/Cargo.lock index a62e3d8c76892..66d67fb6417ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2169,6 +2169,8 @@ dependencies = [ "datafusion-functions-aggregate-common", "datafusion-functions-window-common", "datafusion-physical-expr-common", + "datafusion-proto-common", + "datafusion-proto-models", "env_logger", "indexmap 2.14.0", "insta", diff --git a/datafusion/expr/Cargo.toml b/datafusion/expr/Cargo.toml index 8cec01feb30b5..4fe7b65f6d05f 100644 --- a/datafusion/expr/Cargo.toml +++ b/datafusion/expr/Cargo.toml @@ -42,6 +42,10 @@ name = "datafusion_expr" [features] default = ["sql"] +# Enables protobuf conversions for the expression types owned by this crate. +# Off by default so consumers that never serialize plans pay nothing. Mirrors +# the `proto` feature on `datafusion-datasource` and friends. +proto = ["dep:datafusion-proto-common", "dep:datafusion-proto-models"] recursive_protection = ["dep:recursive"] sql = ["sqlparser"] @@ -56,6 +60,8 @@ datafusion-expr-common = { workspace = true } datafusion-functions-aggregate-common = { workspace = true } datafusion-functions-window-common = { workspace = true } datafusion-physical-expr-common = { workspace = true } +datafusion-proto-common = { workspace = true, optional = true } +datafusion-proto-models = { workspace = true, optional = true } indexmap = { workspace = true } itertools = { workspace = true } recursive = { workspace = true, optional = true } diff --git a/datafusion/expr/src/lib.rs b/datafusion/expr/src/lib.rs index 1033952642a2b..75041c701454a 100644 --- a/datafusion/expr/src/lib.rs +++ b/datafusion/expr/src/lib.rs @@ -69,6 +69,11 @@ pub mod dml { pub use crate::logical_plan::dml::*; } pub mod planner; +/// Protobuf conversions for [`WindowFrame`], [`WindowFrameBound`], +/// [`WindowFrameUnits`], [`MergeIntoClauseKind`](dml::MergeIntoClauseKind) and +/// [`NullTreatment`](expr::NullTreatment), gated on the `proto` feature. +#[cfg(feature = "proto")] +mod proto; pub mod registry; pub mod simplify; pub mod sort_properties { diff --git a/datafusion/expr/src/proto.rs b/datafusion/expr/src/proto.rs new file mode 100644 index 0000000000000..00b340210807d --- /dev/null +++ b/datafusion/expr/src/proto.rs @@ -0,0 +1,246 @@ +// 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. + +//! Protobuf conversions for the expression types owned by this crate: +//! [`WindowFrame`], [`WindowFrameBound`], [`WindowFrameUnits`], +//! [`MergeIntoClauseKind`](crate::dml::MergeIntoClauseKind) and +//! [`NullTreatment`](crate::expr::NullTreatment). +//! +//! These are plain [`From`] / [`TryFrom`] impls rather than something taking a +//! codec: every field is either an enum tag or a [`ScalarValue`], so the +//! conversion needs nothing but the value itself. The orphan rule allows them +//! here because one side of each conversion is a type this crate owns. +//! +//! [`ScalarValue`]: datafusion_common::ScalarValue + +use datafusion_common::ScalarValue; +use datafusion_proto_common::{FromProtoError, ToProtoError}; +use datafusion_proto_models::protobuf; + +use crate::dml::MergeIntoClauseKind; +use crate::expr::NullTreatment; +use crate::{WindowFrame, WindowFrameBound, WindowFrameUnits}; + +impl From for WindowFrameUnits { + fn from(units: protobuf::WindowFrameUnits) -> Self { + match units { + protobuf::WindowFrameUnits::Rows => Self::Rows, + protobuf::WindowFrameUnits::Range => Self::Range, + protobuf::WindowFrameUnits::Groups => Self::Groups, + } + } +} + +impl From for protobuf::WindowFrameUnits { + fn from(units: WindowFrameUnits) -> Self { + match units { + WindowFrameUnits::Rows => Self::Rows, + WindowFrameUnits::Range => Self::Range, + WindowFrameUnits::Groups => Self::Groups, + } + } +} + +impl TryFrom for WindowFrameBound { + type Error = FromProtoError; + + fn try_from(bound: protobuf::WindowFrameBound) -> Result { + let bound_type = + protobuf::WindowFrameBoundType::try_from(bound.window_frame_bound_type) + .map_err(|_| { + FromProtoError::unknown( + "WindowFrameBoundType", + bound.window_frame_bound_type, + ) + })?; + match bound_type { + protobuf::WindowFrameBoundType::CurrentRow => Ok(Self::CurrentRow), + protobuf::WindowFrameBoundType::Preceding => match bound.bound_value { + Some(x) => Ok(Self::Preceding(ScalarValue::try_from(&x)?)), + None => Ok(Self::Preceding(ScalarValue::UInt64(None))), + }, + protobuf::WindowFrameBoundType::Following => match bound.bound_value { + Some(x) => Ok(Self::Following(ScalarValue::try_from(&x)?)), + None => Ok(Self::Following(ScalarValue::UInt64(None))), + }, + } + } +} + +impl TryFrom<&WindowFrameBound> for protobuf::WindowFrameBound { + type Error = ToProtoError; + + fn try_from(bound: &WindowFrameBound) -> Result { + Ok(match bound { + WindowFrameBound::CurrentRow => Self { + window_frame_bound_type: protobuf::WindowFrameBoundType::CurrentRow + .into(), + bound_value: None, + }, + WindowFrameBound::Preceding(v) => Self { + window_frame_bound_type: protobuf::WindowFrameBoundType::Preceding.into(), + bound_value: Some(v.try_into()?), + }, + WindowFrameBound::Following(v) => Self { + window_frame_bound_type: protobuf::WindowFrameBoundType::Following.into(), + bound_value: Some(v.try_into()?), + }, + }) + } +} + +impl TryFrom for WindowFrame { + type Error = FromProtoError; + + fn try_from(window: protobuf::WindowFrame) -> Result { + let units = WindowFrameUnits::from( + protobuf::WindowFrameUnits::try_from(window.window_frame_units).map_err( + |_| { + FromProtoError::unknown("WindowFrameUnits", window.window_frame_units) + }, + )?, + ); + let start_bound = WindowFrameBound::try_from( + window + .start_bound + .ok_or_else(|| FromProtoError::required("start_bound"))?, + )?; + let end_bound = window + .end_bound + .map(|end_bound| match end_bound { + protobuf::window_frame::EndBound::Bound(end_bound) => { + WindowFrameBound::try_from(end_bound) + } + }) + .transpose()? + .unwrap_or(WindowFrameBound::CurrentRow); + Ok(WindowFrame::new_bounds(units, start_bound, end_bound)) + } +} + +impl TryFrom<&WindowFrame> for protobuf::WindowFrame { + type Error = ToProtoError; + + fn try_from(window: &WindowFrame) -> Result { + Ok(Self { + window_frame_units: protobuf::WindowFrameUnits::from(window.units).into(), + start_bound: Some((&window.start_bound).try_into()?), + end_bound: Some(protobuf::window_frame::EndBound::Bound( + (&window.end_bound).try_into()?, + )), + }) + } +} + +impl From for MergeIntoClauseKind { + fn from(kind: protobuf::merge_into_clause_node::Kind) -> Self { + match kind { + protobuf::merge_into_clause_node::Kind::Matched => Self::Matched, + protobuf::merge_into_clause_node::Kind::NotMatched => Self::NotMatched, + protobuf::merge_into_clause_node::Kind::NotMatchedByTarget => { + Self::NotMatchedByTarget + } + protobuf::merge_into_clause_node::Kind::NotMatchedBySource => { + Self::NotMatchedBySource + } + } + } +} + +impl From for protobuf::merge_into_clause_node::Kind { + fn from(kind: MergeIntoClauseKind) -> Self { + match kind { + MergeIntoClauseKind::Matched => Self::Matched, + MergeIntoClauseKind::NotMatched => Self::NotMatched, + MergeIntoClauseKind::NotMatchedByTarget => Self::NotMatchedByTarget, + MergeIntoClauseKind::NotMatchedBySource => Self::NotMatchedBySource, + } + } +} + +impl From for NullTreatment { + fn from(t: protobuf::NullTreatment) -> Self { + match t { + protobuf::NullTreatment::RespectNulls => Self::RespectNulls, + protobuf::NullTreatment::IgnoreNulls => Self::IgnoreNulls, + } + } +} + +impl From for protobuf::NullTreatment { + fn from(t: NullTreatment) -> Self { + match t { + NullTreatment::RespectNulls => Self::RespectNulls, + NullTreatment::IgnoreNulls => Self::IgnoreNulls, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn window_frame_roundtrip() -> Result<(), Box> { + let frame = WindowFrame::new_bounds( + WindowFrameUnits::Range, + WindowFrameBound::Preceding(ScalarValue::UInt64(Some(2))), + WindowFrameBound::Following(ScalarValue::UInt64(Some(3))), + ); + + let encoded = protobuf::WindowFrame::try_from(&frame)?; + let decoded = WindowFrame::try_from(encoded)?; + + assert_eq!(decoded.units, frame.units); + assert_eq!(decoded.start_bound, frame.start_bound); + assert_eq!(decoded.end_bound, frame.end_bound); + Ok(()) + } + + #[test] + fn window_frame_from_proto_rejects_missing_start_bound() { + let proto = protobuf::WindowFrame { + window_frame_units: protobuf::WindowFrameUnits::Rows.into(), + start_bound: None, + end_bound: None, + }; + + let err = WindowFrame::try_from(proto).unwrap_err(); + assert!( + err.to_string().contains("start_bound"), + "unexpected error: {err}" + ); + } + + #[test] + fn missing_end_bound_decodes_as_current_row() -> Result<(), Box> + { + let proto = protobuf::WindowFrame { + window_frame_units: protobuf::WindowFrameUnits::Rows.into(), + start_bound: Some(protobuf::WindowFrameBound { + window_frame_bound_type: protobuf::WindowFrameBoundType::CurrentRow + .into(), + bound_value: None, + }), + end_bound: None, + }; + + let decoded = WindowFrame::try_from(proto)?; + assert_eq!(decoded.end_bound, WindowFrameBound::CurrentRow); + Ok(()) + } +} diff --git a/datafusion/proto/Cargo.toml b/datafusion/proto/Cargo.toml index 6a78e795933c1..b58fac53c5ea4 100644 --- a/datafusion/proto/Cargo.toml +++ b/datafusion/proto/Cargo.toml @@ -66,7 +66,7 @@ datafusion-datasource-csv = { workspace = true, features = ["proto"] } datafusion-datasource-json = { workspace = true, features = ["proto"] } datafusion-datasource-parquet = { workspace = true, optional = true, features = ["proto"] } datafusion-execution = { workspace = true } -datafusion-expr = { workspace = true } +datafusion-expr = { workspace = true, features = ["proto"] } datafusion-functions-table = { workspace = true } datafusion-physical-expr = { workspace = true, features = ["proto"] } datafusion-physical-expr-common = { workspace = true, features = ["proto"] } diff --git a/datafusion/proto/src/logical_plan/from_proto.rs b/datafusion/proto/src/logical_plan/from_proto.rs index 00cc7f6a9d835..9c91799ae9675 100644 --- a/datafusion/proto/src/logical_plan/from_proto.rs +++ b/datafusion/proto/src/logical_plan/from_proto.rs @@ -36,8 +36,7 @@ use datafusion_expr::logical_plan::Subquery; use datafusion_expr::{ Between, BinaryExpr, Case, Cast, Expr, GroupingSet, GroupingSet::GroupingSets, - JoinConstraint, JoinType, Like, Operator, TryCast, WindowFrame, WindowFrameBound, - WindowFrameUnits, + JoinConstraint, JoinType, Like, Operator, TryCast, WindowFrame, expr::{self, InList, WindowFunction}, logical_plan::{PlanType, StringifiedPlan}, }; @@ -91,16 +90,6 @@ impl FromProto<&protobuf::UnnestOptions> for UnnestOptions { } } -impl FromProto for WindowFrameUnits { - fn from_proto(units: protobuf::WindowFrameUnits) -> Self { - match units { - protobuf::WindowFrameUnits::Rows => Self::Rows, - protobuf::WindowFrameUnits::Range => Self::Range, - protobuf::WindowFrameUnits::Groups => Self::Groups, - } - } -} - impl TryFromProto for TableReference { type Error = Error; @@ -170,56 +159,6 @@ impl FromProto<&protobuf::StringifiedPlan> for StringifiedPlan { } } -impl TryFromProto for WindowFrame { - type Error = Error; - - fn try_from_proto(window: protobuf::WindowFrame) -> Result { - let units = WindowFrameUnits::from_proto( - protobuf::WindowFrameUnits::try_from(window.window_frame_units).map_err( - |_| Error::unknown("WindowFrameUnits", window.window_frame_units), - )?, - ); - let start_bound = WindowFrameBound::try_from_proto( - window - .start_bound - .ok_or_else(|| Error::required("start_bound"))?, - )?; - let end_bound = window - .end_bound - .map(|end_bound| match end_bound { - protobuf::window_frame::EndBound::Bound(end_bound) => { - WindowFrameBound::try_from_proto(end_bound) - } - }) - .transpose()? - .unwrap_or(WindowFrameBound::CurrentRow); - Ok(WindowFrame::new_bounds(units, start_bound, end_bound)) - } -} - -impl TryFromProto for WindowFrameBound { - type Error = Error; - - fn try_from_proto(bound: protobuf::WindowFrameBound) -> Result { - let bound_type = - protobuf::WindowFrameBoundType::try_from(bound.window_frame_bound_type) - .map_err(|_| { - Error::unknown("WindowFrameBoundType", bound.window_frame_bound_type) - })?; - match bound_type { - protobuf::WindowFrameBoundType::CurrentRow => Ok(Self::CurrentRow), - protobuf::WindowFrameBoundType::Preceding => match bound.bound_value { - Some(x) => Ok(Self::Preceding(ScalarValue::try_from(&x)?)), - None => Ok(Self::Preceding(ScalarValue::UInt64(None))), - }, - protobuf::WindowFrameBoundType::Following => match bound.bound_value { - Some(x) => Ok(Self::Following(ScalarValue::try_from(&x)?)), - None => Ok(Self::Following(ScalarValue::UInt64(None))), - }, - } - } -} - impl FromProto for JoinType { fn from_proto(t: protobuf::JoinType) -> Self { match t { @@ -255,25 +194,6 @@ impl FromProto for NullEquality { } } -impl FromProto for MergeIntoClauseKind { - fn from_proto(k: protobuf::merge_into_clause_node::Kind) -> Self { - match k { - protobuf::merge_into_clause_node::Kind::Matched => { - MergeIntoClauseKind::Matched - } - protobuf::merge_into_clause_node::Kind::NotMatched => { - MergeIntoClauseKind::NotMatched - } - protobuf::merge_into_clause_node::Kind::NotMatchedByTarget => { - MergeIntoClauseKind::NotMatchedByTarget - } - protobuf::merge_into_clause_node::Kind::NotMatchedBySource => { - MergeIntoClauseKind::NotMatchedBySource - } - } - } -} - /// Reconstruct a [`WriteOp`] from a [`protobuf::DmlNode`], reading the /// `merge_into` payload when the type tag is `MergeInto`. pub fn parse_write_op( @@ -331,7 +251,7 @@ fn parse_merge_into_clause( clause.kind )) }) - .map(MergeIntoClauseKind::from_proto)?; + .map(MergeIntoClauseKind::from)?; let predicate = clause .predicate .as_ref() @@ -382,15 +302,6 @@ fn parse_merge_into_action( }) } -impl FromProto for NullTreatment { - fn from_proto(t: protobuf::NullTreatment) -> Self { - match t { - protobuf::NullTreatment::RespectNulls => NullTreatment::RespectNulls, - protobuf::NullTreatment::IgnoreNulls => NullTreatment::IgnoreNulls, - } - } -} - pub fn parse_expr( proto: &protobuf::LogicalExprNode, ctx: &TaskContext, @@ -439,7 +350,7 @@ pub fn parse_expr( .window_frame .as_ref() .map::, _>(|window_frame| { - let window_frame = WindowFrame::try_from_proto(window_frame.clone())?; + let window_frame = WindowFrame::try_from(window_frame.clone())?; window_frame .regularize_order_bys(&mut order_by) .map(|_| window_frame) @@ -457,7 +368,7 @@ pub fn parse_expr( "Received a WindowExprNode message with unknown NullTreatment {null_treatment}", )) })?; - Some(NullTreatment::from_proto(null_treatment)) + Some(NullTreatment::from(null_treatment)) } None => None, }; @@ -766,7 +677,7 @@ pub fn parse_expr( "Received an AggregateUdfExprNode message with unknown NullTreatment {null_treatment}", )) })?; - Some(NullTreatment::from_proto(null_treatment)) + Some(NullTreatment::from(null_treatment)) } None => None, }; diff --git a/datafusion/proto/src/logical_plan/to_proto.rs b/datafusion/proto/src/logical_plan/to_proto.rs index 67c815add8460..82fdfcc5ee291 100644 --- a/datafusion/proto/src/logical_plan/to_proto.rs +++ b/datafusion/proto/src/logical_plan/to_proto.rs @@ -20,19 +20,16 @@ //! processes. use datafusion_common::{NullEquality, SplitPoint, TableReference, UnnestOptions}; -use datafusion_expr::dml::{ - MergeIntoAction, MergeIntoClause, MergeIntoClauseKind, MergeIntoOp, -}; +use datafusion_expr::dml::{MergeIntoAction, MergeIntoClause, MergeIntoOp}; use datafusion_expr::expr::{ self, AggregateFunctionParams, Alias, Between, BinaryExpr, Cast, GroupingSet, - HigherOrderFunction, InList, Lambda, LambdaVariable, Like, NullTreatment, - Placeholder, ScalarFunction, Unnest, + HigherOrderFunction, InList, Lambda, LambdaVariable, Like, Placeholder, + ScalarFunction, Unnest, }; use datafusion_expr::logical_plan::Subquery; use datafusion_expr::{ - Expr, JoinConstraint, JoinType, SortExpr, TryCast, WindowFrame, WindowFrameBound, - WindowFrameUnits, WindowFunctionDefinition, logical_plan::PlanType, - logical_plan::StringifiedPlan, + Expr, JoinConstraint, JoinType, SortExpr, TryCast, WindowFunctionDefinition, + logical_plan::PlanType, logical_plan::StringifiedPlan, }; use crate::protobuf::RecursionUnnestOption; @@ -50,7 +47,7 @@ use crate::protobuf::{ }; use super::{AsLogicalPlan, LogicalExtensionCodec}; -use crate::convert::{FromProto, TryFromProto}; +use crate::convert::FromProto; use crate::protobuf::LogicalPlanNode; impl FromProto<&UnnestOptions> for protobuf::UnnestOptions { @@ -140,55 +137,6 @@ impl FromProto<&StringifiedPlan> for protobuf::StringifiedPlan { } } -impl FromProto for protobuf::WindowFrameUnits { - fn from_proto(units: WindowFrameUnits) -> Self { - match units { - WindowFrameUnits::Rows => Self::Rows, - WindowFrameUnits::Range => Self::Range, - WindowFrameUnits::Groups => Self::Groups, - } - } -} - -impl TryFromProto<&WindowFrameBound> for protobuf::WindowFrameBound { - type Error = Error; - - fn try_from_proto(bound: &WindowFrameBound) -> Result { - Ok(match bound { - WindowFrameBound::CurrentRow => Self { - window_frame_bound_type: protobuf::WindowFrameBoundType::CurrentRow - .into(), - bound_value: None, - }, - WindowFrameBound::Preceding(v) => Self { - window_frame_bound_type: protobuf::WindowFrameBoundType::Preceding.into(), - bound_value: Some(v.try_into()?), - }, - WindowFrameBound::Following(v) => Self { - window_frame_bound_type: protobuf::WindowFrameBoundType::Following.into(), - bound_value: Some(v.try_into()?), - }, - }) - } -} - -impl TryFromProto<&WindowFrame> for protobuf::WindowFrame { - type Error = Error; - - fn try_from_proto(window: &WindowFrame) -> Result { - Ok(Self { - window_frame_units: protobuf::WindowFrameUnits::from_proto(window.units) - .into(), - start_bound: Some(protobuf::WindowFrameBound::try_from_proto( - &window.start_bound, - )?), - end_bound: Some(protobuf::window_frame::EndBound::Bound( - protobuf::WindowFrameBound::try_from_proto(&window.end_bound)?, - )), - }) - } -} - pub fn serialize_exprs<'a, I>( exprs: I, codec: &dyn LogicalExtensionCodec, @@ -352,7 +300,7 @@ pub fn serialize_expr( let partition_by = serialize_exprs(partition_by, codec)?; let order_by = serialize_sorts(order_by, codec)?; - let window_frame = Some(protobuf::WindowFrame::try_from_proto(window_frame)?); + let window_frame = Some(protobuf::WindowFrame::try_from(window_frame)?); let window_expr = protobuf::WindowExprNode { exprs: serialize_exprs(args, codec)?, @@ -366,7 +314,7 @@ pub fn serialize_expr( None => None, }, null_treatment: null_treatment - .map(|nt| protobuf::NullTreatment::from_proto(nt).into()), + .map(|nt| protobuf::NullTreatment::from(nt).into()), fun_definition, }; protobuf::LogicalExprNode { @@ -399,7 +347,7 @@ pub fn serialize_expr( order_by: serialize_sorts(order_by, codec)?, fun_definition: (!buf.is_empty()).then_some(buf), null_treatment: null_treatment - .map(|nt| protobuf::NullTreatment::from_proto(nt).into()), + .map(|nt| protobuf::NullTreatment::from(nt).into()), }, ))), } @@ -799,25 +747,6 @@ impl FromProto for protobuf::NullEquality { } } -impl FromProto for protobuf::merge_into_clause_node::Kind { - fn from_proto(k: MergeIntoClauseKind) -> Self { - match k { - MergeIntoClauseKind::Matched => { - protobuf::merge_into_clause_node::Kind::Matched - } - MergeIntoClauseKind::NotMatched => { - protobuf::merge_into_clause_node::Kind::NotMatched - } - MergeIntoClauseKind::NotMatchedByTarget => { - protobuf::merge_into_clause_node::Kind::NotMatchedByTarget - } - MergeIntoClauseKind::NotMatchedBySource => { - protobuf::merge_into_clause_node::Kind::NotMatchedBySource - } - } - } -} - pub fn serialize_merge_into_op( op: &MergeIntoOp, codec: &dyn LogicalExtensionCodec, @@ -836,7 +765,7 @@ fn serialize_merge_into_clause( clause: &MergeIntoClause, codec: &dyn LogicalExtensionCodec, ) -> Result { - let kind = protobuf::merge_into_clause_node::Kind::from_proto(clause.kind); + let kind = protobuf::merge_into_clause_node::Kind::from(clause.kind); let predicate = clause .predicate .as_ref() @@ -884,12 +813,3 @@ fn serialize_merge_into_action( action: Some(action), }) } - -impl FromProto for protobuf::NullTreatment { - fn from_proto(t: NullTreatment) -> Self { - match t { - NullTreatment::RespectNulls => protobuf::NullTreatment::RespectNulls, - NullTreatment::IgnoreNulls => protobuf::NullTreatment::IgnoreNulls, - } - } -} diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs index 476cd8aba4bbd..4c6b172e84d0e 100644 --- a/datafusion/proto/src/physical_plan/from_proto.rs +++ b/datafusion/proto/src/physical_plan/from_proto.rs @@ -48,7 +48,6 @@ use super::{ ConverterPlanDecoder, DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalPlanDecodeContext, PhysicalProtoConverterExtension, }; -use crate::convert::TryFromProto; use crate::protobuf::physical_expr_node::ExprType; use crate::{convert_required, protobuf}; use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr; @@ -137,7 +136,7 @@ pub fn parse_physical_window_expr( let window_frame = proto .window_frame .as_ref() - .map(|wf| datafusion_expr::WindowFrame::try_from_proto(wf.clone())) + .map(|wf| datafusion_expr::WindowFrame::try_from(wf.clone())) .transpose() .map_err(|e| internal_datafusion_err!("{e}"))? .ok_or_else(|| { From a36fb674b9b7946c29056bc9c393eca79e07a6c8 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sun, 9 Aug 2026 16:14:27 -0400 Subject: [PATCH 3/7] refactor(proto): restore common-type conversions as From/TryFrom in proto-models MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `UnnestOptions`, `TableReference`, `StringifiedPlan`, `JoinType`, `JoinConstraint`, `NullEquality`, `CsvOptions`, `JsonOptions` and the parquet options types all live in `datafusion-common`, which sits below `datafusion-proto-models` in the crate graph and so cannot host the impls. They move onto the local proto type in `datafusion-proto-models` instead — the arrangement `datafusion-proto-common` already uses for `ScalarValue` and `Statistics` — and go back to being plain `From` / `TryFrom`, the shape they had in 54.1.0. This is also the only crate that can express `TryFrom<&[PartitionedFile]> for protobuf::FileGroup`. `datafusion-datasource` cannot: `&T` is `#[fundamental]` but `[T]` is not, so `&[PartitionedFile]` counts as foreign there. Here the *self* type is local, which is all the orphan rule needs, and staying generic over the element (`&T: TryInto`) means this crate never has to name `PartitionedFile`, which sits above it. Callers get the 54.1.0 spelling back verbatim. New `datafusion_proto_models::{from_proto, to_proto}` modules; the crate gains a direct `datafusion-common` dependency (already present transitively via `datafusion-proto-common`). Part of #24019. Co-Authored-By: Claude Opus 5 --- Cargo.lock | 1 + datafusion/datasource/src/proto.rs | 19 + datafusion/proto-models/Cargo.toml | 1 + datafusion/proto-models/src/from_proto.rs | 519 ++++++++++++++++++ datafusion/proto-models/src/lib.rs | 13 +- datafusion/proto-models/src/to_proto.rs | 325 +++++++++++ .../proto/src/logical_plan/file_formats.rs | 357 +----------- .../proto/src/logical_plan/from_proto.rs | 160 +----- datafusion/proto/src/logical_plan/mod.rs | 41 +- datafusion/proto/src/logical_plan/to_proto.rs | 181 +----- .../tests/cases/roundtrip_logical_plan.rs | 6 +- 11 files changed, 913 insertions(+), 710 deletions(-) create mode 100644 datafusion/proto-models/src/from_proto.rs create mode 100644 datafusion/proto-models/src/to_proto.rs diff --git a/Cargo.lock b/Cargo.lock index 66d67fb6417ec..a3ce12e21ae4d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2567,6 +2567,7 @@ dependencies = [ name = "datafusion-proto-models" version = "54.1.0" dependencies = [ + "datafusion-common", "datafusion-proto-common", "pbjson 0.9.0", "prost", diff --git a/datafusion/datasource/src/proto.rs b/datafusion/datasource/src/proto.rs index 0f183e23a1ca5..6dc6e2ee45ffd 100644 --- a/datafusion/datasource/src/proto.rs +++ b/datafusion/datasource/src/proto.rs @@ -264,6 +264,25 @@ mod tests { ); } + #[test] + fn file_group_from_slice_matches_file_group() -> Result<()> { + // `protobuf::FileGroup: TryFrom<&[T]>` lives in `datafusion-proto-models`, + // generic over the element so that crate never names `PartitionedFile`. + // This is the caller-visible half: the bound resolves via + // `TryFrom<&PartitionedFile> for protobuf::PartitionedFile` above. + let files = vec![ + PartitionedFile::new("a.parquet", 1), + PartitionedFile::new("b.parquet", 2), + ]; + + let from_slice = protobuf::FileGroup::try_from(&files[..])?; + let from_group = protobuf::FileGroup::try_from(&FileGroup::new(files))?; + + assert_eq!(from_slice, from_group); + assert_eq!(from_slice.files.len(), 2); + Ok(()) + } + #[test] fn file_group_roundtrip() -> Result<()> { let group = FileGroup::new(vec![ diff --git a/datafusion/proto-models/Cargo.toml b/datafusion/proto-models/Cargo.toml index d8cf5fcdc3dce..83b1a202c24ed 100644 --- a/datafusion/proto-models/Cargo.toml +++ b/datafusion/proto-models/Cargo.toml @@ -45,6 +45,7 @@ default = [] json = ["serde", "pbjson", "datafusion-proto-common/json"] [dependencies] +datafusion-common = { workspace = true } datafusion-proto-common = { workspace = true } pbjson = { workspace = true, optional = true } prost = { workspace = true } diff --git a/datafusion/proto-models/src/from_proto.rs b/datafusion/proto-models/src/from_proto.rs new file mode 100644 index 0000000000000..74ead8c52049b --- /dev/null +++ b/datafusion/proto-models/src/from_proto.rs @@ -0,0 +1,519 @@ +// 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. + +//! Conversions from the protobuf messages in this crate to their +//! `datafusion-common` counterparts. +//! +//! The DataFusion side of these conversions lives *below* this crate in the +//! dependency graph, so it cannot host the impls itself. They live here +//! instead, on the local proto type — the same arrangement +//! `datafusion-proto-common` uses for `ScalarValue` and `Statistics`. + +use std::sync::Arc; + +use datafusion_common::config::{ + CsvOptions, JsonOptions, MaxRowGroupBytes, ParquetCdcOptions, ParquetColumnOptions, + ParquetOptions, TableParquetOptions, +}; +use datafusion_common::display::{PlanType, StringifiedPlan}; +use datafusion_common::parsers::{CompressionTypeVariant, CsvQuoteStyle}; +use datafusion_common::{ + JoinConstraint, JoinType, NullEquality, RecursionUnnestOption, TableReference, + UnnestOptions, +}; +use datafusion_proto_common::FromProtoError as Error; + +use crate::protobuf::{ + self, AnalyzedLogicalPlanType, CsvOptions as CsvOptionsProto, + CsvQuoteStyle as CsvQuoteStyleProto, JsonOptions as JsonOptionsProto, + OptimizedLogicalPlanType, OptimizedPhysicalPlanType, + ParquetCdcOptions as ParquetCdcOptionsProto, + ParquetColumnOptions as ParquetColumnOptionsProto, + ParquetOptions as ParquetOptionsProto, + TableParquetOptions as TableParquetOptionsProto, parquet_column_options, + parquet_options, + plan_type::PlanTypeEnum::{ + AnalyzedLogicalPlan, FinalAnalyzedLogicalPlan, FinalLogicalPlan, + FinalPhysicalPlan, FinalPhysicalPlanWithSchema, FinalPhysicalPlanWithStats, + InitialLogicalPlan, InitialPhysicalPlan, InitialPhysicalPlanWithSchema, + InitialPhysicalPlanWithStats, OptimizedLogicalPlan, OptimizedPhysicalPlan, + PhysicalPlanError, + }, +}; + +impl From<&protobuf::UnnestOptions> for UnnestOptions { + fn from(opts: &protobuf::UnnestOptions) -> Self { + use datafusion_common::NullHandling; + use protobuf::unnest_options::NullHandling as ProtoNullHandling; + let null_handling = match ProtoNullHandling::try_from(opts.null_handling) { + Ok(ProtoNullHandling::Preserve) => NullHandling::Preserve, + Ok(ProtoNullHandling::Drop) => NullHandling::Drop, + Ok(ProtoNullHandling::PreserveAndExpandEmpty) => { + NullHandling::PreserveAndExpandEmpty + } + // Unknown enum values fall back to the default (Preserve), which + // matches DataFusion's historical behavior. + Err(_) => NullHandling::Preserve, + }; + Self { + null_handling, + recursions: opts + .recursions + .iter() + .map(|r| RecursionUnnestOption { + input_column: r.input_column.as_ref().unwrap().into(), + output_column: r.output_column.as_ref().unwrap().into(), + depth: r.depth as usize, + }) + .collect::>(), + } + } +} + +impl TryFrom for TableReference { + type Error = Error; + + fn try_from(value: protobuf::TableReference) -> Result { + use protobuf::table_reference::TableReferenceEnum; + let table_reference_enum = value + .table_reference_enum + .ok_or_else(|| Error::required("table_reference_enum"))?; + + match table_reference_enum { + TableReferenceEnum::Bare(protobuf::BareTableReference { table }) => { + Ok(TableReference::bare(table)) + } + TableReferenceEnum::Partial(protobuf::PartialTableReference { + schema, + table, + }) => Ok(TableReference::partial(schema, table)), + TableReferenceEnum::Full(protobuf::FullTableReference { + catalog, + schema, + table, + }) => Ok(TableReference::full(catalog, schema, table)), + } + } +} + +impl From<&protobuf::StringifiedPlan> for StringifiedPlan { + fn from(stringified_plan: &protobuf::StringifiedPlan) -> Self { + Self { + plan_type: match stringified_plan + .plan_type + .as_ref() + .and_then(|pt| pt.plan_type_enum.as_ref()) + .unwrap_or_else(|| { + panic!( + "Cannot create protobuf::StringifiedPlan from {stringified_plan:?}" + ) + }) { + InitialLogicalPlan(_) => PlanType::InitialLogicalPlan, + AnalyzedLogicalPlan(AnalyzedLogicalPlanType { analyzer_name }) => { + PlanType::AnalyzedLogicalPlan { + analyzer_name:analyzer_name.clone() + } + } + FinalAnalyzedLogicalPlan(_) => PlanType::FinalAnalyzedLogicalPlan, + OptimizedLogicalPlan(OptimizedLogicalPlanType { optimizer_name }) => { + PlanType::OptimizedLogicalPlan { + optimizer_name: optimizer_name.clone(), + } + } + FinalLogicalPlan(_) => PlanType::FinalLogicalPlan, + InitialPhysicalPlan(_) => PlanType::InitialPhysicalPlan, + InitialPhysicalPlanWithStats(_) => PlanType::InitialPhysicalPlanWithStats, + InitialPhysicalPlanWithSchema(_) => PlanType::InitialPhysicalPlanWithSchema, + OptimizedPhysicalPlan(OptimizedPhysicalPlanType { optimizer_name }) => { + PlanType::OptimizedPhysicalPlan { + optimizer_name: optimizer_name.clone(), + } + } + FinalPhysicalPlan(_) => PlanType::FinalPhysicalPlan, + FinalPhysicalPlanWithStats(_) => PlanType::FinalPhysicalPlanWithStats, + FinalPhysicalPlanWithSchema(_) => PlanType::FinalPhysicalPlanWithSchema, + PhysicalPlanError(_) => PlanType::PhysicalPlanError, + }, + plan: Arc::new(stringified_plan.plan.clone()), + } + } +} + +impl From for JoinType { + fn from(t: protobuf::JoinType) -> Self { + match t { + protobuf::JoinType::Inner => JoinType::Inner, + protobuf::JoinType::Left => JoinType::Left, + protobuf::JoinType::Right => JoinType::Right, + protobuf::JoinType::Full => JoinType::Full, + protobuf::JoinType::Leftsemi => JoinType::LeftSemi, + protobuf::JoinType::Rightsemi => JoinType::RightSemi, + protobuf::JoinType::Leftanti => JoinType::LeftAnti, + protobuf::JoinType::Rightanti => JoinType::RightAnti, + protobuf::JoinType::Leftmark => JoinType::LeftMark, + protobuf::JoinType::Rightmark => JoinType::RightMark, + } + } +} + +impl From for JoinConstraint { + fn from(t: protobuf::JoinConstraint) -> Self { + match t { + protobuf::JoinConstraint::On => JoinConstraint::On, + protobuf::JoinConstraint::Using => JoinConstraint::Using, + } + } +} + +impl From for NullEquality { + fn from(t: protobuf::NullEquality) -> Self { + match t { + protobuf::NullEquality::NullEqualsNothing => NullEquality::NullEqualsNothing, + protobuf::NullEquality::NullEqualsNull => NullEquality::NullEqualsNull, + } + } +} + +impl From<&CsvOptionsProto> for CsvOptions { + fn from(proto: &CsvOptionsProto) -> Self { + CsvOptions { + has_header: if !proto.has_header.is_empty() { + Some(proto.has_header[0] != 0) + } else { + None + }, + delimiter: proto.delimiter.first().copied().unwrap_or(b','), + quote: proto.quote.first().copied().unwrap_or(b'"'), + terminator: if !proto.terminator.is_empty() { + Some(proto.terminator[0]) + } else { + None + }, + escape: if !proto.escape.is_empty() { + Some(proto.escape[0]) + } else { + None + }, + double_quote: if !proto.double_quote.is_empty() { + Some(proto.double_quote[0] != 0) + } else { + None + }, + compression: match proto.compression { + 0 => CompressionTypeVariant::GZIP, + 1 => CompressionTypeVariant::BZIP2, + 2 => CompressionTypeVariant::XZ, + 3 => CompressionTypeVariant::ZSTD, + _ => CompressionTypeVariant::UNCOMPRESSED, + }, + schema_infer_max_rec: proto.schema_infer_max_rec.map(|v| v as usize), + date_format: if proto.date_format.is_empty() { + None + } else { + Some(proto.date_format.clone()) + }, + datetime_format: if proto.datetime_format.is_empty() { + None + } else { + Some(proto.datetime_format.clone()) + }, + timestamp_format: if proto.timestamp_format.is_empty() { + None + } else { + Some(proto.timestamp_format.clone()) + }, + timestamp_tz_format: if proto.timestamp_tz_format.is_empty() { + None + } else { + Some(proto.timestamp_tz_format.clone()) + }, + time_format: if proto.time_format.is_empty() { + None + } else { + Some(proto.time_format.clone()) + }, + null_value: if proto.null_value.is_empty() { + None + } else { + Some(proto.null_value.clone()) + }, + null_regex: if proto.null_regex.is_empty() { + None + } else { + Some(proto.null_regex.clone()) + }, + comment: if !proto.comment.is_empty() { + Some(proto.comment[0]) + } else { + None + }, + newlines_in_values: if proto.newlines_in_values.is_empty() { + None + } else { + Some(proto.newlines_in_values[0] != 0) + }, + truncated_rows: if proto.truncated_rows.is_empty() { + None + } else { + Some(proto.truncated_rows[0] != 0) + }, + compression_level: proto.compression_level, + quote_style: match CsvQuoteStyleProto::try_from(proto.quote_style) { + Ok(CsvQuoteStyleProto::Always) => CsvQuoteStyle::Always, + Ok(CsvQuoteStyleProto::NonNumeric) => CsvQuoteStyle::NonNumeric, + Ok(CsvQuoteStyleProto::Never) => CsvQuoteStyle::Never, + Ok(CsvQuoteStyleProto::Necessary) => CsvQuoteStyle::Necessary, + _ => CsvQuoteStyle::Necessary, + }, + ignore_leading_whitespace: if proto.ignore_leading_whitespace.is_empty() { + None + } else { + Some(proto.ignore_leading_whitespace[0] != 0) + }, + ignore_trailing_whitespace: if proto.ignore_trailing_whitespace.is_empty() { + None + } else { + Some(proto.ignore_trailing_whitespace[0] != 0) + }, + } + } +} + +impl From<&JsonOptionsProto> for JsonOptions { + fn from(proto: &JsonOptionsProto) -> Self { + JsonOptions { + compression: match proto.compression { + 0 => CompressionTypeVariant::GZIP, + 1 => CompressionTypeVariant::BZIP2, + 2 => CompressionTypeVariant::XZ, + 3 => CompressionTypeVariant::ZSTD, + _ => CompressionTypeVariant::UNCOMPRESSED, + }, + schema_infer_max_rec: proto.schema_infer_max_rec.map(|v| v as usize), + compression_level: proto.compression_level, + newline_delimited: proto.newline_delimited.unwrap_or(true), + } + } +} + +impl From for ParquetCdcOptions { + fn from(value: ParquetCdcOptionsProto) -> Self { + ParquetCdcOptions { + enabled: value.enabled, + min_chunk_size: value.min_chunk_size as usize, + max_chunk_size: value.max_chunk_size as usize, + norm_level: value.norm_level, + } + } +} + +impl TryFrom<&ParquetOptionsProto> for ParquetOptions { + type Error = datafusion_common::DataFusionError; + + fn try_from( + proto: &ParquetOptionsProto, + ) -> datafusion_common::Result { + let writer_version = match proto.writer_version.as_str() { + // Proto3 decodes an omitted string field as the empty string. The + // schema documents writer_version's logical default as "1.0", so + // preserve that default when the field is absent on the wire. + "" => ParquetOptions::default().writer_version, + version => version.parse()?, + }; + + Ok(ParquetOptions { + enable_page_index: proto.enable_page_index, + pruning: proto.pruning, + skip_metadata: proto.skip_metadata, + metadata_size_hint: proto + .metadata_size_hint_opt + .as_ref() + .map(|opt| match opt { + parquet_options::MetadataSizeHintOpt::MetadataSizeHint(size) => { + *size as usize + } + }), + pushdown_filters: proto.pushdown_filters, + reorder_filters: proto.reorder_filters, + force_filter_selections: proto.force_filter_selections, + data_pagesize_limit: proto.data_pagesize_limit as usize, + write_batch_size: proto.write_batch_size as usize, + writer_version, + compression: proto.compression_opt.as_ref().map(|opt| match opt { + parquet_options::CompressionOpt::Compression(compression) => { + compression.clone() + } + }), + dictionary_enabled: proto.dictionary_enabled_opt.as_ref().map(|opt| { + match opt { + parquet_options::DictionaryEnabledOpt::DictionaryEnabled( + enabled, + ) => *enabled, + } + }), + dictionary_page_size_limit: proto.dictionary_page_size_limit as usize, + statistics_enabled: proto.statistics_enabled_opt.as_ref().map( + |opt| match opt { + parquet_options::StatisticsEnabledOpt::StatisticsEnabled( + statistics, + ) => statistics.clone(), + }, + ), + max_row_group_size: proto.max_row_group_size as usize, + max_in_list_size: proto.max_in_list_size as usize, + created_by: proto.created_by.clone(), + column_index_truncate_length: proto + .column_index_truncate_length_opt + .as_ref() + .map(|opt| match opt { + parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(length) => *length as usize, + }), + statistics_truncate_length: proto + .statistics_truncate_length_opt + .as_ref() + .map(|opt| match opt { + parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(length) => *length as usize, + }), + data_page_row_count_limit: proto.data_page_row_count_limit as usize, + encoding: proto.encoding_opt.as_ref().map(|opt| match opt { + parquet_options::EncodingOpt::Encoding(encoding) => { + encoding.clone() + } + }), + bloom_filter_on_read: proto.bloom_filter_on_read, + bloom_filter_on_write: proto.bloom_filter_on_write, + bloom_filter_fpp: proto + .bloom_filter_fpp_opt + .as_ref() + .map(|opt| match opt { + parquet_options::BloomFilterFppOpt::BloomFilterFpp(fpp) => *fpp, + }), + bloom_filter_ndv: proto + .bloom_filter_ndv_opt + .as_ref() + .map(|opt| match opt { + parquet_options::BloomFilterNdvOpt::BloomFilterNdv(ndv) => *ndv, + }), + allow_single_file_parallelism: proto.allow_single_file_parallelism, + maximum_parallel_row_group_writers: proto + .maximum_parallel_row_group_writers + as usize, + maximum_buffered_record_batches_per_stream: proto + .maximum_buffered_record_batches_per_stream + as usize, + schema_force_view_types: proto.schema_force_view_types, + binary_as_string: proto.binary_as_string, + skip_arrow_metadata: proto.skip_arrow_metadata, + coerce_int96: proto.coerce_int96_opt.as_ref().map(|opt| match opt { + parquet_options::CoerceInt96Opt::CoerceInt96(coerce_int96) => { + coerce_int96.clone() + } + }), + coerce_int96_tz: proto + .coerce_int96_tz_opt + .as_ref() + .map(|opt| match opt { + parquet_options::CoerceInt96TzOpt::CoerceInt96Tz(tz) => { + tz.clone() + } + }), + max_predicate_cache_size: proto + .max_predicate_cache_size_opt + .as_ref() + .map(|opt| match opt { + parquet_options::MaxPredicateCacheSizeOpt::MaxPredicateCacheSize( + size, + ) => *size as usize, + }), + max_row_group_bytes: proto + .max_row_group_bytes_opt + .as_ref() + .and_then(|opt| match opt { + parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(size) => { + MaxRowGroupBytes::try_new(*size as usize).ok() + } + }), + content_defined_chunking: proto + .content_defined_chunking + .map(ParquetCdcOptions::from) + .unwrap_or_default(), + }) + } +} + +impl From for ParquetColumnOptions { + fn from(proto: ParquetColumnOptionsProto) -> Self { + ParquetColumnOptions { + bloom_filter_enabled: proto.bloom_filter_enabled_opt.map( + |parquet_column_options::BloomFilterEnabledOpt::BloomFilterEnabled(v)| v, + ), + encoding: proto + .encoding_opt + .map(|parquet_column_options::EncodingOpt::Encoding(v)| v), + dictionary_enabled: proto.dictionary_enabled_opt.map( + |parquet_column_options::DictionaryEnabledOpt::DictionaryEnabled(v)| v, + ), + compression: proto + .compression_opt + .map(|parquet_column_options::CompressionOpt::Compression(v)| v), + statistics_enabled: proto.statistics_enabled_opt.map( + |parquet_column_options::StatisticsEnabledOpt::StatisticsEnabled(v)| v, + ), + bloom_filter_fpp: proto + .bloom_filter_fpp_opt + .map(|parquet_column_options::BloomFilterFppOpt::BloomFilterFpp(v)| v), + bloom_filter_ndv: proto + .bloom_filter_ndv_opt + .map(|parquet_column_options::BloomFilterNdvOpt::BloomFilterNdv(v)| v), + } + } +} + +impl TryFrom<&TableParquetOptionsProto> for TableParquetOptions { + type Error = datafusion_common::DataFusionError; + + fn try_from( + proto: &TableParquetOptionsProto, + ) -> datafusion_common::Result { + Ok(TableParquetOptions { + global: proto + .global + .as_ref() + .map(ParquetOptions::try_from) + .transpose()? + .unwrap_or_default(), + column_specific_options: proto + .column_specific_options + .iter() + .map(|parquet_column_options| { + ( + parquet_column_options.column_name.clone(), + ParquetColumnOptions::from( + parquet_column_options.options.clone().unwrap_or_default(), + ), + ) + }) + .collect(), + key_value_metadata: proto + .key_value_metadata + .iter() + .map(|(k, v)| (k.clone(), Some(v.clone()))) + .collect(), + ..Default::default() + }) + } +} diff --git a/datafusion/proto-models/src/lib.rs b/datafusion/proto-models/src/lib.rs index 8f845a8a99ca1..3276c0811e2c5 100644 --- a/datafusion/proto-models/src/lib.rs +++ b/datafusion/proto-models/src/lib.rs @@ -26,10 +26,13 @@ //! `prost`-generated DataFusion protobuf model types. //! -//! This crate contains only the generated structs for DataFusion's logical and -//! physical plan protobuf schemas (see `proto/datafusion.proto`). It is the -//! schema source of truth for [`datafusion-proto`] and intentionally has no -//! DataFusion dependencies beyond [`datafusion-proto-common`]. +//! This crate contains the generated structs for DataFusion's logical and +//! physical plan protobuf schemas (see `proto/datafusion.proto`), plus the +//! [`From`] / [`TryFrom`] conversions between them and the `datafusion-common` +//! types they mirror. Those conversions live here because their DataFusion side +//! sits *below* this crate in the dependency graph and so cannot host the impls +//! itself — see [`from_proto`] and [`to_proto`]. It is the schema source of +//! truth for [`datafusion-proto`]. //! //! Most users should depend on [`datafusion-proto`] instead, which re-exports //! these types under [`datafusion_proto::protobuf`]. @@ -38,7 +41,9 @@ //! [`datafusion-proto-common`]: https://crates.io/crates/datafusion-proto-common //! [`datafusion_proto::protobuf`]: https://docs.rs/datafusion-proto/latest/datafusion_proto/protobuf/index.html +pub mod from_proto; pub mod generated; +pub mod to_proto; /// All DataFusion protobuf model types. /// diff --git a/datafusion/proto-models/src/to_proto.rs b/datafusion/proto-models/src/to_proto.rs new file mode 100644 index 0000000000000..d1c857a7c5cba --- /dev/null +++ b/datafusion/proto-models/src/to_proto.rs @@ -0,0 +1,325 @@ +// 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. + +//! Conversions from `datafusion-common` types to the protobuf messages in this +//! crate. +//! +//! See [`crate::from_proto`] for why the impls live here rather than next to +//! the DataFusion types. + +use datafusion_common::DataFusionError; +use datafusion_common::display::{PlanType, StringifiedPlan}; +use datafusion_common::{ + JoinConstraint, JoinType, NullEquality, TableReference, UnnestOptions, +}; + +use crate::generated::datafusion_common::EmptyMessage; +use crate::protobuf::{ + self, AnalyzedLogicalPlanType, OptimizedLogicalPlanType, OptimizedPhysicalPlanType, + RecursionUnnestOption, + plan_type::PlanTypeEnum::{ + AnalyzedLogicalPlan, FinalAnalyzedLogicalPlan, FinalLogicalPlan, + FinalPhysicalPlan, FinalPhysicalPlanWithSchema, FinalPhysicalPlanWithStats, + InitialLogicalPlan, InitialPhysicalPlan, InitialPhysicalPlanWithSchema, + InitialPhysicalPlanWithStats, OptimizedLogicalPlan, OptimizedPhysicalPlan, + PhysicalPlanError, + }, +}; + +impl From<&UnnestOptions> for protobuf::UnnestOptions { + fn from(opts: &UnnestOptions) -> Self { + use datafusion_common::NullHandling; + use protobuf::unnest_options::NullHandling as ProtoNullHandling; + let null_handling = match opts.null_handling { + NullHandling::Preserve => ProtoNullHandling::Preserve, + NullHandling::Drop => ProtoNullHandling::Drop, + NullHandling::PreserveAndExpandEmpty => { + ProtoNullHandling::PreserveAndExpandEmpty + } + } as i32; + Self { + null_handling, + recursions: opts + .recursions + .iter() + .map(|r| RecursionUnnestOption { + input_column: Some((&r.input_column).into()), + output_column: Some((&r.output_column).into()), + depth: r.depth as u32, + }) + .collect(), + } + } +} + +impl From<&StringifiedPlan> for protobuf::StringifiedPlan { + fn from(stringified_plan: &StringifiedPlan) -> Self { + Self { + plan_type: match stringified_plan.clone().plan_type { + PlanType::InitialLogicalPlan => Some(protobuf::PlanType { + plan_type_enum: Some(InitialLogicalPlan(EmptyMessage {})), + }), + PlanType::AnalyzedLogicalPlan { analyzer_name } => { + Some(protobuf::PlanType { + plan_type_enum: Some(AnalyzedLogicalPlan( + AnalyzedLogicalPlanType { analyzer_name }, + )), + }) + } + PlanType::FinalAnalyzedLogicalPlan => Some(protobuf::PlanType { + plan_type_enum: Some(FinalAnalyzedLogicalPlan(EmptyMessage {})), + }), + PlanType::OptimizedLogicalPlan { optimizer_name } => { + Some(protobuf::PlanType { + plan_type_enum: Some(OptimizedLogicalPlan( + OptimizedLogicalPlanType { optimizer_name }, + )), + }) + } + PlanType::FinalLogicalPlan => Some(protobuf::PlanType { + plan_type_enum: Some(FinalLogicalPlan(EmptyMessage {})), + }), + PlanType::InitialPhysicalPlan => Some(protobuf::PlanType { + plan_type_enum: Some(InitialPhysicalPlan(EmptyMessage {})), + }), + PlanType::OptimizedPhysicalPlan { optimizer_name } => { + Some(protobuf::PlanType { + plan_type_enum: Some(OptimizedPhysicalPlan( + OptimizedPhysicalPlanType { optimizer_name }, + )), + }) + } + PlanType::FinalPhysicalPlan => Some(protobuf::PlanType { + plan_type_enum: Some(FinalPhysicalPlan(EmptyMessage {})), + }), + PlanType::InitialPhysicalPlanWithStats => Some(protobuf::PlanType { + plan_type_enum: Some(InitialPhysicalPlanWithStats(EmptyMessage {})), + }), + PlanType::InitialPhysicalPlanWithSchema => Some(protobuf::PlanType { + plan_type_enum: Some(InitialPhysicalPlanWithSchema(EmptyMessage {})), + }), + PlanType::FinalPhysicalPlanWithStats => Some(protobuf::PlanType { + plan_type_enum: Some(FinalPhysicalPlanWithStats(EmptyMessage {})), + }), + PlanType::FinalPhysicalPlanWithSchema => Some(protobuf::PlanType { + plan_type_enum: Some(FinalPhysicalPlanWithSchema(EmptyMessage {})), + }), + PlanType::PhysicalPlanError => Some(protobuf::PlanType { + plan_type_enum: Some(PhysicalPlanError(EmptyMessage {})), + }), + }, + plan: stringified_plan.plan.to_string(), + } + } +} + +impl From for protobuf::TableReference { + fn from(t: TableReference) -> Self { + use protobuf::table_reference::TableReferenceEnum; + let table_reference_enum = match t { + TableReference::Bare { table } => { + TableReferenceEnum::Bare(protobuf::BareTableReference { + table: table.to_string(), + }) + } + TableReference::Partial { schema, table } => { + TableReferenceEnum::Partial(protobuf::PartialTableReference { + schema: schema.to_string(), + table: table.to_string(), + }) + } + TableReference::Full { + catalog, + schema, + table, + } => TableReferenceEnum::Full(protobuf::FullTableReference { + catalog: catalog.to_string(), + schema: schema.to_string(), + table: table.to_string(), + }), + }; + + protobuf::TableReference { + table_reference_enum: Some(table_reference_enum), + } + } +} + +impl From for protobuf::JoinType { + fn from(t: JoinType) -> Self { + match t { + JoinType::Inner => protobuf::JoinType::Inner, + JoinType::Left => protobuf::JoinType::Left, + JoinType::Right => protobuf::JoinType::Right, + JoinType::Full => protobuf::JoinType::Full, + JoinType::LeftSemi => protobuf::JoinType::Leftsemi, + JoinType::RightSemi => protobuf::JoinType::Rightsemi, + JoinType::LeftAnti => protobuf::JoinType::Leftanti, + JoinType::RightAnti => protobuf::JoinType::Rightanti, + JoinType::LeftMark => protobuf::JoinType::Leftmark, + JoinType::RightMark => protobuf::JoinType::Rightmark, + } + } +} + +impl From for protobuf::JoinConstraint { + fn from(t: JoinConstraint) -> Self { + match t { + JoinConstraint::On => protobuf::JoinConstraint::On, + JoinConstraint::Using => protobuf::JoinConstraint::Using, + } + } +} + +impl From for protobuf::NullEquality { + fn from(t: NullEquality) -> Self { + match t { + NullEquality::NullEqualsNothing => protobuf::NullEquality::NullEqualsNothing, + NullEquality::NullEqualsNull => protobuf::NullEquality::NullEqualsNull, + } + } +} + +/// Encode any slice of file-like values as a [`protobuf::FileGroup`]. +/// +/// `datafusion-datasource` cannot host this impl: `&T` is `#[fundamental]` but +/// `[T]` is not, so `&[PartitionedFile]` counts as foreign there and the orphan +/// rule rejects it. Here the *self* type is local, which is all the orphan rule +/// needs — and staying generic over the element means this crate never has to +/// name `PartitionedFile`, which lives above it in the dependency graph. +/// +/// The element bound is satisfied by +/// `impl TryFrom<&PartitionedFile> for protobuf::PartitionedFile` in +/// `datafusion-datasource`, so `protobuf::FileGroup::try_from(&files[..])` +/// resolves for callers exactly as it did before the proto types were split out. +impl TryFrom<&[T]> for protobuf::FileGroup +where + for<'a> &'a T: TryInto, +{ + type Error = DataFusionError; + + fn try_from(files: &[T]) -> Result { + Ok(protobuf::FileGroup { + files: files + .iter() + .map(TryInto::try_into) + .collect::, _>>()?, + }) + } +} + +#[cfg(test)] +mod tests { + use datafusion_common::{NullHandling, RecursionUnnestOption}; + + use super::*; + + #[test] + fn table_reference_roundtrip() { + for reference in [ + TableReference::bare("t"), + TableReference::partial("s", "t"), + TableReference::full("c", "s", "t"), + ] { + let encoded = protobuf::TableReference::from(reference.clone()); + let decoded = TableReference::try_from(encoded).unwrap(); + assert_eq!(decoded, reference); + } + } + + #[test] + fn table_reference_from_proto_rejects_missing_oneof() { + let proto = protobuf::TableReference { + table_reference_enum: None, + }; + let err = TableReference::try_from(proto).unwrap_err(); + assert!( + err.to_string().contains("table_reference_enum"), + "unexpected error: {err}" + ); + } + + #[test] + fn join_enums_roundtrip() { + for join_type in [ + JoinType::Inner, + JoinType::Left, + JoinType::Right, + JoinType::Full, + JoinType::LeftSemi, + JoinType::RightSemi, + JoinType::LeftAnti, + JoinType::RightAnti, + JoinType::LeftMark, + JoinType::RightMark, + ] { + assert_eq!( + JoinType::from(protobuf::JoinType::from(join_type)), + join_type + ); + } + for constraint in [JoinConstraint::On, JoinConstraint::Using] { + assert_eq!( + JoinConstraint::from(protobuf::JoinConstraint::from(constraint)), + constraint + ); + } + for null_equality in [ + NullEquality::NullEqualsNothing, + NullEquality::NullEqualsNull, + ] { + assert_eq!( + NullEquality::from(protobuf::NullEquality::from(null_equality)), + null_equality + ); + } + } + + #[test] + fn unnest_options_roundtrip() { + let options = UnnestOptions { + null_handling: NullHandling::Drop, + recursions: vec![RecursionUnnestOption { + input_column: "a".into(), + output_column: "b".into(), + depth: 2, + }], + }; + + let encoded = protobuf::UnnestOptions::from(&options); + let decoded = UnnestOptions::from(&encoded); + + assert_eq!(decoded.null_handling, options.null_handling); + assert_eq!(decoded.recursions, options.recursions); + } + + #[test] + fn stringified_plan_roundtrip() { + let plan = StringifiedPlan::new( + PlanType::OptimizedLogicalPlan { + optimizer_name: "push_down_filter".to_string(), + }, + "some plan", + ); + + let encoded = protobuf::StringifiedPlan::from(&plan); + let decoded = StringifiedPlan::from(&encoded); + + assert_eq!(decoded.plan_type, plan.plan_type); + assert_eq!(decoded.plan, plan.plan); + } +} diff --git a/datafusion/proto/src/logical_plan/file_formats.rs b/datafusion/proto/src/logical_plan/file_formats.rs index c63692d20bee6..bad014e874463 100644 --- a/datafusion/proto/src/logical_plan/file_formats.rs +++ b/datafusion/proto/src/logical_plan/file_formats.rs @@ -19,17 +19,9 @@ use std::sync::Arc; use super::LogicalExtensionCodec; use crate::convert::FromProto; -#[cfg(feature = "parquet")] -use crate::convert::TryFromProto; -use crate::protobuf::{ - CsvOptions as CsvOptionsProto, CsvQuoteStyle as CsvQuoteStyleProto, - JsonOptions as JsonOptionsProto, -}; +use crate::protobuf::{CsvOptions as CsvOptionsProto, JsonOptions as JsonOptionsProto}; use datafusion_common::config::{CsvOptions, JsonOptions}; -use datafusion_common::{ - TableReference, exec_datafusion_err, exec_err, not_impl_err, - parsers::{CompressionTypeVariant, CsvQuoteStyle}, -}; +use datafusion_common::{TableReference, exec_datafusion_err, exec_err, not_impl_err}; use datafusion_datasource::file_format::FileFormatFactory; use datafusion_datasource_arrow::file_format::ArrowFormatFactory; use datafusion_datasource_csv::file_format::CsvFormatFactory; @@ -82,111 +74,6 @@ impl FromProto<&CsvFormatFactory> for CsvOptionsProto { } } -impl FromProto<&CsvOptionsProto> for CsvOptions { - fn from_proto(proto: &CsvOptionsProto) -> Self { - CsvOptions { - has_header: if !proto.has_header.is_empty() { - Some(proto.has_header[0] != 0) - } else { - None - }, - delimiter: proto.delimiter.first().copied().unwrap_or(b','), - quote: proto.quote.first().copied().unwrap_or(b'"'), - terminator: if !proto.terminator.is_empty() { - Some(proto.terminator[0]) - } else { - None - }, - escape: if !proto.escape.is_empty() { - Some(proto.escape[0]) - } else { - None - }, - double_quote: if !proto.double_quote.is_empty() { - Some(proto.double_quote[0] != 0) - } else { - None - }, - compression: match proto.compression { - 0 => CompressionTypeVariant::GZIP, - 1 => CompressionTypeVariant::BZIP2, - 2 => CompressionTypeVariant::XZ, - 3 => CompressionTypeVariant::ZSTD, - _ => CompressionTypeVariant::UNCOMPRESSED, - }, - schema_infer_max_rec: proto.schema_infer_max_rec.map(|v| v as usize), - date_format: if proto.date_format.is_empty() { - None - } else { - Some(proto.date_format.clone()) - }, - datetime_format: if proto.datetime_format.is_empty() { - None - } else { - Some(proto.datetime_format.clone()) - }, - timestamp_format: if proto.timestamp_format.is_empty() { - None - } else { - Some(proto.timestamp_format.clone()) - }, - timestamp_tz_format: if proto.timestamp_tz_format.is_empty() { - None - } else { - Some(proto.timestamp_tz_format.clone()) - }, - time_format: if proto.time_format.is_empty() { - None - } else { - Some(proto.time_format.clone()) - }, - null_value: if proto.null_value.is_empty() { - None - } else { - Some(proto.null_value.clone()) - }, - null_regex: if proto.null_regex.is_empty() { - None - } else { - Some(proto.null_regex.clone()) - }, - comment: if !proto.comment.is_empty() { - Some(proto.comment[0]) - } else { - None - }, - newlines_in_values: if proto.newlines_in_values.is_empty() { - None - } else { - Some(proto.newlines_in_values[0] != 0) - }, - truncated_rows: if proto.truncated_rows.is_empty() { - None - } else { - Some(proto.truncated_rows[0] != 0) - }, - compression_level: proto.compression_level, - quote_style: match CsvQuoteStyleProto::try_from(proto.quote_style) { - Ok(CsvQuoteStyleProto::Always) => CsvQuoteStyle::Always, - Ok(CsvQuoteStyleProto::NonNumeric) => CsvQuoteStyle::NonNumeric, - Ok(CsvQuoteStyleProto::Never) => CsvQuoteStyle::Never, - Ok(CsvQuoteStyleProto::Necessary) => CsvQuoteStyle::Necessary, - _ => CsvQuoteStyle::Necessary, - }, - ignore_leading_whitespace: if proto.ignore_leading_whitespace.is_empty() { - None - } else { - Some(proto.ignore_leading_whitespace[0] != 0) - }, - ignore_trailing_whitespace: if proto.ignore_trailing_whitespace.is_empty() { - None - } else { - Some(proto.ignore_trailing_whitespace[0] != 0) - }, - } - } -} - // TODO! This is a placeholder for now and needs to be implemented for real. impl LogicalExtensionCodec for CsvLogicalExtensionCodec { fn try_decode( @@ -233,7 +120,7 @@ impl LogicalExtensionCodec for CsvLogicalExtensionCodec { let proto = CsvOptionsProto::decode(buf).map_err(|e| { exec_datafusion_err!("Failed to decode CsvOptionsProto: {e:?}") })?; - let options = CsvOptions::from_proto(&proto); + let options = CsvOptions::from(&proto); Ok(Arc::new(CsvFormatFactory { options: Some(options), })) @@ -277,23 +164,6 @@ impl FromProto<&JsonFormatFactory> for JsonOptionsProto { } } -impl FromProto<&JsonOptionsProto> for JsonOptions { - fn from_proto(proto: &JsonOptionsProto) -> Self { - JsonOptions { - compression: match proto.compression { - 0 => CompressionTypeVariant::GZIP, - 1 => CompressionTypeVariant::BZIP2, - 2 => CompressionTypeVariant::XZ, - 3 => CompressionTypeVariant::ZSTD, - _ => CompressionTypeVariant::UNCOMPRESSED, - }, - schema_infer_max_rec: proto.schema_infer_max_rec.map(|v| v as usize), - compression_level: proto.compression_level, - newline_delimited: proto.newline_delimited.unwrap_or(true), - } - } -} - #[derive(Debug)] pub struct JsonLogicalExtensionCodec; @@ -343,7 +213,7 @@ impl LogicalExtensionCodec for JsonLogicalExtensionCodec { let proto = JsonOptionsProto::decode(buf).map_err(|e| { exec_datafusion_err!("Failed to decode JsonOptionsProto: {e:?}") })?; - let options = JsonOptions::from_proto(&proto); + let options = JsonOptions::from(&proto); Ok(Arc::new(JsonFormatFactory { options: Some(options), })) @@ -384,10 +254,7 @@ mod parquet { TableParquetOptions as TableParquetOptionsProto, parquet_column_options, parquet_options, }; - use datafusion_common::config::{ - MaxRowGroupBytes, ParquetCdcOptions, ParquetColumnOptions, ParquetOptions, - TableParquetOptions, - }; + use datafusion_common::config::TableParquetOptions; use datafusion_datasource_parquet::file_format::ParquetFormatFactory; impl FromProto<&ParquetFormatFactory> for TableParquetOptionsProto { @@ -507,217 +374,6 @@ mod parquet { } } - impl FromProto for ParquetCdcOptions { - fn from_proto(value: ParquetCdcOptionsProto) -> Self { - ParquetCdcOptions { - enabled: value.enabled, - min_chunk_size: value.min_chunk_size as usize, - max_chunk_size: value.max_chunk_size as usize, - norm_level: value.norm_level, - } - } - } - - impl TryFromProto<&ParquetOptionsProto> for ParquetOptions { - type Error = datafusion_common::DataFusionError; - - fn try_from_proto( - proto: &ParquetOptionsProto, - ) -> datafusion_common::Result { - let writer_version = match proto.writer_version.as_str() { - // Proto3 decodes an omitted string field as the empty string. The - // schema documents writer_version's logical default as "1.0", so - // preserve that default when the field is absent on the wire. - "" => ParquetOptions::default().writer_version, - version => version.parse()?, - }; - - Ok(ParquetOptions { - enable_page_index: proto.enable_page_index, - pruning: proto.pruning, - skip_metadata: proto.skip_metadata, - metadata_size_hint: proto - .metadata_size_hint_opt - .as_ref() - .map(|opt| match opt { - parquet_options::MetadataSizeHintOpt::MetadataSizeHint(size) => { - *size as usize - } - }), - pushdown_filters: proto.pushdown_filters, - reorder_filters: proto.reorder_filters, - force_filter_selections: proto.force_filter_selections, - data_pagesize_limit: proto.data_pagesize_limit as usize, - write_batch_size: proto.write_batch_size as usize, - writer_version, - compression: proto.compression_opt.as_ref().map(|opt| match opt { - parquet_options::CompressionOpt::Compression(compression) => { - compression.clone() - } - }), - dictionary_enabled: proto.dictionary_enabled_opt.as_ref().map(|opt| { - match opt { - parquet_options::DictionaryEnabledOpt::DictionaryEnabled( - enabled, - ) => *enabled, - } - }), - dictionary_page_size_limit: proto.dictionary_page_size_limit as usize, - statistics_enabled: proto.statistics_enabled_opt.as_ref().map( - |opt| match opt { - parquet_options::StatisticsEnabledOpt::StatisticsEnabled( - statistics, - ) => statistics.clone(), - }, - ), - max_row_group_size: proto.max_row_group_size as usize, - max_in_list_size: proto.max_in_list_size as usize, - created_by: proto.created_by.clone(), - column_index_truncate_length: proto - .column_index_truncate_length_opt - .as_ref() - .map(|opt| match opt { - parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(length) => *length as usize, - }), - statistics_truncate_length: proto - .statistics_truncate_length_opt - .as_ref() - .map(|opt| match opt { - parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(length) => *length as usize, - }), - data_page_row_count_limit: proto.data_page_row_count_limit as usize, - encoding: proto.encoding_opt.as_ref().map(|opt| match opt { - parquet_options::EncodingOpt::Encoding(encoding) => { - encoding.clone() - } - }), - bloom_filter_on_read: proto.bloom_filter_on_read, - bloom_filter_on_write: proto.bloom_filter_on_write, - bloom_filter_fpp: proto - .bloom_filter_fpp_opt - .as_ref() - .map(|opt| match opt { - parquet_options::BloomFilterFppOpt::BloomFilterFpp(fpp) => *fpp, - }), - bloom_filter_ndv: proto - .bloom_filter_ndv_opt - .as_ref() - .map(|opt| match opt { - parquet_options::BloomFilterNdvOpt::BloomFilterNdv(ndv) => *ndv, - }), - allow_single_file_parallelism: proto.allow_single_file_parallelism, - maximum_parallel_row_group_writers: proto - .maximum_parallel_row_group_writers - as usize, - maximum_buffered_record_batches_per_stream: proto - .maximum_buffered_record_batches_per_stream - as usize, - schema_force_view_types: proto.schema_force_view_types, - binary_as_string: proto.binary_as_string, - skip_arrow_metadata: proto.skip_arrow_metadata, - coerce_int96: proto.coerce_int96_opt.as_ref().map(|opt| match opt { - parquet_options::CoerceInt96Opt::CoerceInt96(coerce_int96) => { - coerce_int96.clone() - } - }), - coerce_int96_tz: proto - .coerce_int96_tz_opt - .as_ref() - .map(|opt| match opt { - parquet_options::CoerceInt96TzOpt::CoerceInt96Tz(tz) => { - tz.clone() - } - }), - max_predicate_cache_size: proto - .max_predicate_cache_size_opt - .as_ref() - .map(|opt| match opt { - parquet_options::MaxPredicateCacheSizeOpt::MaxPredicateCacheSize( - size, - ) => *size as usize, - }), - max_row_group_bytes: proto - .max_row_group_bytes_opt - .as_ref() - .and_then(|opt| match opt { - parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(size) => { - MaxRowGroupBytes::try_new(*size as usize).ok() - } - }), - content_defined_chunking: proto - .content_defined_chunking - .map(ParquetCdcOptions::from_proto) - .unwrap_or_default(), - }) - } - } - - impl FromProto for ParquetColumnOptions { - fn from_proto(proto: ParquetColumnOptionsProto) -> Self { - ParquetColumnOptions { - bloom_filter_enabled: proto.bloom_filter_enabled_opt.map( - |parquet_column_options::BloomFilterEnabledOpt::BloomFilterEnabled(v)| v, - ), - encoding: proto - .encoding_opt - .map(|parquet_column_options::EncodingOpt::Encoding(v)| v), - dictionary_enabled: proto.dictionary_enabled_opt.map( - |parquet_column_options::DictionaryEnabledOpt::DictionaryEnabled(v)| v, - ), - compression: proto - .compression_opt - .map(|parquet_column_options::CompressionOpt::Compression(v)| v), - statistics_enabled: proto.statistics_enabled_opt.map( - |parquet_column_options::StatisticsEnabledOpt::StatisticsEnabled(v)| v, - ), - bloom_filter_fpp: proto - .bloom_filter_fpp_opt - .map(|parquet_column_options::BloomFilterFppOpt::BloomFilterFpp(v)| v), - bloom_filter_ndv: proto - .bloom_filter_ndv_opt - .map(|parquet_column_options::BloomFilterNdvOpt::BloomFilterNdv(v)| v), - } - } - } - - impl TryFromProto<&TableParquetOptionsProto> for TableParquetOptions { - type Error = datafusion_common::DataFusionError; - - fn try_from_proto( - proto: &TableParquetOptionsProto, - ) -> datafusion_common::Result { - Ok(TableParquetOptions { - global: proto - .global - .as_ref() - .map(ParquetOptions::try_from_proto) - .transpose()? - .unwrap_or_default(), - column_specific_options: proto - .column_specific_options - .iter() - .map(|parquet_column_options| { - ( - parquet_column_options.column_name.clone(), - ParquetColumnOptions::from_proto( - parquet_column_options - .options - .clone() - .unwrap_or_default(), - ), - ) - }) - .collect(), - key_value_metadata: proto - .key_value_metadata - .iter() - .map(|(k, v)| (k.clone(), Some(v.clone()))) - .collect(), - ..Default::default() - }) - } - } - #[derive(Debug)] pub struct ParquetLogicalExtensionCodec; @@ -768,7 +424,7 @@ mod parquet { let proto = TableParquetOptionsProto::decode(buf).map_err(|e| { exec_datafusion_err!("Failed to decode TableParquetOptionsProto: {e:?}") })?; - let options = TableParquetOptions::try_from_proto(&proto)?; + let options = TableParquetOptions::try_from(&proto)?; Ok(Arc::new(ParquetFormatFactory { options: Some(options), })) @@ -804,6 +460,7 @@ mod parquet { #[cfg(test)] mod tests { use super::*; + use datafusion_common::config::ParquetOptions; fn encode_table_options(proto: TableParquetOptionsProto) -> Vec { let mut buf = Vec::new(); diff --git a/datafusion/proto/src/logical_plan/from_proto.rs b/datafusion/proto/src/logical_plan/from_proto.rs index 9c91799ae9675..d4d0ea7292ffe 100644 --- a/datafusion/proto/src/logical_plan/from_proto.rs +++ b/datafusion/proto/src/logical_plan/from_proto.rs @@ -20,8 +20,8 @@ use std::sync::Arc; use arrow::datatypes::{DataType, Field}; use datafusion_common::datatype::DataTypeExt; use datafusion_common::{ - NullEquality, RecursionUnnestOption, Result, ScalarValue, SplitPoint, TableReference, - UnnestOptions, exec_datafusion_err, internal_err, plan_datafusion_err, + Result, ScalarValue, SplitPoint, TableReference, exec_datafusion_err, internal_err, + plan_datafusion_err, }; use datafusion_execution::TaskContext; use datafusion_execution::registry::FunctionRegistry; @@ -36,164 +36,16 @@ use datafusion_expr::logical_plan::Subquery; use datafusion_expr::{ Between, BinaryExpr, Case, Cast, Expr, GroupingSet, GroupingSet::GroupingSets, - JoinConstraint, JoinType, Like, Operator, TryCast, WindowFrame, + Like, Operator, TryCast, WindowFrame, expr::{self, InList, WindowFunction}, - logical_plan::{PlanType, StringifiedPlan}, }; use datafusion_expr::{ExprFunctionExt, WriteOp}; use datafusion_proto_common::{FromProtoError as Error, from_proto::FromOptionalField}; -use crate::protobuf::plan_type::PlanTypeEnum::{ - FinalPhysicalPlanWithSchema, InitialPhysicalPlanWithSchema, -}; -use crate::protobuf::{ - self, AnalyzedLogicalPlanType, CubeNode, GroupingSetNode, OptimizedLogicalPlanType, - OptimizedPhysicalPlanType, PlaceholderNode, RollupNode, - plan_type::PlanTypeEnum::{ - AnalyzedLogicalPlan, FinalAnalyzedLogicalPlan, FinalLogicalPlan, - FinalPhysicalPlan, FinalPhysicalPlanWithStats, InitialLogicalPlan, - InitialPhysicalPlan, InitialPhysicalPlanWithStats, OptimizedLogicalPlan, - OptimizedPhysicalPlan, PhysicalPlanError, - }, -}; - -use crate::convert::{FromProto, TryFromProto}; +use crate::protobuf::{self, CubeNode, GroupingSetNode, PlaceholderNode, RollupNode}; use super::{AsLogicalPlan, LogicalExtensionCodec}; -impl FromProto<&protobuf::UnnestOptions> for UnnestOptions { - fn from_proto(opts: &protobuf::UnnestOptions) -> Self { - use datafusion_common::NullHandling; - use protobuf::unnest_options::NullHandling as ProtoNullHandling; - let null_handling = match ProtoNullHandling::try_from(opts.null_handling) { - Ok(ProtoNullHandling::Preserve) => NullHandling::Preserve, - Ok(ProtoNullHandling::Drop) => NullHandling::Drop, - Ok(ProtoNullHandling::PreserveAndExpandEmpty) => { - NullHandling::PreserveAndExpandEmpty - } - // Unknown enum values fall back to the default (Preserve), which - // matches DataFusion's historical behavior. - Err(_) => NullHandling::Preserve, - }; - Self { - null_handling, - recursions: opts - .recursions - .iter() - .map(|r| RecursionUnnestOption { - input_column: r.input_column.as_ref().unwrap().into(), - output_column: r.output_column.as_ref().unwrap().into(), - depth: r.depth as usize, - }) - .collect::>(), - } - } -} - -impl TryFromProto for TableReference { - type Error = Error; - - fn try_from_proto(value: protobuf::TableReference) -> Result { - use protobuf::table_reference::TableReferenceEnum; - let table_reference_enum = value - .table_reference_enum - .ok_or_else(|| Error::required("table_reference_enum"))?; - - match table_reference_enum { - TableReferenceEnum::Bare(protobuf::BareTableReference { table }) => { - Ok(TableReference::bare(table)) - } - TableReferenceEnum::Partial(protobuf::PartialTableReference { - schema, - table, - }) => Ok(TableReference::partial(schema, table)), - TableReferenceEnum::Full(protobuf::FullTableReference { - catalog, - schema, - table, - }) => Ok(TableReference::full(catalog, schema, table)), - } - } -} - -impl FromProto<&protobuf::StringifiedPlan> for StringifiedPlan { - fn from_proto(stringified_plan: &protobuf::StringifiedPlan) -> Self { - Self { - plan_type: match stringified_plan - .plan_type - .as_ref() - .and_then(|pt| pt.plan_type_enum.as_ref()) - .unwrap_or_else(|| { - panic!( - "Cannot create protobuf::StringifiedPlan from {stringified_plan:?}" - ) - }) { - InitialLogicalPlan(_) => PlanType::InitialLogicalPlan, - AnalyzedLogicalPlan(AnalyzedLogicalPlanType { analyzer_name }) => { - PlanType::AnalyzedLogicalPlan { - analyzer_name:analyzer_name.clone() - } - } - FinalAnalyzedLogicalPlan(_) => PlanType::FinalAnalyzedLogicalPlan, - OptimizedLogicalPlan(OptimizedLogicalPlanType { optimizer_name }) => { - PlanType::OptimizedLogicalPlan { - optimizer_name: optimizer_name.clone(), - } - } - FinalLogicalPlan(_) => PlanType::FinalLogicalPlan, - InitialPhysicalPlan(_) => PlanType::InitialPhysicalPlan, - InitialPhysicalPlanWithStats(_) => PlanType::InitialPhysicalPlanWithStats, - InitialPhysicalPlanWithSchema(_) => PlanType::InitialPhysicalPlanWithSchema, - OptimizedPhysicalPlan(OptimizedPhysicalPlanType { optimizer_name }) => { - PlanType::OptimizedPhysicalPlan { - optimizer_name: optimizer_name.clone(), - } - } - FinalPhysicalPlan(_) => PlanType::FinalPhysicalPlan, - FinalPhysicalPlanWithStats(_) => PlanType::FinalPhysicalPlanWithStats, - FinalPhysicalPlanWithSchema(_) => PlanType::FinalPhysicalPlanWithSchema, - PhysicalPlanError(_) => PlanType::PhysicalPlanError, - }, - plan: Arc::new(stringified_plan.plan.clone()), - } - } -} - -impl FromProto for JoinType { - fn from_proto(t: protobuf::JoinType) -> Self { - match t { - protobuf::JoinType::Inner => JoinType::Inner, - protobuf::JoinType::Left => JoinType::Left, - protobuf::JoinType::Right => JoinType::Right, - protobuf::JoinType::Full => JoinType::Full, - protobuf::JoinType::Leftsemi => JoinType::LeftSemi, - protobuf::JoinType::Rightsemi => JoinType::RightSemi, - protobuf::JoinType::Leftanti => JoinType::LeftAnti, - protobuf::JoinType::Rightanti => JoinType::RightAnti, - protobuf::JoinType::Leftmark => JoinType::LeftMark, - protobuf::JoinType::Rightmark => JoinType::RightMark, - } - } -} - -impl FromProto for JoinConstraint { - fn from_proto(t: protobuf::JoinConstraint) -> Self { - match t { - protobuf::JoinConstraint::On => JoinConstraint::On, - protobuf::JoinConstraint::Using => JoinConstraint::Using, - } - } -} - -impl FromProto for NullEquality { - fn from_proto(t: protobuf::NullEquality) -> Self { - match t { - protobuf::NullEquality::NullEqualsNothing => NullEquality::NullEqualsNothing, - protobuf::NullEquality::NullEqualsNull => NullEquality::NullEqualsNull, - } - } -} - /// Reconstruct a [`WriteOp`] from a [`protobuf::DmlNode`], reading the /// `merge_into` payload when the type tag is `MergeInto`. pub fn parse_write_op( @@ -417,7 +269,7 @@ pub fn parse_expr( alias .relation .first() - .map(|r| TableReference::try_from_proto(r.clone())) + .map(|r| TableReference::try_from(r.clone())) .transpose()?, alias.alias.clone(), ))), @@ -622,7 +474,7 @@ pub fn parse_expr( ExprType::Wildcard(protobuf::Wildcard { qualifier }) => { let qualifier = qualifier .to_owned() - .map(TableReference::try_from_proto) + .map(TableReference::try_from) .transpose()?; #[expect(deprecated)] Ok(Expr::Wildcard { diff --git a/datafusion/proto/src/logical_plan/mod.rs b/datafusion/proto/src/logical_plan/mod.rs index 653ae9ab05355..42c368090a40c 100644 --- a/datafusion/proto/src/logical_plan/mod.rs +++ b/datafusion/proto/src/logical_plan/mod.rs @@ -19,7 +19,6 @@ use std::collections::HashMap; use std::fmt::Debug; use std::sync::Arc; -use crate::convert::{FromProto, TryFromProto}; use crate::protobuf::logical_plan_node::LogicalPlanType::CustomScan; use crate::protobuf::{ ColumnUnnestListItem, ColumnUnnestListRecursion, CteWorkTableScanNode, @@ -348,7 +347,7 @@ fn from_table_reference( ) })?; - Ok(TableReference::try_from_proto(table_ref.clone())?) + Ok(TableReference::try_from(table_ref.clone())?) } /// Converts [LogicalPlan::TableScan] to [TableSource] @@ -1027,9 +1026,9 @@ impl AsLogicalPlan for LogicalPlanNode { Arc::new(right), on, filter, - datafusion_expr::JoinType::from_proto(join_type), - JoinConstraint::from_proto(join_constraint), - NullEquality::from_proto(null_equality), + datafusion_expr::JoinType::from(join_type), + JoinConstraint::from(join_constraint), + NullEquality::from(null_equality), join.null_aware, )?)) } @@ -1195,7 +1194,7 @@ impl AsLogicalPlan for LogicalPlanNode { unnest .options .as_ref() - .map(datafusion_common::UnnestOptions::from_proto) + .map(datafusion_common::UnnestOptions::from) .ok_or_else(|| { proto_error("Missing required field in protobuf") })?, @@ -1430,7 +1429,7 @@ impl AsLogicalPlan for LogicalPlanNode { logical_plan_type: Some(LogicalPlanType::ListingScan( protobuf::ListingTableScanNode { file_format_type: Some(file_format_type), - table_name: Some(protobuf::TableReference::from_proto( + table_name: Some(protobuf::TableReference::from( table_name.clone(), )), file_extension: options.file_extension.clone(), @@ -1452,7 +1451,7 @@ impl AsLogicalPlan for LogicalPlanNode { Ok(LogicalPlanNode { logical_plan_type: Some(LogicalPlanType::ViewScan(Box::new( protobuf::ViewTableScanNode { - table_name: Some(protobuf::TableReference::from_proto( + table_name: Some(protobuf::TableReference::from( table_name.clone(), )), input: Some(Box::new( @@ -1491,7 +1490,7 @@ impl AsLogicalPlan for LogicalPlanNode { Ok(LogicalPlanNode { logical_plan_type: Some(LogicalPlanType::EmptyTableScan( protobuf::EmptyTableScanNode { - table_name: Some(protobuf::TableReference::from_proto( + table_name: Some(protobuf::TableReference::from( table_name.clone(), )), schema: Some(schema), @@ -1507,7 +1506,7 @@ impl AsLogicalPlan for LogicalPlanNode { .try_encode_table_provider(table_name, provider, &mut bytes) .map_err(|e| context!("Error serializing custom table", e))?; let scan = CustomScan(CustomTableScanNode { - table_name: Some(protobuf::TableReference::from_proto( + table_name: Some(protobuf::TableReference::from( table_name.clone(), )), projection, @@ -1660,11 +1659,11 @@ impl AsLogicalPlan for LogicalPlanNode { .collect::, ToProtoError>>()? .into_iter() .unzip(); - let join_type = protobuf::JoinType::from_proto(join_type.to_owned()); + let join_type = protobuf::JoinType::from(join_type.to_owned()); let join_constraint = - protobuf::JoinConstraint::from_proto(join_constraint.to_owned()); + protobuf::JoinConstraint::from(join_constraint.to_owned()); let null_equality = - protobuf::NullEquality::from_proto(null_equality.to_owned()); + protobuf::NullEquality::from(null_equality.to_owned()); let filter = filter .as_ref() .map(|e| serialize_expr(e, extension_codec).map(Box::new)) @@ -1703,9 +1702,7 @@ impl AsLogicalPlan for LogicalPlanNode { logical_plan_type: Some(LogicalPlanType::SubqueryAlias(Box::new( protobuf::SubqueryAliasNode { input: Some(Box::new(input)), - alias: Some(protobuf::TableReference::from_proto( - (*alias).clone(), - )), + alias: Some(protobuf::TableReference::from((*alias).clone())), }, ))), }) @@ -1852,9 +1849,7 @@ impl AsLogicalPlan for LogicalPlanNode { Ok(LogicalPlanNode { logical_plan_type: Some(LogicalPlanType::CreateExternalTable( protobuf::CreateExternalTableNode { - name: Some(protobuf::TableReference::from_proto( - name.clone(), - )), + name: Some(protobuf::TableReference::from(name.clone())), location: legacy_location, locations: proto_locations, file_type: file_type.clone(), @@ -1882,7 +1877,7 @@ impl AsLogicalPlan for LogicalPlanNode { })) => Ok(LogicalPlanNode { logical_plan_type: Some(LogicalPlanType::CreateView(Box::new( protobuf::CreateViewNode { - name: Some(protobuf::TableReference::from_proto(name.clone())), + name: Some(protobuf::TableReference::from(name.clone())), input: Some(Box::new(LogicalPlanNode::try_from_logical_plan( input, extension_codec, @@ -2072,7 +2067,7 @@ impl AsLogicalPlan for LogicalPlanNode { .map(|c| *c as u64) .collect(), schema: Some(schema.try_into()?), - options: Some(protobuf::UnnestOptions::from_proto(options)), + options: Some(protobuf::UnnestOptions::from(options)), }, ))), }) @@ -2093,7 +2088,7 @@ impl AsLogicalPlan for LogicalPlanNode { })) => Ok(LogicalPlanNode { logical_plan_type: Some(LogicalPlanType::DropView( protobuf::DropViewNode { - name: Some(protobuf::TableReference::from_proto(name.clone())), + name: Some(protobuf::TableReference::from(name.clone())), if_exists: *if_exists, schema: Some(schema.try_into()?), }, @@ -2155,7 +2150,7 @@ impl AsLogicalPlan for LogicalPlanNode { Arc::clone(target), extension_codec, )?)), - table_name: Some(protobuf::TableReference::from_proto( + table_name: Some(protobuf::TableReference::from( table_name.clone(), )), dml_type: dml_type.into(), diff --git a/datafusion/proto/src/logical_plan/to_proto.rs b/datafusion/proto/src/logical_plan/to_proto.rs index 82fdfcc5ee291..16c3468465541 100644 --- a/datafusion/proto/src/logical_plan/to_proto.rs +++ b/datafusion/proto/src/logical_plan/to_proto.rs @@ -19,7 +19,7 @@ //! DataFusion logical plans to be serialized and transmitted between //! processes. -use datafusion_common::{NullEquality, SplitPoint, TableReference, UnnestOptions}; +use datafusion_common::SplitPoint; use datafusion_expr::dml::{MergeIntoAction, MergeIntoClause, MergeIntoOp}; use datafusion_expr::expr::{ self, AggregateFunctionParams, Alias, Between, BinaryExpr, Cast, GroupingSet, @@ -27,116 +27,16 @@ use datafusion_expr::expr::{ ScalarFunction, Unnest, }; use datafusion_expr::logical_plan::Subquery; -use datafusion_expr::{ - Expr, JoinConstraint, JoinType, SortExpr, TryCast, WindowFunctionDefinition, - logical_plan::PlanType, logical_plan::StringifiedPlan, -}; +use datafusion_expr::{Expr, SortExpr, TryCast, WindowFunctionDefinition}; -use crate::protobuf::RecursionUnnestOption; use crate::protobuf::{ - self, AnalyzedLogicalPlanType, CubeNode, EmptyMessage, GroupingSetNode, - LogicalExprList, OptimizedLogicalPlanType, OptimizedPhysicalPlanType, - PlaceholderNode, RollupNode, ToProtoError as Error, - plan_type::PlanTypeEnum::{ - AnalyzedLogicalPlan, FinalAnalyzedLogicalPlan, FinalLogicalPlan, - FinalPhysicalPlan, FinalPhysicalPlanWithSchema, FinalPhysicalPlanWithStats, - InitialLogicalPlan, InitialPhysicalPlan, InitialPhysicalPlanWithSchema, - InitialPhysicalPlanWithStats, OptimizedLogicalPlan, OptimizedPhysicalPlan, - PhysicalPlanError, - }, + self, CubeNode, GroupingSetNode, LogicalExprList, PlaceholderNode, RollupNode, + ToProtoError as Error, }; use super::{AsLogicalPlan, LogicalExtensionCodec}; -use crate::convert::FromProto; use crate::protobuf::LogicalPlanNode; -impl FromProto<&UnnestOptions> for protobuf::UnnestOptions { - fn from_proto(opts: &UnnestOptions) -> Self { - use datafusion_common::NullHandling; - use protobuf::unnest_options::NullHandling as ProtoNullHandling; - let null_handling = match opts.null_handling { - NullHandling::Preserve => ProtoNullHandling::Preserve, - NullHandling::Drop => ProtoNullHandling::Drop, - NullHandling::PreserveAndExpandEmpty => { - ProtoNullHandling::PreserveAndExpandEmpty - } - } as i32; - Self { - null_handling, - recursions: opts - .recursions - .iter() - .map(|r| RecursionUnnestOption { - input_column: Some((&r.input_column).into()), - output_column: Some((&r.output_column).into()), - depth: r.depth as u32, - }) - .collect(), - } - } -} - -impl FromProto<&StringifiedPlan> for protobuf::StringifiedPlan { - fn from_proto(stringified_plan: &StringifiedPlan) -> Self { - Self { - plan_type: match stringified_plan.clone().plan_type { - PlanType::InitialLogicalPlan => Some(protobuf::PlanType { - plan_type_enum: Some(InitialLogicalPlan(EmptyMessage {})), - }), - PlanType::AnalyzedLogicalPlan { analyzer_name } => { - Some(protobuf::PlanType { - plan_type_enum: Some(AnalyzedLogicalPlan( - AnalyzedLogicalPlanType { analyzer_name }, - )), - }) - } - PlanType::FinalAnalyzedLogicalPlan => Some(protobuf::PlanType { - plan_type_enum: Some(FinalAnalyzedLogicalPlan(EmptyMessage {})), - }), - PlanType::OptimizedLogicalPlan { optimizer_name } => { - Some(protobuf::PlanType { - plan_type_enum: Some(OptimizedLogicalPlan( - OptimizedLogicalPlanType { optimizer_name }, - )), - }) - } - PlanType::FinalLogicalPlan => Some(protobuf::PlanType { - plan_type_enum: Some(FinalLogicalPlan(EmptyMessage {})), - }), - PlanType::InitialPhysicalPlan => Some(protobuf::PlanType { - plan_type_enum: Some(InitialPhysicalPlan(EmptyMessage {})), - }), - PlanType::OptimizedPhysicalPlan { optimizer_name } => { - Some(protobuf::PlanType { - plan_type_enum: Some(OptimizedPhysicalPlan( - OptimizedPhysicalPlanType { optimizer_name }, - )), - }) - } - PlanType::FinalPhysicalPlan => Some(protobuf::PlanType { - plan_type_enum: Some(FinalPhysicalPlan(EmptyMessage {})), - }), - PlanType::InitialPhysicalPlanWithStats => Some(protobuf::PlanType { - plan_type_enum: Some(InitialPhysicalPlanWithStats(EmptyMessage {})), - }), - PlanType::InitialPhysicalPlanWithSchema => Some(protobuf::PlanType { - plan_type_enum: Some(InitialPhysicalPlanWithSchema(EmptyMessage {})), - }), - PlanType::FinalPhysicalPlanWithStats => Some(protobuf::PlanType { - plan_type_enum: Some(FinalPhysicalPlanWithStats(EmptyMessage {})), - }), - PlanType::FinalPhysicalPlanWithSchema => Some(protobuf::PlanType { - plan_type_enum: Some(FinalPhysicalPlanWithSchema(EmptyMessage {})), - }), - PlanType::PhysicalPlanError => Some(protobuf::PlanType { - plan_type_enum: Some(PhysicalPlanError(EmptyMessage {})), - }), - }, - plan: stringified_plan.plan.to_string(), - } - } -} - pub fn serialize_exprs<'a, I>( exprs: I, codec: &dyn LogicalExtensionCodec, @@ -170,7 +70,7 @@ pub fn serialize_expr( expr: Some(Box::new(serialize_expr(expr.as_ref(), codec)?)), relation: relation .to_owned() - .map(|r| vec![protobuf::TableReference::from_proto(r)]) + .map(|r| vec![protobuf::TableReference::from(r)]) .unwrap_or(vec![]), alias: name.to_owned(), metadata: metadata @@ -552,9 +452,7 @@ pub fn serialize_expr( #[expect(deprecated)] Expr::Wildcard { qualifier, .. } => protobuf::LogicalExprNode { expr_type: Some(ExprType::Wildcard(protobuf::Wildcard { - qualifier: qualifier - .to_owned() - .map(protobuf::TableReference::from_proto), + qualifier: qualifier.to_owned().map(protobuf::TableReference::from), })), }, Expr::ScalarSubquery(subquery) => protobuf::LogicalExprNode { @@ -680,73 +578,6 @@ pub(super) fn serialize_range_split_point( }) } -impl FromProto for protobuf::TableReference { - fn from_proto(t: TableReference) -> Self { - use protobuf::table_reference::TableReferenceEnum; - let table_reference_enum = match t { - TableReference::Bare { table } => { - TableReferenceEnum::Bare(protobuf::BareTableReference { - table: table.to_string(), - }) - } - TableReference::Partial { schema, table } => { - TableReferenceEnum::Partial(protobuf::PartialTableReference { - schema: schema.to_string(), - table: table.to_string(), - }) - } - TableReference::Full { - catalog, - schema, - table, - } => TableReferenceEnum::Full(protobuf::FullTableReference { - catalog: catalog.to_string(), - schema: schema.to_string(), - table: table.to_string(), - }), - }; - - protobuf::TableReference { - table_reference_enum: Some(table_reference_enum), - } - } -} - -impl FromProto for protobuf::JoinType { - fn from_proto(t: JoinType) -> Self { - match t { - JoinType::Inner => protobuf::JoinType::Inner, - JoinType::Left => protobuf::JoinType::Left, - JoinType::Right => protobuf::JoinType::Right, - JoinType::Full => protobuf::JoinType::Full, - JoinType::LeftSemi => protobuf::JoinType::Leftsemi, - JoinType::RightSemi => protobuf::JoinType::Rightsemi, - JoinType::LeftAnti => protobuf::JoinType::Leftanti, - JoinType::RightAnti => protobuf::JoinType::Rightanti, - JoinType::LeftMark => protobuf::JoinType::Leftmark, - JoinType::RightMark => protobuf::JoinType::Rightmark, - } - } -} - -impl FromProto for protobuf::JoinConstraint { - fn from_proto(t: JoinConstraint) -> Self { - match t { - JoinConstraint::On => protobuf::JoinConstraint::On, - JoinConstraint::Using => protobuf::JoinConstraint::Using, - } - } -} - -impl FromProto for protobuf::NullEquality { - fn from_proto(t: NullEquality) -> Self { - match t { - NullEquality::NullEqualsNothing => protobuf::NullEquality::NullEqualsNothing, - NullEquality::NullEqualsNull => protobuf::NullEquality::NullEqualsNull, - } - } -} - pub fn serialize_merge_into_op( op: &MergeIntoOp, codec: &dyn LogicalExtensionCodec, diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs index b32ed268c07d8..a450f7a7e888f 100644 --- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs @@ -116,7 +116,7 @@ use datafusion_proto::logical_plan::to_proto::serialize_expr; use datafusion_proto::logical_plan::{ DefaultLogicalExtensionCodec, LogicalExtensionCodec, from_proto, }; -use datafusion_proto::{FromProto, protobuf}; +use datafusion_proto::protobuf; use crate::cases::{ MyAggregateUDF, MyAggregateUdfNode, MyHigherOrderUDF, MyHigherOrderUdfNode, @@ -495,9 +495,7 @@ async fn roundtrip_create_external_table_legacy_location() -> Result<()> { let ctx = SessionContext::new(); let schema = DFSchema::empty(); let create_external_table = protobuf::CreateExternalTableNode { - name: Some(protobuf::TableReference::from_proto(TableReference::bare( - "t", - ))), + name: Some(protobuf::TableReference::from(TableReference::bare("t"))), location: "legacy.csv".to_string(), locations: vec![], file_type: "CSV".to_string(), From 736b50e9a300cefdf3a2684c1ce9b1057c625d62 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sun, 9 Aug 2026 10:06:43 -0400 Subject: [PATCH 4/7] refactor(proto): restore format-factory conversions as From in the format crates `CsvFormatFactory`, `JsonFormatFactory` and `ParquetFormatFactory` are owned by `datafusion-datasource-{csv,json,parquet}`, so their options-encoding conversions become plain `From` impls next to the types, behind each crate's existing `proto` feature. This is the last `FromProto` / `TryFromProto` impl in the tree: `convert.rs` and `convert_required_proto!` now have no implementors, clearing the way for the final cleanup in #24019. Part of #24019. Co-Authored-By: Claude Opus 5 --- datafusion/datasource-csv/src/file_format.rs | 48 +++++ datafusion/datasource-json/src/file_format.rs | 21 ++ .../datasource-parquet/src/file_format.rs | 126 ++++++++++++ .../proto/src/logical_plan/file_formats.rs | 190 +----------------- 4 files changed, 200 insertions(+), 185 deletions(-) diff --git a/datafusion/datasource-csv/src/file_format.rs b/datafusion/datasource-csv/src/file_format.rs index 7161519001643..c0d22b80f08d0 100644 --- a/datafusion/datasource-csv/src/file_format.rs +++ b/datafusion/datasource-csv/src/file_format.rs @@ -926,6 +926,54 @@ impl CsvSink { } } +/// Encode a [`CsvFormatFactory`]'s options as their protobuf form. +/// +/// The reverse direction is `From<&protobuf::CsvOptions> for CsvOptions` in +/// `datafusion-proto-models`: `CsvOptions` is a `datafusion-common` type, so +/// that half cannot live here. +#[cfg(feature = "proto")] +impl From<&CsvFormatFactory> for datafusion_proto_models::protobuf::CsvOptions { + fn from(factory: &CsvFormatFactory) -> Self { + if let Some(options) = &factory.options { + datafusion_proto_models::protobuf::CsvOptions { + has_header: options.has_header.map_or(vec![], |v| vec![v as u8]), + delimiter: vec![options.delimiter], + quote: vec![options.quote], + terminator: options.terminator.map_or(vec![], |v| vec![v]), + escape: options.escape.map_or(vec![], |v| vec![v]), + double_quote: options.double_quote.map_or(vec![], |v| vec![v as u8]), + compression: options.compression as i32, + schema_infer_max_rec: options.schema_infer_max_rec.map(|v| v as u64), + date_format: options.date_format.clone().unwrap_or_default(), + datetime_format: options.datetime_format.clone().unwrap_or_default(), + timestamp_format: options.timestamp_format.clone().unwrap_or_default(), + timestamp_tz_format: options + .timestamp_tz_format + .clone() + .unwrap_or_default(), + time_format: options.time_format.clone().unwrap_or_default(), + null_value: options.null_value.clone().unwrap_or_default(), + null_regex: options.null_regex.clone().unwrap_or_default(), + comment: options.comment.map_or(vec![], |v| vec![v]), + newlines_in_values: options + .newlines_in_values + .map_or(vec![], |v| vec![v as u8]), + truncated_rows: options.truncated_rows.map_or(vec![], |v| vec![v as u8]), + compression_level: options.compression_level, + quote_style: options.quote_style as i32, + ignore_leading_whitespace: options + .ignore_leading_whitespace + .map_or(vec![], |v| vec![v as u8]), + ignore_trailing_whitespace: options + .ignore_trailing_whitespace + .map_or(vec![], |v| vec![v as u8]), + } + } else { + datafusion_proto_models::protobuf::CsvOptions::default() + } + } +} + #[cfg(test)] mod tests { use super::build_schema_helper; diff --git a/datafusion/datasource-json/src/file_format.rs b/datafusion/datasource-json/src/file_format.rs index 1ef8ba7e4a957..62d03d67ccd43 100644 --- a/datafusion/datasource-json/src/file_format.rs +++ b/datafusion/datasource-json/src/file_format.rs @@ -615,3 +615,24 @@ impl Decoder for JsonDecoder { false } } + +/// Encode a [`JsonFormatFactory`]'s options as their protobuf form. +/// +/// The reverse direction is `From<&protobuf::JsonOptions> for JsonOptions` in +/// `datafusion-proto-models`: `JsonOptions` is a `datafusion-common` type, so +/// that half cannot live here. +#[cfg(feature = "proto")] +impl From<&JsonFormatFactory> for datafusion_proto_models::protobuf::JsonOptions { + fn from(factory: &JsonFormatFactory) -> Self { + if let Some(options) = &factory.options { + datafusion_proto_models::protobuf::JsonOptions { + compression: options.compression as i32, + schema_infer_max_rec: options.schema_infer_max_rec.map(|v| v as u64), + compression_level: options.compression_level, + newline_delimited: Some(options.newline_delimited), + } + } else { + datafusion_proto_models::protobuf::JsonOptions::default() + } + } +} diff --git a/datafusion/datasource-parquet/src/file_format.rs b/datafusion/datasource-parquet/src/file_format.rs index 8d19bedeb7155..88ca56476d579 100644 --- a/datafusion/datasource-parquet/src/file_format.rs +++ b/datafusion/datasource-parquet/src/file_format.rs @@ -680,3 +680,129 @@ pub fn statistics_from_parquet_meta_calc( ) -> Result { DFParquetMetadata::statistics_from_parquet_metadata(metadata, &table_schema) } + +#[cfg(feature = "proto")] +use datafusion_proto_models::protobuf::{self, parquet_column_options, parquet_options}; + +/// Encode a [`ParquetFormatFactory`]'s options as their protobuf form. +/// +/// The reverse direction is `TryFrom<&protobuf::TableParquetOptions> for +/// TableParquetOptions` in `datafusion-proto-models`: `TableParquetOptions` is +/// a `datafusion-common` type, so that half cannot live here. +#[cfg(feature = "proto")] +impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions { + fn from(factory: &ParquetFormatFactory) -> Self { + let global_options = if let Some(ref options) = factory.options { + options.clone() + } else { + return protobuf::TableParquetOptions::default(); + }; + + let column_specific_options = global_options.column_specific_options; + protobuf::TableParquetOptions { + global: Some(protobuf::ParquetOptions { + enable_page_index: global_options.global.enable_page_index, + pruning: global_options.global.pruning, + skip_metadata: global_options.global.skip_metadata, + metadata_size_hint_opt: global_options.global.metadata_size_hint.map(|size| { + parquet_options::MetadataSizeHintOpt::MetadataSizeHint(size as u64) + }), + pushdown_filters: global_options.global.pushdown_filters, + reorder_filters: global_options.global.reorder_filters, + force_filter_selections: global_options.global.force_filter_selections, + data_pagesize_limit: global_options.global.data_pagesize_limit as u64, + write_batch_size: global_options.global.write_batch_size as u64, + writer_version: global_options.global.writer_version.to_string(), + compression_opt: global_options.global.compression.map(|compression| { + parquet_options::CompressionOpt::Compression(compression) + }), + dictionary_enabled_opt: global_options.global.dictionary_enabled.map(|enabled| { + parquet_options::DictionaryEnabledOpt::DictionaryEnabled(enabled) + }), + dictionary_page_size_limit: global_options.global.dictionary_page_size_limit as u64, + statistics_enabled_opt: global_options.global.statistics_enabled.map(|enabled| { + parquet_options::StatisticsEnabledOpt::StatisticsEnabled(enabled) + }), + max_row_group_size: global_options.global.max_row_group_size as u64, + max_in_list_size: global_options.global.max_in_list_size as u64, + created_by: global_options.global.created_by.clone(), + column_index_truncate_length_opt: global_options.global.column_index_truncate_length.map(|length| { + parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(length as u64) + }), + statistics_truncate_length_opt: global_options.global.statistics_truncate_length.map(|length| { + parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(length as u64) + }), + data_page_row_count_limit: global_options.global.data_page_row_count_limit as u64, + encoding_opt: global_options.global.encoding.map(|encoding| { + parquet_options::EncodingOpt::Encoding(encoding) + }), + bloom_filter_on_read: global_options.global.bloom_filter_on_read, + bloom_filter_on_write: global_options.global.bloom_filter_on_write, + bloom_filter_fpp_opt: global_options.global.bloom_filter_fpp.map(|fpp| { + parquet_options::BloomFilterFppOpt::BloomFilterFpp(fpp) + }), + bloom_filter_ndv_opt: global_options.global.bloom_filter_ndv.map(|ndv| { + parquet_options::BloomFilterNdvOpt::BloomFilterNdv(ndv) + }), + allow_single_file_parallelism: global_options.global.allow_single_file_parallelism, + maximum_parallel_row_group_writers: global_options.global.maximum_parallel_row_group_writers as u64, + maximum_buffered_record_batches_per_stream: global_options.global.maximum_buffered_record_batches_per_stream as u64, + schema_force_view_types: global_options.global.schema_force_view_types, + binary_as_string: global_options.global.binary_as_string, + skip_arrow_metadata: global_options.global.skip_arrow_metadata, + coerce_int96_opt: global_options.global.coerce_int96.map(|compression| { + parquet_options::CoerceInt96Opt::CoerceInt96(compression) + }), + coerce_int96_tz_opt: global_options.global.coerce_int96_tz.map(|tz| { + parquet_options::CoerceInt96TzOpt::CoerceInt96Tz(tz) + }), + max_predicate_cache_size_opt: global_options.global.max_predicate_cache_size.map(|size| { + parquet_options::MaxPredicateCacheSizeOpt::MaxPredicateCacheSize(size as u64) + }), + max_row_group_bytes_opt: global_options.global.max_row_group_bytes.map(|size| { + parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(size.get() as u64) + }), + content_defined_chunking: Some(protobuf::ParquetCdcOptions { + enabled: global_options.global.content_defined_chunking.enabled, + min_chunk_size: global_options.global.content_defined_chunking.min_chunk_size as u64, + max_chunk_size: global_options.global.content_defined_chunking.max_chunk_size as u64, + norm_level: global_options.global.content_defined_chunking.norm_level, + }), + }), + column_specific_options: column_specific_options.into_iter().map(|(column_name, options)| { + protobuf::ParquetColumnSpecificOptions { + column_name, + options: Some(protobuf::ParquetColumnOptions { + bloom_filter_enabled_opt: options.bloom_filter_enabled.map(|enabled| { + parquet_column_options::BloomFilterEnabledOpt::BloomFilterEnabled(enabled) + }), + encoding_opt: options.encoding.map(|encoding| { + parquet_column_options::EncodingOpt::Encoding(encoding) + }), + dictionary_enabled_opt: options.dictionary_enabled.map(|enabled| { + parquet_column_options::DictionaryEnabledOpt::DictionaryEnabled(enabled) + }), + compression_opt: options.compression.map(|compression| { + parquet_column_options::CompressionOpt::Compression(compression) + }), + statistics_enabled_opt: options.statistics_enabled.map(|enabled| { + parquet_column_options::StatisticsEnabledOpt::StatisticsEnabled(enabled) + }), + bloom_filter_fpp_opt: options.bloom_filter_fpp.map(|fpp| { + parquet_column_options::BloomFilterFppOpt::BloomFilterFpp(fpp) + }), + bloom_filter_ndv_opt: options.bloom_filter_ndv.map(|ndv| { + parquet_column_options::BloomFilterNdvOpt::BloomFilterNdv(ndv) + }), + }) + } + }).collect(), + key_value_metadata: global_options.key_value_metadata + .iter() + .filter_map(|(key, value)| { + value.as_ref().map(|v| (key.clone(), v.clone())) + }) + .collect(), + } + } +} diff --git a/datafusion/proto/src/logical_plan/file_formats.rs b/datafusion/proto/src/logical_plan/file_formats.rs index bad014e874463..10c54cf55c5e7 100644 --- a/datafusion/proto/src/logical_plan/file_formats.rs +++ b/datafusion/proto/src/logical_plan/file_formats.rs @@ -18,7 +18,6 @@ use std::sync::Arc; use super::LogicalExtensionCodec; -use crate::convert::FromProto; use crate::protobuf::{CsvOptions as CsvOptionsProto, JsonOptions as JsonOptionsProto}; use datafusion_common::config::{CsvOptions, JsonOptions}; use datafusion_common::{TableReference, exec_datafusion_err, exec_err, not_impl_err}; @@ -32,48 +31,6 @@ use prost::Message; #[derive(Debug)] pub struct CsvLogicalExtensionCodec; -impl FromProto<&CsvFormatFactory> for CsvOptionsProto { - fn from_proto(factory: &CsvFormatFactory) -> Self { - if let Some(options) = &factory.options { - CsvOptionsProto { - has_header: options.has_header.map_or(vec![], |v| vec![v as u8]), - delimiter: vec![options.delimiter], - quote: vec![options.quote], - terminator: options.terminator.map_or(vec![], |v| vec![v]), - escape: options.escape.map_or(vec![], |v| vec![v]), - double_quote: options.double_quote.map_or(vec![], |v| vec![v as u8]), - compression: options.compression as i32, - schema_infer_max_rec: options.schema_infer_max_rec.map(|v| v as u64), - date_format: options.date_format.clone().unwrap_or_default(), - datetime_format: options.datetime_format.clone().unwrap_or_default(), - timestamp_format: options.timestamp_format.clone().unwrap_or_default(), - timestamp_tz_format: options - .timestamp_tz_format - .clone() - .unwrap_or_default(), - time_format: options.time_format.clone().unwrap_or_default(), - null_value: options.null_value.clone().unwrap_or_default(), - null_regex: options.null_regex.clone().unwrap_or_default(), - comment: options.comment.map_or(vec![], |v| vec![v]), - newlines_in_values: options - .newlines_in_values - .map_or(vec![], |v| vec![v as u8]), - truncated_rows: options.truncated_rows.map_or(vec![], |v| vec![v as u8]), - compression_level: options.compression_level, - quote_style: options.quote_style as i32, - ignore_leading_whitespace: options - .ignore_leading_whitespace - .map_or(vec![], |v| vec![v as u8]), - ignore_trailing_whitespace: options - .ignore_trailing_whitespace - .map_or(vec![], |v| vec![v as u8]), - } - } else { - CsvOptionsProto::default() - } - } -} - // TODO! This is a placeholder for now and needs to be implemented for real. impl LogicalExtensionCodec for CsvLogicalExtensionCodec { fn try_decode( @@ -137,7 +94,7 @@ impl LogicalExtensionCodec for CsvLogicalExtensionCodec { return exec_err!("{}", "Unsupported FileFormatFactory type".to_string()); }; - let proto = CsvOptionsProto::from_proto(&CsvFormatFactory { + let proto = CsvOptionsProto::from(&CsvFormatFactory { options: Some(options), }); @@ -149,21 +106,6 @@ impl LogicalExtensionCodec for CsvLogicalExtensionCodec { } } -impl FromProto<&JsonFormatFactory> for JsonOptionsProto { - fn from_proto(factory: &JsonFormatFactory) -> Self { - if let Some(options) = &factory.options { - JsonOptionsProto { - compression: options.compression as i32, - schema_infer_max_rec: options.schema_infer_max_rec.map(|v| v as u64), - compression_level: options.compression_level, - newline_delimited: Some(options.newline_delimited), - } - } else { - JsonOptionsProto::default() - } - } -} - #[derive(Debug)] pub struct JsonLogicalExtensionCodec; @@ -231,7 +173,7 @@ impl LogicalExtensionCodec for JsonLogicalExtensionCodec { return exec_err!("Unsupported FileFormatFactory type"); }; - let proto = JsonOptionsProto::from_proto(&JsonFormatFactory { + let proto = JsonOptionsProto::from(&JsonFormatFactory { options: Some(options), }); @@ -247,133 +189,10 @@ impl LogicalExtensionCodec for JsonLogicalExtensionCodec { mod parquet { use super::*; - use crate::protobuf::{ - ParquetCdcOptions as ParquetCdcOptionsProto, - ParquetColumnOptions as ParquetColumnOptionsProto, ParquetColumnSpecificOptions, - ParquetOptions as ParquetOptionsProto, - TableParquetOptions as TableParquetOptionsProto, parquet_column_options, - parquet_options, - }; + use crate::protobuf::TableParquetOptions as TableParquetOptionsProto; use datafusion_common::config::TableParquetOptions; use datafusion_datasource_parquet::file_format::ParquetFormatFactory; - impl FromProto<&ParquetFormatFactory> for TableParquetOptionsProto { - fn from_proto(factory: &ParquetFormatFactory) -> Self { - let global_options = if let Some(ref options) = factory.options { - options.clone() - } else { - return TableParquetOptionsProto::default(); - }; - - let column_specific_options = global_options.column_specific_options; - TableParquetOptionsProto { - global: Some(ParquetOptionsProto { - enable_page_index: global_options.global.enable_page_index, - pruning: global_options.global.pruning, - skip_metadata: global_options.global.skip_metadata, - metadata_size_hint_opt: global_options.global.metadata_size_hint.map(|size| { - parquet_options::MetadataSizeHintOpt::MetadataSizeHint(size as u64) - }), - pushdown_filters: global_options.global.pushdown_filters, - reorder_filters: global_options.global.reorder_filters, - force_filter_selections: global_options.global.force_filter_selections, - data_pagesize_limit: global_options.global.data_pagesize_limit as u64, - write_batch_size: global_options.global.write_batch_size as u64, - writer_version: global_options.global.writer_version.to_string(), - compression_opt: global_options.global.compression.map(|compression| { - parquet_options::CompressionOpt::Compression(compression) - }), - dictionary_enabled_opt: global_options.global.dictionary_enabled.map(|enabled| { - parquet_options::DictionaryEnabledOpt::DictionaryEnabled(enabled) - }), - dictionary_page_size_limit: global_options.global.dictionary_page_size_limit as u64, - statistics_enabled_opt: global_options.global.statistics_enabled.map(|enabled| { - parquet_options::StatisticsEnabledOpt::StatisticsEnabled(enabled) - }), - max_row_group_size: global_options.global.max_row_group_size as u64, - max_in_list_size: global_options.global.max_in_list_size as u64, - created_by: global_options.global.created_by.clone(), - column_index_truncate_length_opt: global_options.global.column_index_truncate_length.map(|length| { - parquet_options::ColumnIndexTruncateLengthOpt::ColumnIndexTruncateLength(length as u64) - }), - statistics_truncate_length_opt: global_options.global.statistics_truncate_length.map(|length| { - parquet_options::StatisticsTruncateLengthOpt::StatisticsTruncateLength(length as u64) - }), - data_page_row_count_limit: global_options.global.data_page_row_count_limit as u64, - encoding_opt: global_options.global.encoding.map(|encoding| { - parquet_options::EncodingOpt::Encoding(encoding) - }), - bloom_filter_on_read: global_options.global.bloom_filter_on_read, - bloom_filter_on_write: global_options.global.bloom_filter_on_write, - bloom_filter_fpp_opt: global_options.global.bloom_filter_fpp.map(|fpp| { - parquet_options::BloomFilterFppOpt::BloomFilterFpp(fpp) - }), - bloom_filter_ndv_opt: global_options.global.bloom_filter_ndv.map(|ndv| { - parquet_options::BloomFilterNdvOpt::BloomFilterNdv(ndv) - }), - allow_single_file_parallelism: global_options.global.allow_single_file_parallelism, - maximum_parallel_row_group_writers: global_options.global.maximum_parallel_row_group_writers as u64, - maximum_buffered_record_batches_per_stream: global_options.global.maximum_buffered_record_batches_per_stream as u64, - schema_force_view_types: global_options.global.schema_force_view_types, - binary_as_string: global_options.global.binary_as_string, - skip_arrow_metadata: global_options.global.skip_arrow_metadata, - coerce_int96_opt: global_options.global.coerce_int96.map(|compression| { - parquet_options::CoerceInt96Opt::CoerceInt96(compression) - }), - coerce_int96_tz_opt: global_options.global.coerce_int96_tz.map(|tz| { - parquet_options::CoerceInt96TzOpt::CoerceInt96Tz(tz) - }), - max_predicate_cache_size_opt: global_options.global.max_predicate_cache_size.map(|size| { - parquet_options::MaxPredicateCacheSizeOpt::MaxPredicateCacheSize(size as u64) - }), - max_row_group_bytes_opt: global_options.global.max_row_group_bytes.map(|size| { - parquet_options::MaxRowGroupBytesOpt::MaxRowGroupBytes(size.get() as u64) - }), - content_defined_chunking: Some(ParquetCdcOptionsProto { - enabled: global_options.global.content_defined_chunking.enabled, - min_chunk_size: global_options.global.content_defined_chunking.min_chunk_size as u64, - max_chunk_size: global_options.global.content_defined_chunking.max_chunk_size as u64, - norm_level: global_options.global.content_defined_chunking.norm_level, - }), - }), - column_specific_options: column_specific_options.into_iter().map(|(column_name, options)| { - ParquetColumnSpecificOptions { - column_name, - options: Some(ParquetColumnOptionsProto { - bloom_filter_enabled_opt: options.bloom_filter_enabled.map(|enabled| { - parquet_column_options::BloomFilterEnabledOpt::BloomFilterEnabled(enabled) - }), - encoding_opt: options.encoding.map(|encoding| { - parquet_column_options::EncodingOpt::Encoding(encoding) - }), - dictionary_enabled_opt: options.dictionary_enabled.map(|enabled| { - parquet_column_options::DictionaryEnabledOpt::DictionaryEnabled(enabled) - }), - compression_opt: options.compression.map(|compression| { - parquet_column_options::CompressionOpt::Compression(compression) - }), - statistics_enabled_opt: options.statistics_enabled.map(|enabled| { - parquet_column_options::StatisticsEnabledOpt::StatisticsEnabled(enabled) - }), - bloom_filter_fpp_opt: options.bloom_filter_fpp.map(|fpp| { - parquet_column_options::BloomFilterFppOpt::BloomFilterFpp(fpp) - }), - bloom_filter_ndv_opt: options.bloom_filter_ndv.map(|ndv| { - parquet_column_options::BloomFilterNdvOpt::BloomFilterNdv(ndv) - }), - }) - } - }).collect(), - key_value_metadata: global_options.key_value_metadata - .iter() - .filter_map(|(key, value)| { - value.as_ref().map(|v| (key.clone(), v.clone())) - }) - .collect(), - } - } - } - #[derive(Debug)] pub struct ParquetLogicalExtensionCodec; @@ -445,7 +264,7 @@ mod parquet { return exec_err!("Unsupported FileFormatFactory type"); }; - let proto = TableParquetOptionsProto::from_proto(&ParquetFormatFactory { + let proto = TableParquetOptionsProto::from(&ParquetFormatFactory { options: Some(options), }); @@ -460,6 +279,7 @@ mod parquet { #[cfg(test)] mod tests { use super::*; + use crate::protobuf::ParquetOptions as ParquetOptionsProto; use datafusion_common::config::ParquetOptions; fn encode_table_options(proto: TableParquetOptionsProto) -> Vec { From cbd8c17675e2c87f4623512957f055625435567c Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sun, 9 Aug 2026 10:32:01 -0400 Subject: [PATCH 5/7] test(proto): guard the public From/TryFrom proto conversions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restores `From<&protobuf::PhysicalColumn> for Column` (and adds the encoding direction, which `Column::try_to_proto` now uses) — the last of the 39 conversions 54.1.0 published. Adds `tests/cases/public_conversions.rs`, which coerces every one of those conversions to a `fn` pointer. `cargo-semver-checks` has no lint for a removed hand-written trait impl, which is why this class of break went unnoticed; a compile-time reference does catch it, and stays quiet when an impl merely moves between crates. Part of #24019. Co-Authored-By: Claude Opus 5 --- .../physical-expr/src/expressions/column.rs | 28 ++-- datafusion/proto/tests/cases/mod.rs | 1 + .../proto/tests/cases/public_conversions.rs | 128 ++++++++++++++++++ 3 files changed, 149 insertions(+), 8 deletions(-) create mode 100644 datafusion/proto/tests/cases/public_conversions.rs diff --git a/datafusion/physical-expr/src/expressions/column.rs b/datafusion/physical-expr/src/expressions/column.rs index 0a96b00444850..482ab6ef1e787 100644 --- a/datafusion/physical-expr/src/expressions/column.rs +++ b/datafusion/physical-expr/src/expressions/column.rs @@ -155,16 +155,28 @@ impl PhysicalExpr for Column { use datafusion_proto_models::protobuf; Ok(Some(protobuf::PhysicalExprNode { expr_id: None, - expr_type: Some(protobuf::physical_expr_node::ExprType::Column( - protobuf::PhysicalColumn { - name: self.name.clone(), - index: self.index as u32, - }, - )), + expr_type: Some(protobuf::physical_expr_node::ExprType::Column(self.into())), })) } } +#[cfg(feature = "proto")] +impl From<&datafusion_proto_models::protobuf::PhysicalColumn> for Column { + fn from(c: &datafusion_proto_models::protobuf::PhysicalColumn) -> Self { + Column::new(&c.name, c.index as usize) + } +} + +#[cfg(feature = "proto")] +impl From<&Column> for datafusion_proto_models::protobuf::PhysicalColumn { + fn from(c: &Column) -> Self { + Self { + name: c.name.clone(), + index: c.index as u32, + } + } +} + #[cfg(feature = "proto")] impl Column { /// Reconstruct a [`Column`] from its protobuf representation. @@ -184,12 +196,12 @@ impl Column { ) -> Result> { use datafusion_physical_expr_common::expect_expr_variant; use datafusion_proto_models::protobuf; - let protobuf::PhysicalColumn { name, index } = expect_expr_variant!( + let column = expect_expr_variant!( node, protobuf::physical_expr_node::ExprType::Column, "Column", ); - Ok(Arc::new(Column::new(name, *index as usize))) + Ok(Arc::new(Column::from(column))) } } diff --git a/datafusion/proto/tests/cases/mod.rs b/datafusion/proto/tests/cases/mod.rs index 7a95ee0c29e5d..bdf626eb41a0a 100644 --- a/datafusion/proto/tests/cases/mod.rs +++ b/datafusion/proto/tests/cases/mod.rs @@ -32,6 +32,7 @@ use std::fmt::Debug; use std::hash::Hash; use std::sync::Arc; +mod public_conversions; mod roundtrip_logical_plan; mod roundtrip_physical_plan; mod serialize; diff --git a/datafusion/proto/tests/cases/public_conversions.rs b/datafusion/proto/tests/cases/public_conversions.rs new file mode 100644 index 0000000000000..1cda8e01a0765 --- /dev/null +++ b/datafusion/proto/tests/cases/public_conversions.rs @@ -0,0 +1,128 @@ +// 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. + +//! Compile-time guard for the `From` / `TryFrom` conversions between DataFusion +//! types and `datafusion_proto::protobuf` messages that downstream crates call. +//! +//! These impls were silently dropped once (see +//! ): they were replaced by +//! crate-local conversion traits as a stopgap during the `datafusion-proto-models` +//! extraction, and `cargo-semver-checks` has no lint for a removed hand-written +//! trait impl, so nothing caught the break. Coercing each conversion to a `fn` +//! pointer here does — moving an impl between crates is fine, removing one stops +//! compiling. +//! +//! Only the spelling is asserted. Behaviour is covered by the round-trip tests +//! next to each impl. + +use datafusion_common::config::{ + CsvOptions, JsonOptions, ParquetCdcOptions, ParquetColumnOptions, ParquetOptions, + TableParquetOptions, +}; +use datafusion_common::display::StringifiedPlan; +use datafusion_common::{ + JoinConstraint, JoinType, NullEquality, TableReference, UnnestOptions, +}; +use datafusion_datasource::file_groups::FileGroup; +use datafusion_datasource::file_sink_config::FileSinkConfig; +use datafusion_datasource::{FileRange, PartitionedFile}; +use datafusion_datasource_csv::file_format::{CsvFormatFactory, CsvSink}; +use datafusion_datasource_json::file_format::{JsonFormatFactory, JsonSink}; +use datafusion_datasource_parquet::file_format::{ParquetFormatFactory, ParquetSink}; +use datafusion_expr::dml::MergeIntoClauseKind; +use datafusion_expr::expr::NullTreatment; +use datafusion_expr::{WindowFrame, WindowFrameBound, WindowFrameUnits}; +use datafusion_physical_expr::expressions::Column; +use datafusion_proto::protobuf; + +/// Asserts `T: From` by naming the conversion. +fn assert_from>() { + let _: fn(F) -> T = From::from; +} + +/// Asserts `T: TryFrom` by naming the conversion. +fn assert_try_from>() { + let _: fn(F) -> Result = TryFrom::try_from; +} + +#[test] +fn file_scan_conversions_are_std_traits() { + assert_try_from::<&protobuf::PartitionedFile, PartitionedFile>(); + assert_try_from::<&PartitionedFile, protobuf::PartitionedFile>(); + assert_try_from::<&protobuf::FileRange, FileRange>(); + assert_try_from::<&FileRange, protobuf::FileRange>(); + assert_try_from::<&protobuf::FileGroup, FileGroup>(); + assert_try_from::<&FileGroup, protobuf::FileGroup>(); + assert_from::<&protobuf::PhysicalColumn, Column>(); + assert_from::<&Column, protobuf::PhysicalColumn>(); + assert_try_from::<&[PartitionedFile], protobuf::FileGroup>(); +} + +#[test] +fn file_sink_conversions_are_std_traits() { + assert_try_from::<&protobuf::FileSinkConfig, FileSinkConfig>(); + assert_try_from::<&FileSinkConfig, protobuf::FileSinkConfig>(); + assert_try_from::<&protobuf::JsonSink, JsonSink>(); + assert_try_from::<&JsonSink, protobuf::JsonSink>(); + assert_try_from::<&protobuf::CsvSink, CsvSink>(); + assert_try_from::<&CsvSink, protobuf::CsvSink>(); + assert_try_from::<&protobuf::ParquetSink, ParquetSink>(); + assert_try_from::<&ParquetSink, protobuf::ParquetSink>(); +} + +#[test] +fn window_frame_conversions_are_std_traits() { + assert_try_from::(); + assert_try_from::<&WindowFrame, protobuf::WindowFrame>(); + assert_try_from::(); + assert_try_from::<&WindowFrameBound, protobuf::WindowFrameBound>(); + assert_from::(); + assert_from::(); + assert_from::(); + assert_from::(); + assert_from::(); + assert_from::(); +} + +#[test] +fn common_type_conversions_are_std_traits() { + assert_from::<&protobuf::UnnestOptions, UnnestOptions>(); + assert_from::<&UnnestOptions, protobuf::UnnestOptions>(); + assert_try_from::(); + assert_from::(); + assert_from::<&protobuf::StringifiedPlan, StringifiedPlan>(); + assert_from::<&StringifiedPlan, protobuf::StringifiedPlan>(); + assert_from::(); + assert_from::(); + assert_from::(); + assert_from::(); + assert_from::(); + assert_from::(); +} + +#[test] +fn file_format_option_conversions_are_std_traits() { + assert_from::<&protobuf::CsvOptions, CsvOptions>(); + assert_from::<&protobuf::JsonOptions, JsonOptions>(); + assert_try_from::<&protobuf::ParquetOptions, ParquetOptions>(); + assert_from::(); + assert_from::(); + assert_try_from::<&protobuf::TableParquetOptions, TableParquetOptions>(); + assert_from::<&CsvFormatFactory, protobuf::CsvOptions>(); + assert_from::<&JsonFormatFactory, protobuf::JsonOptions>(); + assert_from::<&ParquetFormatFactory, protobuf::TableParquetOptions>(); +} From 8a101e58ecd647d8e7cad8e5aa713084aa02673c Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sun, 9 Aug 2026 16:14:27 -0400 Subject: [PATCH 6/7] docs: document the one proto conversion that changed shape since 54.1.0 The parquet options conversions validate `writer_version` now, so they are `TryFrom` rather than `From`. Everything else needs no entry: the impls that moved crates still resolve unchanged, because trait impls are global. Part of #24019. Co-Authored-By: Claude Opus 5 --- .../library-user-guide/upgrading/55.0.0.md | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md index 2811fb4df2900..58ff2c6dc4b9e 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -1188,3 +1188,29 @@ See [PR #24030](https://github.com/apache/datafusion/pull/24030) for details. [`bufwriter`]: https://docs.rs/tokio/latest/tokio/io/struct.BufWriter.html [`asyncarrowwriter`]: https://docs.rs/parquet/59.1.0/parquet/arrow/async_writer/struct.AsyncArrowWriter.html [`parquet/examples/object_store.rs`]: https://github.com/apache/arrow-rs/blob/main/parquet/examples/object_store.rs + +### `datafusion-proto`: parquet options conversions are fallible + +`protobuf::ParquetOptions` and `protobuf::TableParquetOptions` validate +`writer_version` when converting into their `datafusion-common` counterparts, so +those conversions are `TryFrom` rather than `From`. + +Every other `From` / `TryFrom` conversion between DataFusion types and +`datafusion_proto::protobuf` messages is unchanged. Several impls moved to the +crate that owns their DataFusion type, but trait impls are global, so +`X::try_from(&proto)` and `proto.try_into()` still resolve with no import +changes. + +**Migration guide:** + +```rust,ignore +// Before +let opts = ParquetOptions::from(&proto_opts); +let table_opts = TableParquetOptions::from(&proto_table_opts); + +// After +let opts = ParquetOptions::try_from(&proto_opts)?; +let table_opts = TableParquetOptions::try_from(&proto_table_opts)?; +``` + +See [issue #24019](https://github.com/apache/datafusion/issues/24019) for details. From 783d5ad69fc21adee8bc641ac1903ae911a8becb Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Sun, 9 Aug 2026 14:01:02 -0400 Subject: [PATCH 7/7] refactor(proto): remove the FromProto / TryFromProto workaround #21929 introduced these traits so the `datafusion-proto-models` extraction could land without simultaneously relocating ~39 conversions, and flagged them there as "a known workaround, not the end state". Every one of those conversions now lives in a crate that owns one side of it, as a plain `From` / `TryFrom`, so the traits and `convert_required_proto!` have no implementors and no callers. Neither trait has ever shipped in a release, so this removes them outright rather than deprecating: there is nothing for downstream users to migrate off. Doing it before 55.0.0 keeps the workaround out of the released API entirely. Closes #24019. Co-Authored-By: Claude Opus 5 --- .../physical-expr-common/src/physical_expr.rs | 4 +- datafusion/proto/src/common.rs | 18 -------- datafusion/proto/src/convert.rs | 44 ------------------- datafusion/proto/src/lib.rs | 3 -- 4 files changed, 2 insertions(+), 67 deletions(-) delete mode 100644 datafusion/proto/src/convert.rs diff --git a/datafusion/physical-expr-common/src/physical_expr.rs b/datafusion/physical-expr-common/src/physical_expr.rs index 679a44e85ee9a..59393e75786bd 100644 --- a/datafusion/physical-expr-common/src/physical_expr.rs +++ b/datafusion/physical-expr-common/src/physical_expr.rs @@ -498,8 +498,8 @@ pub trait PhysicalExpr: Any + Send + Sync + Display + Debug + DynEq + DynHash { /// . /// /// The `try_` prefix matches the fallible `try_from_proto` decode - /// constructors (and the `TryFromProto` trait in `datafusion-proto`); - /// both sides of the round-trip are fallible and named consistently. + /// constructors; both sides of the round-trip are fallible and named + /// consistently. /// /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode #[cfg(feature = "proto")] diff --git a/datafusion/proto/src/common.rs b/datafusion/proto/src/common.rs index bff017edbc998..22ded708d8c71 100644 --- a/datafusion/proto/src/common.rs +++ b/datafusion/proto/src/common.rs @@ -47,24 +47,6 @@ macro_rules! convert_required { }}; } -/// Like [`convert_required`] but for types whose proto conversion goes through -/// the [`TryFromProto`](crate::convert::TryFromProto) trait instead of -/// [`TryFrom`]. Required because some prost-generated types now live in a -/// separate crate, so `TryFrom`/`From` cannot be implemented on foreign-foreign -/// pairs from `datafusion-proto` directly. -#[macro_export] -macro_rules! convert_required_proto { - ($T:ty, $PB:expr) => {{ - if let Some(field) = $PB.as_ref() { - Ok::<$T, _>(<$T as $crate::convert::TryFromProto<_>>::try_from_proto( - field, - )?) - } else { - Err(proto_error("Missing required field in protobuf")) - } - }}; -} - #[macro_export] macro_rules! into_required { ($PB:expr) => {{ diff --git a/datafusion/proto/src/convert.rs b/datafusion/proto/src/convert.rs deleted file mode 100644 index 87e9a431dcb80..0000000000000 --- a/datafusion/proto/src/convert.rs +++ /dev/null @@ -1,44 +0,0 @@ -// 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. - -//! Conversion traits between proto-generated types and DataFusion types. -//! -//! The `prost`-generated structs now live in `datafusion-proto-models`, while -//! their counterparts (`StringifiedPlan`, `JoinType`, `WindowFrame`, ...) live -//! in `datafusion-common` / `datafusion-expr` / `datafusion-datasource` etc. -//! Both sides are foreign to `datafusion-proto`, which means the orphan rule -//! forbids a direct `impl From<&protobuf::X> for Y` written here. -//! -//! To keep the conversion logic colocated with serialization while satisfying -//! the orphan rule, we route those conversions through the `FromProto` / -//! `TryFromProto` traits defined in this module. Their signatures mirror the -//! standard library's `From` / `TryFrom`, so callers spell the conversion -//! `Y::from_proto(&p)` / `Y::try_from_proto(&p)?` instead of -//! `(&p).into()` / `(&p).try_into()?`. - -/// Infallible conversion from a proto value into a DataFusion value (or vice -/// versa). Mirrors [`From`]. -pub trait FromProto: Sized { - fn from_proto(value: T) -> Self; -} - -/// Fallible conversion from a proto value into a DataFusion value (or vice -/// versa). Mirrors [`TryFrom`]. -pub trait TryFromProto: Sized { - type Error; - fn try_from_proto(value: T) -> Result; -} diff --git a/datafusion/proto/src/lib.rs b/datafusion/proto/src/lib.rs index 0e63bcf5f5acb..71feae506dc6f 100644 --- a/datafusion/proto/src/lib.rs +++ b/datafusion/proto/src/lib.rs @@ -123,12 +123,9 @@ //! ``` pub mod bytes; pub mod common; -pub mod convert; pub mod logical_plan; pub mod physical_plan; -pub use convert::{FromProto, TryFromProto}; - pub mod protobuf { pub use datafusion_proto_common::common::proto_error; pub use datafusion_proto_common::protobuf_common::{