You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/07/30 18:27:40 UTC
(datafusion) branch gh-readonly-queue/main/pr-24006-f398301748b4a6f5279fc4ff553103e1c778469a created (now 30ae8bf340)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-24006-f398301748b4a6f5279fc4ff553103e1c778469a
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 30ae8bf340 refactor(proto): move PartitionedFile / FileGroup serde into datafusion-datasource (#24006)
This branch includes the following new commits:
new 30ae8bf340 refactor(proto): move PartitionedFile / FileGroup serde into datafusion-datasource (#24006)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: refactor(proto): move PartitionedFile / FileGroup serde into datafusion-datasource (#24006)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24006-f398301748b4a6f5279fc4ff553103e1c778469a
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 30ae8bf34070399966a7015678fe70bf31b3eeda
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Thu Jul 30 13:27:17 2026 -0500
refactor(proto): move PartitionedFile / FileGroup serde into datafusion-datasource (#24006)
## Which issue does this PR close?
- Part of #23494. Precursor for #23497 / #23683 (`DataSource` /
`FileSource` proto hooks) and for #23752.
## Rationale for this change
The protobuf conversions for the file-scan leaf types —
`PartitionedFile`,
`FileGroup`, `FileRange` — live in `datafusion-proto` as `TryFromProto`
impls,
because that is historically the only crate that can name both sides
(the
DataFusion type and the prost message are both foreign to it, hence the
`TryFromProto` workaround trait in the first place).
That placement means any *other* crate that needs those conversions has
to
reimplement them. #23683 hits exactly this: a `FileSource` serializing
its own
scan config needs to encode file groups, so the first cut of that PR
grew a
private second copy of the `PartitionedFile` wire logic inside
`datafusion-datasource`, which can then drift from the central
serializer.
The same will be true of every source migrated under #23516–#23518.
Nothing about these conversions needs `datafusion-proto`: they are plain
data,
with `ScalarValue` / `Statistics` / `Schema` going through
`datafusion-proto-common`. They belong next to the types.
## What changes are included in this PR?
- New `datafusion_datasource::proto` module, behind a new `proto`
feature on
`datafusion-datasource` (off by default; `datafusion-proto` enables it):
- `FileRange::try_to_proto` / `try_from_proto`
- `PartitionedFile::try_to_proto` / `try_from_proto`
- `FileGroup` <-> `protobuf::FileGroup`
- `datafusion-proto`'s `TryFromProto` impls for those types become
one-line
shims delegating to the new impls, so every existing caller keeps
working
and the two sides cannot disagree.
### Why these are `TryFrom` and not `try_to_proto` hooks
`TryFromProto` exists because `datafusion-proto` owns neither side of
the
conversions it hosts: with both the DataFusion type and the prost
message
foreign to it, `impl TryFrom<protobuf::X> for X` is rejected by the
orphan
rule, so a local trait was the only way to say the same thing.
Moving a conversion into the crate that owns the DataFusion type removes
that
constraint, and `&T` is `#[fundamental]`, so both directions are
expressible
with the standard trait (checked, not assumed):
```rust
impl TryFrom<&protobuf::PartitionedFile> for PartitionedFile // ok
impl TryFrom<&PartitionedFile> for protobuf::PartitionedFile // ok
impl TryFrom<&[PartitionedFile]> for protobuf::FileGroup // E0117
```
The last one is why `protobuf::FileGroup`'s *slice* conversion stays a
`TryFromProto` shim: `&[PartitionedFile]` is not a type this crate owns,
while
`&FileGroup` is. Callers inside DataFusion go through `FileGroup`.
So the rule this PR sets for the rest of #23494: **plain data uses
`TryFrom`;
anything needing an encode/decode context keeps the `try_to_proto(ctx)`
/
`try_from_proto(node, ctx)` hooks**, because the standard trait cannot
carry
that second argument. Usefully, none of the ~40
`TryFromProto`/`FromProto`
impls needs a context, and nothing that needs one was ever a
`TryFromProto`
impl — the two categories are already disjoint, so the shape now tells a
reader whether a conversion recurses.
### Why now
`FromProto` / `TryFromProto` were added in #21929, *after* the 54.0.0
release,
and 54.1.0 was cut before any of this landed — so they have never
shipped in a
release. Replacing them with the standard traits, and eventually
deleting them,
is a no-op for semver **today** and a major breaking change the moment
55.0.0
goes out. The same applies to the six inherent `try_to_proto` /
`try_from_proto`
methods this PR would otherwise have added: they are new, unreleased
API, so
choosing their final shape costs nothing right now.
The other reason to settle it here rather than in a follow-up: this is
the PR
that establishes the pattern for the data-source family (#23516-#23519
and
#23752 / #23781 are all queued behind it). Whichever shape merges first
is the
one they will copy.
Retiring the remaining ~34 impls is still its own follow-up. Two notes
for
whoever picks it up: the sink and format-option conversions can move
next to
their types the same way, but the ones for `datafusion-common`-owned
types
(`JoinType`, `NullEquality`, `TableReference`, `UnnestOptions`, ...)
cannot —
`datafusion-common` cannot depend on `datafusion-proto-models` (it is
underneath it via `datafusion-proto-common`). Their legal home is
`proto-models` itself, implementing on the local proto type, which is
already
how `proto-common` hosts the `ScalarValue` / `Statistics` conversions.
## Are these changes tested?
Yes.
- New unit tests in `datafusion_datasource::proto` covering the
`PartitionedFile` round trip (path, size, mtime, partition values,
range,
arrow schema, statistics), the `FileGroup` round trip, and the
invalid-path
error.
- The existing `datafusion-proto` tests now exercise the delegating
shims, so
they also pin the shims themselves.
- `datafusion-proto`, all features: 227 passed / 0 failed.
- `datafusion-datasource` with `proto`: 180 passed / 0 failed.
- Full workspace run: 10347 passed
(`cargo test --profile ci --workspace --lib --tests --features
avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption`).
The only failures are 8 backtrace-symbolization tests in
`datafusion-common`,
a crate this PR does not touch and which sits below every crate it does;
they
fail the same way on the base commit on macOS.
- `cargo fmt` and `ci/scripts/rust_clippy.sh` clean.
## Are there any user-facing changes?
The protobuf wire format is unchanged, and no existing API changes
shape.
Additive:
- New `proto` feature on `datafusion-datasource` (off by default).
- New `TryFrom` impls in both directions between `FileRange`,
`PartitionedFile`, `FileGroup` and their protobuf messages, under that
feature. No new names are added to the crate's API surface: the trait is
`core::convert::TryFrom`.
Note for reviewers: while writing the round-trip test I found that
`PartitionedFile` statistics do not round-trip cleanly on `main` — filed
as
#23998. This PR preserves that behavior exactly rather than changing
decode
semantics in a refactor; the test documents it.
---------
Co-authored-by: Claude Opus 5 <no...@anthropic.com>
---
Cargo.lock | 1 +
datafusion/datasource/Cargo.toml | 5 +
datafusion/datasource/src/mod.rs | 4 +
datafusion/datasource/src/proto.rs | 238 +++++++++++++++++++++++
datafusion/proto/Cargo.toml | 2 +-
datafusion/proto/src/physical_plan/from_proto.rs | 54 +----
datafusion/proto/src/physical_plan/to_proto.rs | 47 ++---
7 files changed, 272 insertions(+), 79 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 451ff70b1c..9cd5d12d5b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1944,6 +1944,7 @@ dependencies = [
"datafusion-physical-expr-adapter",
"datafusion-physical-expr-common",
"datafusion-physical-plan",
+ "datafusion-proto-models",
"datafusion-session",
"flate2",
"futures",
diff --git a/datafusion/datasource/Cargo.toml b/datafusion/datasource/Cargo.toml
index 2ac42ed900..459ca436f3 100644
--- a/datafusion/datasource/Cargo.toml
+++ b/datafusion/datasource/Cargo.toml
@@ -34,6 +34,10 @@ all-features = true
backtrace = ["datafusion-common/backtrace"]
compression = ["async-compression", "liblzma", "bzip2", "flate2", "zstd", "tokio-util"]
default = ["compression"]
+# Enables the protobuf conversions for the file-scan leaf types owned by this
+# crate (`FileRange`, `PartitionedFile`, `FileGroup`). Off by default so
+# consumers that never serialize plans pay nothing.
+proto = ["dep:datafusion-proto-models"]
[dependencies]
arrow = { workspace = true }
@@ -56,6 +60,7 @@ datafusion-physical-expr = { workspace = true }
datafusion-physical-expr-adapter = { workspace = true }
datafusion-physical-expr-common = { workspace = true }
datafusion-physical-plan = { workspace = true }
+datafusion-proto-models = { workspace = true, optional = true }
datafusion-session = { workspace = true }
flate2 = { workspace = true, optional = true }
futures = { workspace = true }
diff --git a/datafusion/datasource/src/mod.rs b/datafusion/datasource/src/mod.rs
index 7c8cae337f..e415b3e48a 100644
--- a/datafusion/datasource/src/mod.rs
+++ b/datafusion/datasource/src/mod.rs
@@ -41,6 +41,10 @@ pub mod file_stream;
pub mod memory;
pub mod morsel;
pub mod projection;
+/// Protobuf conversions for [`FileRange`], [`PartitionedFile`] and
+/// [`FileGroup`](crate::file_groups::FileGroup), gated on the `proto` feature.
+#[cfg(feature = "proto")]
+mod proto;
pub mod schema_adapter;
pub mod sink;
pub mod source;
diff --git a/datafusion/datasource/src/proto.rs b/datafusion/datasource/src/proto.rs
new file mode 100644
index 0000000000..cf48a46165
--- /dev/null
+++ b/datafusion/datasource/src/proto.rs
@@ -0,0 +1,238 @@
+// 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 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.
+//!
+//! 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
+//! they are plain [`TryFrom`] impls rather than the `try_to_proto(ctx)` /
+//! `try_from_proto(node, ctx)` hooks used for plans, expressions and scan
+//! configs: the standard trait can express a conversion that takes nothing but
+//! the value, and the orphan rule allows it here because one side of each
+//! conversion is a type this crate owns.
+
+use std::sync::Arc;
+
+use chrono::{TimeZone, Utc};
+use datafusion_common::{DataFusionError, Result, internal_datafusion_err};
+use datafusion_proto_models::protobuf;
+use object_store::ObjectMeta;
+use object_store::path::Path;
+
+use crate::file_groups::FileGroup;
+use crate::{FileRange, PartitionedFile};
+
+impl TryFrom<&FileRange> for protobuf::FileRange {
+ type Error = DataFusionError;
+
+ fn try_from(range: &FileRange) -> Result<Self> {
+ Ok(protobuf::FileRange {
+ start: range.start,
+ end: range.end,
+ })
+ }
+}
+
+impl TryFrom<&protobuf::FileRange> for FileRange {
+ type Error = DataFusionError;
+
+ fn try_from(range: &protobuf::FileRange) -> Result<Self> {
+ Ok(FileRange {
+ start: range.start,
+ end: range.end,
+ })
+ }
+}
+
+impl TryFrom<&PartitionedFile> for protobuf::PartitionedFile {
+ type Error = DataFusionError;
+
+ fn try_from(file: &PartitionedFile) -> Result<Self> {
+ let last_modified = file.object_meta.last_modified;
+ let last_modified_ns = last_modified.timestamp_nanos_opt().ok_or_else(|| {
+ DataFusionError::Plan(format!(
+ "Invalid timestamp on PartitionedFile::ObjectMeta: {last_modified}"
+ ))
+ })? as u64;
+ Ok(protobuf::PartitionedFile {
+ arrow_schema: file
+ .arrow_schema
+ .as_ref()
+ .map(|s| s.as_ref().try_into())
+ .transpose()?,
+ path: file.object_meta.location.as_ref().to_owned(),
+ size: file.object_meta.size,
+ last_modified_ns,
+ partition_values: file
+ .partition_values
+ .iter()
+ .map(|v| v.try_into())
+ .collect::<Result<Vec<_>, _>>()?,
+ range: file.range.as_ref().map(TryInto::try_into).transpose()?,
+ statistics: file.statistics.as_ref().map(|s| s.as_ref().into()),
+ })
+ }
+}
+
+impl TryFrom<&protobuf::PartitionedFile> for PartitionedFile {
+ type Error = DataFusionError;
+
+ fn try_from(file: &protobuf::PartitionedFile) -> Result<Self> {
+ let mut pf = PartitionedFile::new_from_meta(ObjectMeta {
+ location: Path::parse(file.path.as_str()).map_err(|e| {
+ internal_datafusion_err!("Invalid object_store path: {e}")
+ })?,
+ last_modified: Utc.timestamp_nanos(file.last_modified_ns as i64),
+ size: file.size,
+ e_tag: None,
+ version: None,
+ })
+ .with_partition_values(
+ file.partition_values
+ .iter()
+ .map(|v| v.try_into())
+ .collect::<Result<Vec<_>, _>>()?,
+ );
+ if let Some(proto_schema) = file.arrow_schema.as_ref() {
+ pf = pf.with_arrow_schema(Arc::new(
+ proto_schema.try_into().map_err(DataFusionError::from)?,
+ ));
+ }
+ if let Some(range) = file.range.as_ref() {
+ let range = FileRange::try_from(range)?;
+ pf = pf.with_range(range.start, range.end);
+ }
+ if let Some(proto_stats) = file.statistics.as_ref() {
+ // The wire format carries statistics for the full table schema (file + partition
+ // columns), so assign directly — `with_statistics` would append the partition
+ // column stats a second time.
+ pf.statistics = Some(Arc::new(proto_stats.try_into()?));
+ }
+ Ok(pf)
+ }
+}
+
+impl TryFrom<&FileGroup> for protobuf::FileGroup {
+ type Error = DataFusionError;
+
+ fn try_from(group: &FileGroup) -> Result<Self> {
+ Ok(protobuf::FileGroup {
+ files: group
+ .files()
+ .iter()
+ .map(TryInto::try_into)
+ .collect::<Result<Vec<_>>>()?,
+ })
+ }
+}
+
+impl TryFrom<&protobuf::FileGroup> for FileGroup {
+ type Error = DataFusionError;
+
+ fn try_from(group: &protobuf::FileGroup) -> Result<Self> {
+ Ok(FileGroup::new(
+ group
+ .files
+ .iter()
+ .map(TryInto::try_into)
+ .collect::<Result<Vec<_>>>()?,
+ ))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use arrow::datatypes::{DataType, Field, Schema};
+ use datafusion_common::{ScalarValue, Statistics};
+
+ use super::*;
+
+ #[test]
+ fn partitioned_file_roundtrip_preserves_all_fields() -> Result<()> {
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
+ let pf = PartitionedFile::new_from_meta(ObjectMeta {
+ location: Path::parse("foo/bar.parquet")?,
+ last_modified: Utc.timestamp_nanos(1_000_000_000),
+ size: 1234,
+ e_tag: None,
+ version: None,
+ })
+ .with_partition_values(vec![ScalarValue::from("2024-01-01")])
+ .with_range(10, 20)
+ .with_arrow_schema(Arc::clone(&schema))
+ .with_statistics(Arc::new(Statistics::new_unknown(&schema)));
+
+ let encoded = protobuf::PartitionedFile::try_from(&pf)?;
+ let decoded = PartitionedFile::try_from(&encoded)?;
+
+ assert_eq!(decoded.object_meta.location, pf.object_meta.location);
+ assert_eq!(decoded.object_meta.size, pf.object_meta.size);
+ assert_eq!(
+ decoded.object_meta.last_modified,
+ pf.object_meta.last_modified
+ );
+ assert_eq!(decoded.partition_values, pf.partition_values);
+ assert_eq!(decoded.range, pf.range);
+ assert_eq!(decoded.arrow_schema.as_deref(), Some(schema.as_ref()));
+ // Statistics span the full table schema (file columns followed by one
+ // entry per partition column), and survive the round trip intact.
+ assert_eq!(
+ pf.statistics.as_ref().unwrap().column_statistics.len(),
+ schema.fields().len() + pf.partition_values.len()
+ );
+ assert_eq!(decoded.statistics, pf.statistics);
+ Ok(())
+ }
+
+ #[test]
+ fn partitioned_file_from_proto_rejects_invalid_path() {
+ let proto = protobuf::PartitionedFile {
+ path: "foo//bar.parquet".to_string(),
+ ..Default::default()
+ };
+
+ let err = PartitionedFile::try_from(&proto).unwrap_err();
+ assert!(
+ err.to_string().contains("Invalid object_store path"),
+ "unexpected error: {err}"
+ );
+ }
+
+ #[test]
+ fn file_group_roundtrip() -> Result<()> {
+ let group = FileGroup::new(vec![
+ PartitionedFile::new("a.parquet", 1),
+ PartitionedFile::new("b.parquet", 2),
+ ]);
+
+ let encoded = protobuf::FileGroup::try_from(&group)?;
+ let decoded = FileGroup::try_from(&encoded)?;
+
+ assert_eq!(decoded.len(), 2);
+ assert_eq!(
+ decoded.files()[1].object_meta.location,
+ group.files()[1].object_meta.location
+ );
+ Ok(())
+ }
+}
diff --git a/datafusion/proto/Cargo.toml b/datafusion/proto/Cargo.toml
index cfff8a9494..037be27769 100644
--- a/datafusion/proto/Cargo.toml
+++ b/datafusion/proto/Cargo.toml
@@ -54,7 +54,7 @@ chrono = { workspace = true }
datafusion-catalog = { workspace = true }
datafusion-catalog-listing = { workspace = true }
datafusion-common = { workspace = true }
-datafusion-datasource = { workspace = true }
+datafusion-datasource = { workspace = true, features = ["proto"] }
datafusion-datasource-arrow = { workspace = true }
datafusion-datasource-avro = { workspace = true, optional = true }
datafusion-datasource-csv = { workspace = true }
diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs
index 7aa6376313..34ad8c7a62 100644
--- a/datafusion/proto/src/physical_plan/from_proto.rs
+++ b/datafusion/proto/src/physical_plan/from_proto.rs
@@ -23,7 +23,6 @@ use arrow::array::RecordBatch;
use arrow::compute::SortOptions;
use arrow::datatypes::{Field, Schema};
use arrow::ipc::reader::StreamReader;
-use chrono::{TimeZone, Utc};
use datafusion_common::{
DataFusionError, Result, ScalarValue, internal_datafusion_err, not_impl_err,
};
@@ -56,8 +55,6 @@ use datafusion_physical_plan::{
Partitioning, PhysicalExpr, RangePartitioning, SplitPoint, WindowExpr,
};
use datafusion_proto_common::common::proto_error;
-use object_store::ObjectMeta;
-use object_store::path::Path;
use super::{
DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalPlanDecodeContext,
@@ -632,64 +629,30 @@ pub fn parse_record_batches(buf: &[u8]) -> Result<Vec<RecordBatch>> {
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<Self, Self::Error> {
- let mut pf = PartitionedFile::new_from_meta(ObjectMeta {
- location: Path::parse(val.path.as_str())
- .map_err(|e| proto_error(format!("Invalid object_store path: {e}")))?,
- last_modified: Utc.timestamp_nanos(val.last_modified_ns as i64),
- size: val.size,
- e_tag: None,
- version: None,
- })
- .with_partition_values(
- val.partition_values
- .iter()
- .map(|v| v.try_into())
- .collect::<Result<Vec<_>, _>>()?,
- );
- if let Some(proto_schema) = val.arrow_schema.as_ref() {
- pf = pf.with_arrow_schema(Arc::new(
- proto_schema.try_into().map_err(DataFusionError::from)?,
- ));
- }
- if let Some(range) = val.range.as_ref() {
- let file_range = FileRange::try_from_proto(range)?;
- pf = pf.with_range(file_range.start, file_range.end);
- }
- if let Some(proto_stats) = val.statistics.as_ref() {
- // The wire format carries statistics for the full table schema (file + partition
- // columns), so assign directly — `with_statistics` would append the partition
- // column stats a second time.
- pf.statistics = Some(Arc::new(proto_stats.try_into()?));
- }
- Ok(pf)
+ 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<Self, Self::Error> {
- Ok(FileRange {
- start: value.start,
- end: value.end,
- })
+ 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<Self, Self::Error> {
- let files = val
- .files
- .iter()
- .map(PartitionedFile::try_from_proto)
- .collect::<Result<Vec<_>, _>>()?;
- Ok(FileGroup::new(files))
+ FileGroup::try_from(val)
}
}
@@ -734,7 +697,7 @@ impl TryFromProto<&protobuf::FileSinkConfig> for FileSinkConfig {
let file_group = FileGroup::new(
conf.file_groups
.iter()
- .map(PartitionedFile::try_from_proto)
+ .map(TryInto::try_into)
.collect::<Result<Vec<_>>>()?,
);
let table_paths = conf
@@ -812,6 +775,9 @@ impl datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprD
mod tests {
use super::*;
use arrow::datatypes::{DataType, Field, Schema};
+ use chrono::{TimeZone, Utc};
+ use object_store::ObjectMeta;
+ use object_store::path::Path;
#[test]
fn partitioned_file_path_roundtrip_percent_encoded() {
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs b/datafusion/proto/src/physical_plan/to_proto.rs
index e13923dbb9..5189972f0e 100644
--- a/datafusion/proto/src/physical_plan/to_proto.rs
+++ b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -424,51 +424,30 @@ fn serialize_range_split_point(
})
}
+/// 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<Self> {
- let last_modified = pf.object_meta.last_modified;
- let last_modified_ns = last_modified.timestamp_nanos_opt().ok_or_else(|| {
- DataFusionError::Plan(format!(
- "Invalid timestamp on PartitionedFile::ObjectMeta: {last_modified}"
- ))
- })? as u64;
- Ok(protobuf::PartitionedFile {
- arrow_schema: pf
- .arrow_schema
- .as_ref()
- .map(|s| s.as_ref().try_into())
- .transpose()?,
- path: pf.object_meta.location.as_ref().to_owned(),
- size: pf.object_meta.size,
- last_modified_ns,
- partition_values: pf
- .partition_values
- .iter()
- .map(|v| v.try_into())
- .collect::<Result<Vec<_>, _>>()?,
- range: pf
- .range
- .as_ref()
- .map(protobuf::FileRange::try_from_proto)
- .transpose()?,
- statistics: pf.statistics.as_ref().map(|s| s.as_ref().into()),
- })
+ 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<Self> {
- Ok(protobuf::FileRange {
- start: value.start,
- end: value.end,
- })
+ 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;
@@ -476,8 +455,8 @@ impl TryFromProto<&[PartitionedFile]> for protobuf::FileGroup {
Ok(protobuf::FileGroup {
files: gr
.iter()
- .map(protobuf::PartitionedFile::try_from_proto)
- .collect::<Result<Vec<_>, _>>()?,
+ .map(TryInto::try_into)
+ .collect::<Result<Vec<_>>>()?,
})
}
}
@@ -490,7 +469,7 @@ pub fn serialize_file_scan_config(
let file_groups = conf
.file_groups
.iter()
- .map(|p| protobuf::FileGroup::try_from_proto(p.files()))
+ .map(TryInto::try_into)
.collect::<Result<Vec<_>, _>>()?;
let mut output_orderings = vec![];
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org