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/05/22 02:32:47 UTC
(datafusion) branch gh-readonly-queue/main/pr-21929-f7d5575e51583f1caf06e19f00f340a3b7bb8d37 created (now 077f08a9a6)
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-21929-f7d5575e51583f1caf06e19f00f340a3b7bb8d37
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 077f08a9a6 Split proto serialization to encapsulate private state (#21835) (#21929)
This branch includes the following new commits:
new 077f08a9a6 Split proto serialization to encapsulate private state (#21835) (#21929)
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: Split proto serialization to encapsulate private state (#21835) (#21929)
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-21929-f7d5575e51583f1caf06e19f00f340a3b7bb8d37
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 077f08a9a6632324c95275dd15b5dd5b1f14006f
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Thu May 21 21:32:26 2026 -0500
Split proto serialization to encapsulate private state (#21835) (#21929)
## Which issue does this PR close?
- Closes #21835.
## Rationale for this change
`datafusion-proto` serializes every built-in `PhysicalExpr` through a
single
~300-line `downcast_ref` chain, with a mirror `match` on the decode
side. That
chain lives outside the crate where each expression is defined, so every
field
an expression wants to round-trip has to be made `pub`. #21807 is the
cautionary tale: it had to add five `pub` "proto-only, not stable" items
to
`DynamicFilterPhysicalExpr` just to serialize an `RwLock`-wrapped inner.
This PR adds the infrastructure so a `PhysicalExpr` can serialize itself
and
keep its state private.
## What changes are included in this PR?
A `PhysicalExpr` can now opt into serializing itself, in both
directions:
```rust
fn try_to_proto(&self, ctx: &PhysicalExprEncodeCtx) -> Result<Option<PhysicalExprNode>>
fn try_from_proto(node: &PhysicalExprNode, ctx: &PhysicalExprDecodeCtx) -> Result<Arc<dyn PhysicalExpr>>
```
`try_to_proto` returning `Ok(None)` (the default) means "fall through to
the
old downcast chain", so the change is purely additive — nothing is
forced to
migrate.
`Column` and `BinaryExpr` are migrated as working demos; everything
else stays on the old path and migrates later, one expression at a time,
with
no wire-format change.
Five stacked commits, each builds green on its own and is independently
reviewable (or splittable into its own PR):
1. **Extract `datafusion-proto-models` crate** — move the `.proto` file
and
prost-generated types into a lightweight crate (mirrors the existing
`datafusion-proto-common` split).
2. **Add the `try_to_proto` hook** — feature-gated, off by default.
3. **Migrate `Column` encode.**
4. **Add the decode side and migrate `Column` decode.**
5. **Migrate `BinaryExpr`** (both directions).
## A few design decisions worth flagging
- **`FromProto` / `TryFromProto` traits instead of plain `From` /
`TryFrom`.**
Once the prost types move into their own crate they are *foreign* to
`datafusion-proto`, and the orphan rule forbids `impl From<&protobuf::X>
for Y`
when both `X` and `Y` are foreign. So those conversions become
`FromProto` /
`TryFromProto` traits in `datafusion_proto::convert`, and callers go
from
`(&x).into()` to `Y::from_proto(&x)`. This is a known workaround, not
the end
state — see Future work.
- **The ctx is a concrete struct, not `&dyn`.** `PhysicalExprEncodeCtx`
/
`PhysicalExprDecodeCtx` wrap a sealed dispatch trait. Keeping them
concrete
keeps `&dyn` out of every expression's signature and gives a stable
place to
add helpers (UDF encoding, registry hooks) later without churning a
public
trait.
- **`try_from_proto` takes the whole `PhysicalExprNode`**, not the
pre-unwrapped variant payload, so every expression's decoder has the
same
signature and can still see outer-node fields like `expr_id`.
## Are these changes tested?
No new behavior, so no new tests. `Column` and `BinaryExpr` produce and
consume
the same wire format as before; the existing `roundtrip_physical_plan` /
`roundtrip_physical_expr` tests already cover both directions and now
exercise
the new path.
## Are there any user-facing changes?
Small API breaks in `datafusion-proto`:
- `try_from_physical_plan_with_converter` /
`try_into_physical_plan_with_converter`
move to a `PhysicalPlanNodeExt` trait — callers add
`use datafusion_proto::physical_plan::PhysicalPlanNodeExt;`.
- Foreign-foreign `From` / `TryFrom` conversions become `FromProto` /
`TryFromProto` (see Design decisions above).
- `datafusion_proto::generated::*` is deprecated in favor of
`datafusion_proto::protobuf`; it still works.
The new `proto` feature on `datafusion-physical-expr(-common)` is off by
default, so crates that don't serialize plans pay nothing.
## Future work
- Migrate the remaining built-in expressions — including
`DynamicFilterPhysicalExpr`, the original motivation — one per follow-up
PR.
- Apply the same pattern to `ExecutionPlan` serialization.
- Drop the `FromProto` / `TryFromProto` workaround: collapse
`datafusion-proto-common` into `datafusion-proto-models` and push the
conversion impls down to the target-type crates so callers use plain
`From` /
`TryFrom` again. Full dep-graph analysis and a step-by-step plan are in
[#21835
(comment)](https://github.com/apache/datafusion/issues/21835#issuecomment-4348350257).
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 4.7 (1M context) <no...@anthropic.com>
---
.github/workflows/rust.yml | 1 -
Cargo.lock | 23 +-
Cargo.toml | 4 +-
.../custom_data_source/adapter_serialization.rs | 3 +-
.../examples/proto/expression_deduplication.rs | 2 +-
datafusion/expr-common/src/operator.rs | 44 ++++
datafusion/physical-expr-common/Cargo.toml | 7 +
.../physical-expr-common/src/physical_expr.rs | 174 ++++++++++++++++
datafusion/physical-expr/Cargo.toml | 7 +
datafusion/physical-expr/src/expressions/binary.rs | 118 +++++++++++
datafusion/physical-expr/src/expressions/column.rs | 43 ++++
datafusion/{proto/gen => proto-models}/Cargo.toml | 34 +--
datafusion/proto-models/LICENSE.txt | 212 +++++++++++++++++++
datafusion/proto-models/NOTICE.txt | 5 +
datafusion/proto-models/README.md | 43 ++++
datafusion/{proto => proto-models}/gen/Cargo.toml | 0
datafusion/{proto => proto-models}/gen/src/main.rs | 6 +-
.../{proto => proto-models}/proto/datafusion.proto | 0
datafusion/{proto => proto-models}/regen.sh | 2 +-
.../src/generated/datafusion_proto_common.rs | 0
.../{proto => proto-models}/src/generated/mod.rs | 0
.../src/generated/pbjson.rs | 0
.../{proto => proto-models}/src/generated/prost.rs | 0
datafusion/proto-models/src/lib.rs | 54 +++++
datafusion/proto/Cargo.toml | 13 +-
datafusion/proto/regen.sh | 4 +-
datafusion/proto/src/common.rs | 18 ++
datafusion/proto/src/convert.rs | 44 ++++
datafusion/proto/src/generated/datafusion.rs | 1 -
datafusion/proto/src/lib.rs | 39 +++-
datafusion/proto/src/logical_plan/file_formats.rs | 49 ++---
datafusion/proto/src/logical_plan/from_proto.rs | 116 +++++------
datafusion/proto/src/logical_plan/mod.rs | 75 ++++---
datafusion/proto/src/logical_plan/to_proto.rs | 67 +++---
datafusion/proto/src/physical_plan/from_proto.rs | 153 +++++++-------
datafusion/proto/src/physical_plan/mod.rs | 231 ++++++++++++---------
datafusion/proto/src/physical_plan/to_proto.rs | 146 ++++++-------
.../proto/tests/cases/roundtrip_physical_plan.rs | 2 +-
dev/release/rat_exclude_files.txt | 6 +-
licenserc.toml | 1 +
40 files changed, 1302 insertions(+), 445 deletions(-)
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index 5ff1f6467b..bb2075ee01 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -605,7 +605,6 @@ jobs:
rust-version: stable
- name: Run
run: |
- echo '' > datafusion/proto/src/generated/datafusion.rs
ci/scripts/rust_fmt.sh
# Coverage job disabled due to
diff --git a/Cargo.lock b/Cargo.lock
index 4d5b15075e..63d57b2e69 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2344,6 +2344,7 @@ dependencies = [
"datafusion-functions",
"datafusion-functions-aggregate-common",
"datafusion-physical-expr-common",
+ "datafusion-proto-models",
"half",
"hashbrown 0.17.1",
"indexmap 2.14.0",
@@ -2379,6 +2380,7 @@ dependencies = [
"criterion",
"datafusion-common",
"datafusion-expr-common",
+ "datafusion-proto-models",
"hashbrown 0.17.1",
"indexmap 2.14.0",
"itertools 0.14.0",
@@ -2473,12 +2475,11 @@ dependencies = [
"datafusion-physical-expr-common",
"datafusion-physical-plan",
"datafusion-proto-common",
+ "datafusion-proto-models",
"doc-comment",
"object_store",
- "pbjson 0.9.0",
"pretty_assertions",
"prost",
- "serde",
"serde_json",
"tokio",
]
@@ -2495,6 +2496,16 @@ dependencies = [
"serde",
]
+[[package]]
+name = "datafusion-proto-models"
+version = "53.1.0"
+dependencies = [
+ "datafusion-proto-common",
+ "pbjson 0.9.0",
+ "prost",
+ "serde",
+]
+
[[package]]
name = "datafusion-pruning"
version = "53.1.0"
@@ -6201,9 +6212,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
[[package]]
name = "tokio"
-version = "1.52.3"
+version = "1.52.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8fc7f01b389ac15039e4dc9531aa973a135d7a4135281b12d7c1bc79fd57fffe"
+checksum = "110a78583f19d5cdb2c5ccf321d1290344e71313c6c37d43520d386027d18386"
dependencies = [
"bytes",
"libc",
@@ -6341,9 +6352,9 @@ checksum = "756daf9b1013ebe47a8776667b466417e2d4c5679d441c26230efd9ef78692db"
[[package]]
name = "tonic"
-version = "0.14.6"
+version = "0.14.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef"
+checksum = "fec7c61a0695dc1887c1b53952990f3ad2e3a31453e1f49f10e75424943a93ec"
dependencies = [
"async-trait",
"axum",
diff --git a/Cargo.toml b/Cargo.toml
index 78c271d524..2862a7a97b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -47,9 +47,10 @@ members = [
"datafusion/pruning",
"datafusion/physical-plan",
"datafusion/proto",
- "datafusion/proto/gen",
"datafusion/proto-common",
"datafusion/proto-common/gen",
+ "datafusion/proto-models",
+ "datafusion/proto-models/gen",
"datafusion/session",
"datafusion/spark",
"datafusion/sql",
@@ -152,6 +153,7 @@ datafusion-physical-optimizer = { path = "datafusion/physical-optimizer", versio
datafusion-physical-plan = { path = "datafusion/physical-plan", version = "53.1.0" }
datafusion-proto = { path = "datafusion/proto", version = "53.1.0" }
datafusion-proto-common = { path = "datafusion/proto-common", version = "53.1.0" }
+datafusion-proto-models = { path = "datafusion/proto-models", version = "53.1.0" }
datafusion-pruning = { path = "datafusion/pruning", version = "53.1.0" }
datafusion-session = { path = "datafusion/session", version = "53.1.0" }
datafusion-spark = { path = "datafusion/spark", version = "53.1.0" }
diff --git a/datafusion-examples/examples/custom_data_source/adapter_serialization.rs b/datafusion-examples/examples/custom_data_source/adapter_serialization.rs
index a956348279..f18b888f3e 100644
--- a/datafusion-examples/examples/custom_data_source/adapter_serialization.rs
+++ b/datafusion-examples/examples/custom_data_source/adapter_serialization.rs
@@ -62,7 +62,8 @@ use datafusion_proto::bytes::{
use datafusion_proto::physical_plan::from_proto::parse_physical_expr_with_converter;
use datafusion_proto::physical_plan::to_proto::serialize_physical_expr_with_converter;
use datafusion_proto::physical_plan::{
- PhysicalExtensionCodec, PhysicalPlanDecodeContext, PhysicalProtoConverterExtension,
+ PhysicalExtensionCodec, PhysicalPlanDecodeContext, PhysicalPlanNodeExt,
+ PhysicalProtoConverterExtension,
};
use datafusion_proto::protobuf::physical_plan_node::PhysicalPlanType;
use datafusion_proto::protobuf::{
diff --git a/datafusion-examples/examples/proto/expression_deduplication.rs b/datafusion-examples/examples/proto/expression_deduplication.rs
index ae4a76e79f..31bb234e28 100644
--- a/datafusion-examples/examples/proto/expression_deduplication.rs
+++ b/datafusion-examples/examples/proto/expression_deduplication.rs
@@ -52,7 +52,7 @@ use datafusion_proto::physical_plan::from_proto::parse_physical_expr_with_conver
use datafusion_proto::physical_plan::to_proto::serialize_physical_expr_with_converter;
use datafusion_proto::physical_plan::{
DefaultPhysicalExtensionCodec, PhysicalExtensionCodec, PhysicalPlanDecodeContext,
- PhysicalProtoConverterExtension,
+ PhysicalPlanNodeExt, PhysicalProtoConverterExtension,
};
use datafusion_proto::protobuf::{PhysicalExprNode, PhysicalPlanNode};
use prost::Message;
diff --git a/datafusion/expr-common/src/operator.rs b/datafusion/expr-common/src/operator.rs
index a078a27f2a..7b10d9bfae 100644
--- a/datafusion/expr-common/src/operator.rs
+++ b/datafusion/expr-common/src/operator.rs
@@ -390,6 +390,50 @@ impl Operator {
| Operator::StringConcat => false,
}
}
+
+ /// Parse an `Operator` from the string name `datafusion-proto` uses on the
+ /// wire (the `Debug` name of the variant, e.g. `"Eq"`).
+ ///
+ /// Returns `None` for names with no binary-operator counterpart. This is
+ /// the canonical proto-string mapping, shared by `datafusion-proto`
+ /// (logical plans) and `PhysicalExpr` decoders such as `BinaryExpr`, so the
+ /// mapping is not duplicated across crates.
+ pub fn from_proto_name(name: &str) -> Option<Operator> {
+ Some(match name {
+ "And" => Operator::And,
+ "Or" => Operator::Or,
+ "Eq" => Operator::Eq,
+ "NotEq" => Operator::NotEq,
+ "LtEq" => Operator::LtEq,
+ "Lt" => Operator::Lt,
+ "Gt" => Operator::Gt,
+ "GtEq" => Operator::GtEq,
+ "Plus" => Operator::Plus,
+ "Minus" => Operator::Minus,
+ "Multiply" => Operator::Multiply,
+ "Divide" => Operator::Divide,
+ "Modulo" => Operator::Modulo,
+ "IsDistinctFrom" => Operator::IsDistinctFrom,
+ "IsNotDistinctFrom" => Operator::IsNotDistinctFrom,
+ "BitwiseAnd" => Operator::BitwiseAnd,
+ "BitwiseOr" => Operator::BitwiseOr,
+ "BitwiseXor" => Operator::BitwiseXor,
+ "BitwiseShiftLeft" => Operator::BitwiseShiftLeft,
+ "BitwiseShiftRight" => Operator::BitwiseShiftRight,
+ "RegexIMatch" => Operator::RegexIMatch,
+ "RegexMatch" => Operator::RegexMatch,
+ "RegexNotIMatch" => Operator::RegexNotIMatch,
+ "RegexNotMatch" => Operator::RegexNotMatch,
+ "LikeMatch" => Operator::LikeMatch,
+ "ILikeMatch" => Operator::ILikeMatch,
+ "NotLikeMatch" => Operator::NotLikeMatch,
+ "NotILikeMatch" => Operator::NotILikeMatch,
+ "StringConcat" => Operator::StringConcat,
+ "AtArrow" => Operator::AtArrow,
+ "ArrowAt" => Operator::ArrowAt,
+ _ => return None,
+ })
+ }
}
impl fmt::Display for Operator {
diff --git a/datafusion/physical-expr-common/Cargo.toml b/datafusion/physical-expr-common/Cargo.toml
index 0e4748b81d..d1ee7feb29 100644
--- a/datafusion/physical-expr-common/Cargo.toml
+++ b/datafusion/physical-expr-common/Cargo.toml
@@ -40,11 +40,18 @@ workspace = true
[lib]
name = "datafusion_physical_expr_common"
+[features]
+default = []
+# Enables the `PhysicalExpr::to_proto` hook used by `datafusion-proto`.
+# Off by default so crates that never serialize plans pay nothing.
+proto = ["dep:datafusion-proto-models"]
+
[dependencies]
arrow = { workspace = true }
chrono = { workspace = true }
datafusion-common = { workspace = true }
datafusion-expr-common = { workspace = true }
+datafusion-proto-models = { workspace = true, optional = true }
hashbrown = { workspace = true }
indexmap = { workspace = true }
itertools = { workspace = true }
diff --git a/datafusion/physical-expr-common/src/physical_expr.rs b/datafusion/physical-expr-common/src/physical_expr.rs
index 7b3f7dcc76..887ed73745 100644
--- a/datafusion/physical-expr-common/src/physical_expr.rs
+++ b/datafusion/physical-expr-common/src/physical_expr.rs
@@ -477,6 +477,180 @@ pub trait PhysicalExpr: Any + Send + Sync + Display + Debug + DynEq + DynHash {
fn expression_id(&self) -> Option<u64> {
None
}
+
+ /// Serialize this expression to a [`PhysicalExprNode`] proto message.
+ ///
+ /// Returning `Ok(None)` means "this expression does not know how to
+ /// serialize itself"; the caller (typically `datafusion-proto`) will fall
+ /// back to its existing codec / extension paths. This matches today's
+ /// behavior for expressions that aren't built into `datafusion-proto`.
+ ///
+ /// Returning `Ok(Some(node))` means the expression has serialized itself
+ /// fully; the caller should not try any further fallback path.
+ ///
+ /// Returning `Err(_)` means a real serialization failure (e.g. the
+ /// expression knows it should serialize but a child failed).
+ ///
+ /// The motivating use case is letting expressions with private state
+ /// (e.g. `DynamicFilterPhysicalExpr`'s `RwLock`-protected inner fields)
+ /// reach into their own internals for `try_to_proto`/`try_from_proto`
+ /// without having to expose `pub` accessors to `datafusion-proto`. See
+ /// <https://github.com/apache/datafusion/issues/21835>.
+ ///
+ /// 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.
+ ///
+ /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
+ #[cfg(feature = "proto")]
+ fn try_to_proto(
+ &self,
+ _ctx: &proto_encode::PhysicalExprEncodeCtx<'_>,
+ ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
+ Ok(None)
+ }
+}
+
+/// Encode-side context for [`PhysicalExpr::try_to_proto`].
+///
+/// Expression authors only ever see [`proto_encode::PhysicalExprEncodeCtx`]:
+/// a concrete struct with stable methods. Internally it dispatches to a
+/// [`proto_encode::PhysicalExprEncode`] implementor that lives in
+/// `datafusion-proto`, which is what lets `physical-expr-common` stay free
+/// of `datafusion-proto` as a dep.
+///
+/// More specialized helpers (e.g. encoding UDFs/UDAFs/UDWFs through the
+/// extension codec) can be added to the context as expressions migrate;
+/// today they're not required because the encoder forwards to the existing
+/// codec via the proto converter.
+#[cfg(feature = "proto")]
+pub mod proto_encode {
+ use std::sync::Arc;
+
+ use datafusion_common::Result;
+ use datafusion_proto_models::protobuf::PhysicalExprNode;
+
+ use super::PhysicalExpr;
+
+ /// Encoder context handed to [`super::PhysicalExpr::try_to_proto`].
+ ///
+ /// Wraps an internal [`PhysicalExprEncode`] trait object so callers see a
+ /// stable concrete type while implementations can evolve in
+ /// `datafusion-proto`.
+ pub struct PhysicalExprEncodeCtx<'a> {
+ encoder: &'a dyn PhysicalExprEncode,
+ }
+
+ impl<'a> PhysicalExprEncodeCtx<'a> {
+ /// Construct a new encode context. Typically called by
+ /// `datafusion-proto`; expression authors receive `&PhysicalExprEncodeCtx`.
+ pub fn new(encoder: &'a dyn PhysicalExprEncode) -> Self {
+ Self { encoder }
+ }
+
+ /// Encode a child expression. Routes through the configured encoder
+ /// so dedup-aware encoding is preserved.
+ pub fn encode_child(
+ &self,
+ expr: &Arc<dyn PhysicalExpr>,
+ ) -> Result<PhysicalExprNode> {
+ self.encoder.encode(expr)
+ }
+ }
+
+ /// Internal dispatch trait. Implementors live in `datafusion-proto` and
+ /// wrap the existing `PhysicalExtensionCodec` +
+ /// `PhysicalProtoConverterExtension` plumbing. Expression authors should
+ /// use [`PhysicalExprEncodeCtx`] instead of calling this directly.
+ pub trait PhysicalExprEncode {
+ /// Encode an expression to a protobuf node.
+ fn encode(&self, expr: &Arc<dyn PhysicalExpr>) -> Result<PhysicalExprNode>;
+ }
+}
+
+/// Decode-side counterpart to [`proto_encode`].
+///
+/// Expression authors implement an associated `try_from_proto` on their
+/// concrete type, with the signature
+///
+/// ```ignore
+/// fn try_from_proto(
+/// node: &PhysicalExprNode,
+/// ctx: &PhysicalExprDecodeCtx<'_>,
+/// ) -> Result<Arc<dyn PhysicalExpr>>
+/// ```
+///
+/// It takes the whole [`PhysicalExprNode`] — the exact inverse of what
+/// [`PhysicalExpr::try_to_proto`] returns — so the constructor can also see
+/// outer-node fields such as `expr_id`. The central match in
+/// `datafusion-proto` dispatches `ExprType` variants to these constructors.
+///
+/// As with the encode side, the public surface is a struct (not a `&dyn`
+/// trait) so future fields/helpers (registries for third-party expressions,
+/// schema-resolution caches, etc.) can be added without changing the
+/// signature every expression depends on.
+///
+/// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
+#[cfg(feature = "proto")]
+pub mod proto_decode {
+ use std::sync::Arc;
+
+ use arrow::datatypes::Schema;
+ use datafusion_common::Result;
+ use datafusion_proto_models::protobuf::PhysicalExprNode;
+
+ use super::PhysicalExpr;
+
+ /// Decoder context handed to per-expression `try_from_proto` constructors.
+ ///
+ /// Wraps an internal [`PhysicalExprDecode`] trait object plus a borrowed
+ /// schema. The trait stays an implementation detail of `datafusion-proto`;
+ /// expression authors only see this struct.
+ pub struct PhysicalExprDecodeCtx<'a> {
+ schema: &'a Schema,
+ decoder: &'a dyn PhysicalExprDecode,
+ }
+
+ impl<'a> PhysicalExprDecodeCtx<'a> {
+ /// Construct a new decode context. Typically called by
+ /// `datafusion-proto`; expression authors receive
+ /// `&PhysicalExprDecodeCtx`.
+ pub fn new(schema: &'a Schema, decoder: &'a dyn PhysicalExprDecode) -> Self {
+ Self { schema, decoder }
+ }
+
+ /// The schema bound to this decode context. Use it for column lookups,
+ /// data-type resolution, etc.
+ pub fn schema(&self) -> &Schema {
+ self.schema
+ }
+
+ /// Decode an expression node, recursing into child sub-expressions.
+ ///
+ /// Routes built-in `ExprType` variants through `datafusion-proto`'s
+ /// central match and forwards extension nodes to the registered codec
+ /// (today via [`PhysicalExtensionCodec::try_decode_expr`]; later via
+ /// a per-type registry — see #21835).
+ ///
+ /// [`PhysicalExtensionCodec::try_decode_expr`]: https://docs.rs/datafusion-proto/latest/datafusion_proto/physical_plan/trait.PhysicalExtensionCodec.html#method.try_decode_expr
+ pub fn decode(&self, node: &PhysicalExprNode) -> Result<Arc<dyn PhysicalExpr>> {
+ self.decoder.decode(node, self.schema)
+ }
+ }
+
+ /// Internal dispatch trait. Implementors live in `datafusion-proto`.
+ /// Expression authors should use [`PhysicalExprDecodeCtx`] instead of
+ /// calling this directly.
+ pub trait PhysicalExprDecode {
+ /// Decode a proto node into a concrete `PhysicalExpr`. The schema is
+ /// passed alongside so implementations can support recursive children
+ /// and rebind the context per call (e.g. for nested plans).
+ fn decode(
+ &self,
+ node: &PhysicalExprNode,
+ schema: &Schema,
+ ) -> Result<Arc<dyn PhysicalExpr>>;
+ }
}
#[deprecated(
diff --git a/datafusion/physical-expr/Cargo.toml b/datafusion/physical-expr/Cargo.toml
index b755353d75..65ef2a3ceb 100644
--- a/datafusion/physical-expr/Cargo.toml
+++ b/datafusion/physical-expr/Cargo.toml
@@ -42,6 +42,12 @@ name = "datafusion_physical_expr"
[features]
recursive_protection = ["dep:recursive"]
+# Forwards the `proto` feature to `datafusion-physical-expr-common`, exposing
+# `PhysicalExpr::to_proto` and letting expressions in this crate implement it.
+proto = [
+ "dep:datafusion-proto-models",
+ "datafusion-physical-expr-common/proto",
+]
[dependencies]
arrow = { workspace = true }
@@ -50,6 +56,7 @@ datafusion-expr = { workspace = true }
datafusion-expr-common = { workspace = true }
datafusion-functions-aggregate-common = { workspace = true }
datafusion-physical-expr-common = { workspace = true }
+datafusion-proto-models = { workspace = true, optional = true }
hashbrown = { workspace = true }
indexmap = { workspace = true }
itertools = { workspace = true, features = ["use_std"] }
diff --git a/datafusion/physical-expr/src/expressions/binary.rs b/datafusion/physical-expr/src/expressions/binary.rs
index b92668fe9b..712f8f58f3 100644
--- a/datafusion/physical-expr/src/expressions/binary.rs
+++ b/datafusion/physical-expr/src/expressions/binary.rs
@@ -610,6 +610,124 @@ impl PhysicalExpr for BinaryExpr {
write!(f, " {} ", self.op)?;
write_child(f, self.right.as_ref(), precedence)
}
+
+ #[cfg(feature = "proto")]
+ fn try_to_proto(
+ &self,
+ ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
+ ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
+ use datafusion_proto_models::protobuf;
+
+ // Linearize a nested binary expression tree of the same operator
+ // into a flat vector of operands to avoid deep recursion in proto.
+ let op = self.op;
+ let mut operand_refs: Vec<&Arc<dyn PhysicalExpr>> = vec![&self.right];
+ let mut current_expr: &BinaryExpr = self;
+ loop {
+ match current_expr.left.downcast_ref::<BinaryExpr>() {
+ Some(bin) if bin.op == op => {
+ operand_refs.push(&bin.right);
+ current_expr = bin;
+ }
+ _ => {
+ operand_refs.push(¤t_expr.left);
+ break;
+ }
+ }
+ }
+ // Reverse so operands are ordered from left innermost to right outermost.
+ operand_refs.reverse();
+
+ let operands = operand_refs
+ .iter()
+ .map(|e| ctx.encode_child(e))
+ .collect::<Result<Vec<_>>>()?;
+
+ Ok(Some(protobuf::PhysicalExprNode {
+ expr_id: None,
+ expr_type: Some(protobuf::physical_expr_node::ExprType::BinaryExpr(
+ Box::new(protobuf::PhysicalBinaryExprNode {
+ l: None,
+ r: None,
+ op: format!("{op:?}"),
+ operands,
+ }),
+ )),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+impl BinaryExpr {
+ /// Reconstruct a [`BinaryExpr`] (or a left-deep tree of them when the proto
+ /// uses the linearized `operands` form) from its protobuf representation.
+ ///
+ /// Takes the whole [`PhysicalExprNode`] — the exact inverse of what
+ /// [`PhysicalExpr::try_to_proto`] produces — so every expression's
+ /// `try_from_proto` shares one signature. The operator string is parsed
+ /// via the canonical [`Operator::from_proto_name`] mapping, so no `op`
+ /// argument needs to be threaded in by the caller.
+ ///
+ /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
+ /// [`PhysicalExpr::try_to_proto`]: datafusion_physical_expr_common::physical_expr::PhysicalExpr::try_to_proto
+ /// [`PhysicalExprDecodeCtx::decode`]: datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx::decode
+ pub fn try_from_proto(
+ node: &datafusion_proto_models::protobuf::PhysicalExprNode,
+ ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
+ ) -> Result<Arc<dyn PhysicalExpr>> {
+ use datafusion_proto_models::protobuf;
+ let node = match &node.expr_type {
+ Some(protobuf::physical_expr_node::ExprType::BinaryExpr(b)) => b.as_ref(),
+ _ => return internal_err!("PhysicalExprNode is not a BinaryExpr"),
+ };
+ let op = Operator::from_proto_name(&node.op).ok_or_else(|| {
+ datafusion_common::DataFusionError::Internal(format!(
+ "Unsupported binary operator '{}'",
+ node.op
+ ))
+ })?;
+
+ if !node.operands.is_empty() {
+ // New linearized format: reduce the flat operands list back into
+ // a nested binary expression tree.
+ let operands = node
+ .operands
+ .iter()
+ .map(|e| ctx.decode(e))
+ .collect::<Result<Vec<_>>>()?;
+
+ if operands.len() < 2 {
+ return Err(datafusion_common::DataFusionError::Internal(
+ "A binary expression must always have at least 2 operands"
+ .to_string(),
+ ));
+ }
+
+ Ok(operands
+ .into_iter()
+ .reduce(|left, right| {
+ Arc::new(BinaryExpr::new(left, op, right)) as Arc<dyn PhysicalExpr>
+ })
+ .expect("Binary expression could not be reduced to a single expression."))
+ } else {
+ // Legacy format with l/r fields.
+ let left = node.l.as_deref().ok_or_else(|| {
+ datafusion_common::DataFusionError::Internal(
+ "BinaryExpr is missing required field 'left'".to_string(),
+ )
+ })?;
+ let right = node.r.as_deref().ok_or_else(|| {
+ datafusion_common::DataFusionError::Internal(
+ "BinaryExpr is missing required field 'right'".to_string(),
+ )
+ })?;
+ Ok(Arc::new(BinaryExpr::new(
+ ctx.decode(left)?,
+ op,
+ ctx.decode(right)?,
+ )))
+ }
+ }
}
/// Casts dictionary array to result type for binary numerical operators. Such operators
diff --git a/datafusion/physical-expr/src/expressions/column.rs b/datafusion/physical-expr/src/expressions/column.rs
index 7d4b0e7e2f..2b1de870e7 100644
--- a/datafusion/physical-expr/src/expressions/column.rs
+++ b/datafusion/physical-expr/src/expressions/column.rs
@@ -146,6 +146,49 @@ impl PhysicalExpr for Column {
fn placement(&self) -> ExpressionPlacement {
ExpressionPlacement::Column
}
+
+ #[cfg(feature = "proto")]
+ fn try_to_proto(
+ &self,
+ _ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
+ ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
+ 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,
+ },
+ )),
+ }))
+ }
+}
+
+#[cfg(feature = "proto")]
+impl Column {
+ /// Reconstruct a [`Column`] from its protobuf representation.
+ ///
+ /// Takes the whole [`PhysicalExprNode`] — the exact inverse of what
+ /// [`PhysicalExpr::try_to_proto`] produces — so every expression's
+ /// `try_from_proto` shares one signature. The decode context is currently
+ /// unused, but is threaded through so that future expressions with child
+ /// sub-expressions can recurse via [`PhysicalExprDecodeCtx::decode`].
+ ///
+ /// [`PhysicalExprNode`]: datafusion_proto_models::protobuf::PhysicalExprNode
+ /// [`PhysicalExpr::try_to_proto`]: datafusion_physical_expr_common::physical_expr::PhysicalExpr::try_to_proto
+ /// [`PhysicalExprDecodeCtx::decode`]: datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx::decode
+ pub fn try_from_proto(
+ node: &datafusion_proto_models::protobuf::PhysicalExprNode,
+ _ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
+ ) -> Result<Arc<dyn PhysicalExpr>> {
+ use datafusion_proto_models::protobuf;
+ let protobuf::PhysicalColumn { name, index } = match &node.expr_type {
+ Some(protobuf::physical_expr_node::ExprType::Column(c)) => c,
+ _ => return internal_err!("PhysicalExprNode is not a Column"),
+ };
+ Ok(Arc::new(Column::new(name, *index as usize)))
+ }
}
impl Column {
diff --git a/datafusion/proto/gen/Cargo.toml b/datafusion/proto-models/Cargo.toml
similarity index 64%
copy from datafusion/proto/gen/Cargo.toml
copy to datafusion/proto-models/Cargo.toml
index 8b48dfe70e..e37c4a2dba 100644
--- a/datafusion/proto/gen/Cargo.toml
+++ b/datafusion/proto-models/Cargo.toml
@@ -16,26 +16,30 @@
# under the License.
[package]
-name = "gen"
-description = "Code generation for proto"
-version = "0.1.0"
+name = "datafusion-proto-models"
+description = "Protobuf-generated model types for DataFusion logical and physical plans"
+keywords = ["arrow", "query", "sql"]
+readme = "README.md"
+version = { workspace = true }
edition = { workspace = true }
-rust-version = { workspace = true }
-authors = { workspace = true }
homepage = { workspace = true }
repository = { workspace = true }
license = { workspace = true }
-publish = false
+authors = { workspace = true }
+rust-version = { workspace = true }
+
+[package.metadata.docs.rs]
+all-features = true
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[lib]
+name = "datafusion_proto_models"
-# Note: add additional linter rules in lib.rs.
-# Rust does not support workspace + new linter rules in subcrates yet
-# https://github.com/rust-lang/cargo/issues/13157
-[lints]
-workspace = true
+[features]
+default = []
+json = ["serde", "pbjson", "datafusion-proto-common/json"]
[dependencies]
-# Pin these dependencies so that the generated output is deterministic
-pbjson-build = "=0.9.0"
-prost-build = "=0.14.3"
+datafusion-proto-common = { workspace = true }
+pbjson = { workspace = true, optional = true }
+prost = { workspace = true }
+serde = { version = "1.0", optional = true }
diff --git a/datafusion/proto-models/LICENSE.txt b/datafusion/proto-models/LICENSE.txt
new file mode 100644
index 0000000000..d74c6b599d
--- /dev/null
+++ b/datafusion/proto-models/LICENSE.txt
@@ -0,0 +1,212 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ Licensed 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.
+
+
+This project includes code from Apache Aurora.
+
+* dev/release/{release,changelog,release-candidate} are based on the scripts from
+ Apache Aurora
+
+Copyright: 2016 The Apache Software Foundation.
+Home page: https://aurora.apache.org/
+License: http://www.apache.org/licenses/LICENSE-2.0
diff --git a/datafusion/proto-models/NOTICE.txt b/datafusion/proto-models/NOTICE.txt
new file mode 100644
index 0000000000..0bd2d52368
--- /dev/null
+++ b/datafusion/proto-models/NOTICE.txt
@@ -0,0 +1,5 @@
+Apache DataFusion
+Copyright 2019-2026 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
diff --git a/datafusion/proto-models/README.md b/datafusion/proto-models/README.md
new file mode 100644
index 0000000000..34adae9c1a
--- /dev/null
+++ b/datafusion/proto-models/README.md
@@ -0,0 +1,43 @@
+<!---
+ 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.
+-->
+
+# Apache DataFusion Protobuf Models
+
+[Apache DataFusion] is an extensible query execution framework, written in Rust, that uses [Apache Arrow] as its in-memory format.
+
+This crate contains the [prost]-generated Rust types for DataFusion's logical
+and physical plan protobuf schemas. It is intentionally kept narrow: it has no
+DataFusion dependencies beyond [`datafusion-proto-common`] and exposes only the
+generated structs (and optional [pbjson]/[serde] support).
+
+This crate is consumed by [`datafusion-proto`] and may also be depended on
+directly by other DataFusion crates that need to refer to the proto schema
+types without pulling in the full [`datafusion-proto`] surface.
+
+Most projects should use the [`datafusion-proto`] crate directly, which
+re-exports this module. If you are already using the [`datafusion-proto`]
+crate, there is no reason to use this crate directly in your project as well.
+
+[apache arrow]: https://arrow.apache.org/
+[apache datafusion]: https://datafusion.apache.org/
+[prost]: https://docs.rs/prost/latest/prost/
+[pbjson]: https://docs.rs/pbjson/latest/pbjson/
+[serde]: https://serde.rs/
+[`datafusion-proto`]: https://crates.io/crates/datafusion-proto
+[`datafusion-proto-common`]: https://crates.io/crates/datafusion-proto-common
diff --git a/datafusion/proto/gen/Cargo.toml b/datafusion/proto-models/gen/Cargo.toml
similarity index 100%
rename from datafusion/proto/gen/Cargo.toml
rename to datafusion/proto-models/gen/Cargo.toml
diff --git a/datafusion/proto/gen/src/main.rs b/datafusion/proto-models/gen/src/main.rs
similarity index 92%
rename from datafusion/proto/gen/src/main.rs
rename to datafusion/proto-models/gen/src/main.rs
index 7f16316203..4da674c43c 100644
--- a/datafusion/proto/gen/src/main.rs
+++ b/datafusion/proto-models/gen/src/main.rs
@@ -18,9 +18,9 @@
use std::path::Path;
fn main() -> Result<(), String> {
- let proto_dir = Path::new("datafusion/proto");
- let proto_path = Path::new("datafusion/proto/proto/datafusion.proto");
- let out_dir = Path::new("datafusion/proto/src");
+ let proto_dir = Path::new("datafusion/proto-models");
+ let proto_path = Path::new("datafusion/proto-models/proto/datafusion.proto");
+ let out_dir = Path::new("datafusion/proto-models/src");
// proto definitions has to be there
let descriptor_path = proto_dir.join("proto/proto_descriptor.bin");
diff --git a/datafusion/proto/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto
similarity index 100%
rename from datafusion/proto/proto/datafusion.proto
rename to datafusion/proto-models/proto/datafusion.proto
diff --git a/datafusion/proto/regen.sh b/datafusion/proto-models/regen.sh
similarity index 91%
copy from datafusion/proto/regen.sh
copy to datafusion/proto-models/regen.sh
index 02970a90ad..4bb07a1f32 100755
--- a/datafusion/proto/regen.sh
+++ b/datafusion/proto-models/regen.sh
@@ -18,4 +18,4 @@
# under the License.
repo_root=$(git rev-parse --show-toplevel)
-cd "$repo_root" && cargo run --manifest-path datafusion/proto/gen/Cargo.toml
+cd "$repo_root" && cargo run --manifest-path datafusion/proto-models/gen/Cargo.toml
diff --git a/datafusion/proto/src/generated/datafusion_proto_common.rs b/datafusion/proto-models/src/generated/datafusion_proto_common.rs
similarity index 100%
rename from datafusion/proto/src/generated/datafusion_proto_common.rs
rename to datafusion/proto-models/src/generated/datafusion_proto_common.rs
diff --git a/datafusion/proto/src/generated/mod.rs b/datafusion/proto-models/src/generated/mod.rs
similarity index 100%
rename from datafusion/proto/src/generated/mod.rs
rename to datafusion/proto-models/src/generated/mod.rs
diff --git a/datafusion/proto/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs
similarity index 100%
rename from datafusion/proto/src/generated/pbjson.rs
rename to datafusion/proto-models/src/generated/pbjson.rs
diff --git a/datafusion/proto/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs
similarity index 100%
rename from datafusion/proto/src/generated/prost.rs
rename to datafusion/proto-models/src/generated/prost.rs
diff --git a/datafusion/proto-models/src/lib.rs b/datafusion/proto-models/src/lib.rs
new file mode 100644
index 0000000000..8f845a8a99
--- /dev/null
+++ b/datafusion/proto-models/src/lib.rs
@@ -0,0 +1,54 @@
+// 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.
+
+#![doc(
+ html_logo_url = "https://raw.githubusercontent.com/apache/datafusion/19fe44cf2f30cbdd63d4a4f52c74055163c6cc38/docs/logos/standalone_logo/logo_original.svg",
+ html_favicon_url = "https://raw.githubusercontent.com/apache/datafusion/19fe44cf2f30cbdd63d4a4f52c74055163c6cc38/docs/logos/standalone_logo/logo_original.svg"
+)]
+#![cfg_attr(docsrs, feature(doc_cfg))]
+// Make sure fast / cheap clones on Arc are explicit:
+// https://github.com/apache/datafusion/issues/11143
+#![deny(clippy::clone_on_ref_ptr)]
+
+//! `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`].
+//!
+//! Most users should depend on [`datafusion-proto`] instead, which re-exports
+//! these types under [`datafusion_proto::protobuf`].
+//!
+//! [`datafusion-proto`]: https://crates.io/crates/datafusion-proto
+//! [`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 generated;
+
+/// All DataFusion protobuf model types.
+///
+/// Includes both the types declared in `datafusion.proto` and the
+/// `datafusion_proto_common` types it imports, in a single flat namespace
+/// so consumers can `use datafusion_proto_models::protobuf::*;`.
+pub mod protobuf {
+ pub use crate::generated::datafusion::*;
+}
+
+/// Re-export of the `datafusion_proto_common` types as exposed through this
+/// crate's generated module, for callers that want the common-only namespace.
+pub use generated::datafusion_common;
diff --git a/datafusion/proto/Cargo.toml b/datafusion/proto/Cargo.toml
index 4484846813..53e7cd78dc 100644
--- a/datafusion/proto/Cargo.toml
+++ b/datafusion/proto/Cargo.toml
@@ -36,7 +36,11 @@ name = "datafusion_proto"
[features]
default = ["parquet"]
-json = ["pbjson", "serde", "serde_json", "datafusion-proto-common/json"]
+json = [
+ "serde_json",
+ "datafusion-proto-common/json",
+ "datafusion-proto-models/json",
+]
parquet = ["datafusion-datasource-parquet", "datafusion-common/parquet", "datafusion/parquet"]
avro = ["datafusion-datasource-avro"]
@@ -59,14 +63,13 @@ datafusion-datasource-parquet = { workspace = true, optional = true }
datafusion-execution = { workspace = true }
datafusion-expr = { workspace = true }
datafusion-functions-table = { workspace = true }
-datafusion-physical-expr = { workspace = true }
-datafusion-physical-expr-common = { workspace = true }
+datafusion-physical-expr = { workspace = true, features = ["proto"] }
+datafusion-physical-expr-common = { workspace = true, features = ["proto"] }
datafusion-physical-plan = { workspace = true }
datafusion-proto-common = { workspace = true }
+datafusion-proto-models = { workspace = true }
object_store = { workspace = true }
-pbjson = { workspace = true, optional = true }
prost = { workspace = true }
-serde = { version = "1.0", optional = true }
serde_json = { workspace = true, optional = true }
[dev-dependencies]
diff --git a/datafusion/proto/regen.sh b/datafusion/proto/regen.sh
index 02970a90ad..c4bcea9ff5 100755
--- a/datafusion/proto/regen.sh
+++ b/datafusion/proto/regen.sh
@@ -17,5 +17,7 @@
# specific language governing permissions and limitations
# under the License.
+# The proto schema and code generation now live in `datafusion-proto-models`.
+# This script is kept as a convenience wrapper.
repo_root=$(git rev-parse --show-toplevel)
-cd "$repo_root" && cargo run --manifest-path datafusion/proto/gen/Cargo.toml
+exec "$repo_root/datafusion/proto-models/regen.sh"
diff --git a/datafusion/proto/src/common.rs b/datafusion/proto/src/common.rs
index 22ded708d8..bff017edbc 100644
--- a/datafusion/proto/src/common.rs
+++ b/datafusion/proto/src/common.rs
@@ -47,6 +47,24 @@ 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
new file mode 100644
index 0000000000..cb5c5bd7f8
--- /dev/null
+++ b/datafusion/proto/src/convert.rs
@@ -0,0 +1,44 @@
+// 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<T>: 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<T>: Sized {
+ type Error;
+ fn try_from_proto(value: T) -> std::result::Result<Self, Self::Error>;
+}
diff --git a/datafusion/proto/src/generated/datafusion.rs b/datafusion/proto/src/generated/datafusion.rs
deleted file mode 100644
index 8b13789179..0000000000
--- a/datafusion/proto/src/generated/datafusion.rs
+++ /dev/null
@@ -1 +0,0 @@
-
diff --git a/datafusion/proto/src/lib.rs b/datafusion/proto/src/lib.rs
index 7ddc930fa2..0e63bcf5f5 100644
--- a/datafusion/proto/src/lib.rs
+++ b/datafusion/proto/src/lib.rs
@@ -123,12 +123,13 @@
//! ```
pub mod bytes;
pub mod common;
-pub mod generated;
+pub mod convert;
pub mod logical_plan;
pub mod physical_plan;
+pub use convert::{FromProto, TryFromProto};
+
pub mod protobuf {
- pub use crate::generated::datafusion::*;
pub use datafusion_proto_common::common::proto_error;
pub use datafusion_proto_common::protobuf_common::{
ArrowFormat, ArrowOptions, ArrowType, AvroFormat, AvroOptions, CsvFormat,
@@ -136,6 +137,40 @@ pub mod protobuf {
ScalarValue, Schema,
};
pub use datafusion_proto_common::{FromProtoError, ToProtoError};
+ // Re-export every type from `datafusion-proto-models`'s generated module
+ // so the existing `datafusion_proto::protobuf::Foo` paths keep resolving.
+ // Going through the deeper `generated::datafusion` path (rather than
+ // `datafusion_proto_models::protobuf`, which is itself a `pub use ::*`)
+ // avoids a double wildcard re-export that some tools (cargo-semver-checks)
+ // don't follow.
+ pub use datafusion_proto_models::generated::datafusion::*;
+}
+
+/// Backwards-compatible re-export of the moved generated types.
+///
+/// The prost-generated structs now live in `datafusion-proto-models`;
+/// this module preserves the legacy `datafusion_proto::generated::*` paths
+/// for downstream callers. Prefer the [`protobuf`] module (or
+/// [`datafusion_proto_models`] directly) in new code.
+#[deprecated(
+ since = "53.1.0",
+ note = "use `datafusion_proto::protobuf` (or `datafusion_proto_models::protobuf`) instead"
+)]
+pub mod generated {
+ /// Re-export of the prost-generated types defined in `datafusion.proto`.
+ #[deprecated(
+ since = "53.1.0",
+ note = "use `datafusion_proto::protobuf` (or `datafusion_proto_models::protobuf`) instead"
+ )]
+ pub use datafusion_proto_models::generated::datafusion;
+
+ /// Re-export of the prost-generated common types defined in
+ /// `datafusion_common.proto`.
+ #[deprecated(
+ since = "53.1.0",
+ note = "use `datafusion_proto_common::protobuf_common` instead"
+ )]
+ pub use datafusion_proto_models::generated::datafusion_common;
}
#[cfg(doctest)]
diff --git a/datafusion/proto/src/logical_plan/file_formats.rs b/datafusion/proto/src/logical_plan/file_formats.rs
index 683b6a612a..d5af7be485 100644
--- a/datafusion/proto/src/logical_plan/file_formats.rs
+++ b/datafusion/proto/src/logical_plan/file_formats.rs
@@ -18,6 +18,7 @@
use std::sync::Arc;
use super::LogicalExtensionCodec;
+use crate::convert::FromProto;
use crate::protobuf::{
CsvOptions as CsvOptionsProto, CsvQuoteStyle as CsvQuoteStyleProto,
JsonOptions as JsonOptionsProto,
@@ -37,8 +38,8 @@ use prost::Message;
#[derive(Debug)]
pub struct CsvLogicalExtensionCodec;
-impl CsvOptionsProto {
- fn from_factory(factory: &CsvFormatFactory) -> Self {
+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]),
@@ -79,8 +80,8 @@ impl CsvOptionsProto {
}
}
-impl From<&CsvOptionsProto> for CsvOptions {
- fn from(proto: &CsvOptionsProto) -> Self {
+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)
@@ -230,7 +231,7 @@ impl LogicalExtensionCodec for CsvLogicalExtensionCodec {
let proto = CsvOptionsProto::decode(buf).map_err(|e| {
exec_datafusion_err!("Failed to decode CsvOptionsProto: {e:?}")
})?;
- let options: CsvOptions = (&proto).into();
+ let options = CsvOptions::from_proto(&proto);
Ok(Arc::new(CsvFormatFactory {
options: Some(options),
}))
@@ -247,7 +248,7 @@ impl LogicalExtensionCodec for CsvLogicalExtensionCodec {
return exec_err!("{}", "Unsupported FileFormatFactory type".to_string());
};
- let proto = CsvOptionsProto::from_factory(&CsvFormatFactory {
+ let proto = CsvOptionsProto::from_proto(&CsvFormatFactory {
options: Some(options),
});
@@ -259,8 +260,8 @@ impl LogicalExtensionCodec for CsvLogicalExtensionCodec {
}
}
-impl JsonOptionsProto {
- fn from_factory(factory: &JsonFormatFactory) -> Self {
+impl FromProto<&JsonFormatFactory> for JsonOptionsProto {
+ fn from_proto(factory: &JsonFormatFactory) -> Self {
if let Some(options) = &factory.options {
JsonOptionsProto {
compression: options.compression as i32,
@@ -274,8 +275,8 @@ impl JsonOptionsProto {
}
}
-impl From<&JsonOptionsProto> for JsonOptions {
- fn from(proto: &JsonOptionsProto) -> Self {
+impl FromProto<&JsonOptionsProto> for JsonOptions {
+ fn from_proto(proto: &JsonOptionsProto) -> Self {
JsonOptions {
compression: match proto.compression {
0 => CompressionTypeVariant::GZIP,
@@ -340,7 +341,7 @@ impl LogicalExtensionCodec for JsonLogicalExtensionCodec {
let proto = JsonOptionsProto::decode(buf).map_err(|e| {
exec_datafusion_err!("Failed to decode JsonOptionsProto: {e:?}")
})?;
- let options: JsonOptions = (&proto).into();
+ let options = JsonOptions::from_proto(&proto);
Ok(Arc::new(JsonFormatFactory {
options: Some(options),
}))
@@ -358,7 +359,7 @@ impl LogicalExtensionCodec for JsonLogicalExtensionCodec {
return exec_err!("Unsupported FileFormatFactory type");
};
- let proto = JsonOptionsProto::from_factory(&JsonFormatFactory {
+ let proto = JsonOptionsProto::from_proto(&JsonFormatFactory {
options: Some(options),
});
@@ -385,8 +386,8 @@ mod parquet {
};
use datafusion_datasource_parquet::file_format::ParquetFormatFactory;
- impl TableParquetOptionsProto {
- fn from_factory(factory: &ParquetFormatFactory) -> Self {
+ impl FromProto<&ParquetFormatFactory> for TableParquetOptionsProto {
+ fn from_proto(factory: &ParquetFormatFactory) -> Self {
let global_options = if let Some(ref options) = factory.options {
options.clone()
} else {
@@ -499,8 +500,8 @@ mod parquet {
}
}
- impl From<&ParquetOptionsProto> for ParquetOptions {
- fn from(proto: &ParquetOptionsProto) -> Self {
+ impl FromProto<&ParquetOptionsProto> for ParquetOptions {
+ fn from_proto(proto: &ParquetOptionsProto) -> Self {
ParquetOptions {
enable_page_index: proto.enable_page_index,
pruning: proto.pruning,
@@ -577,8 +578,8 @@ mod parquet {
}
}
- impl From<ParquetColumnOptionsProto> for ParquetColumnOptions {
- fn from(proto: ParquetColumnOptionsProto) -> Self {
+ impl FromProto<ParquetColumnOptionsProto> 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,
@@ -605,13 +606,13 @@ mod parquet {
}
}
- impl From<&TableParquetOptionsProto> for TableParquetOptions {
- fn from(proto: &TableParquetOptionsProto) -> Self {
+ impl FromProto<&TableParquetOptionsProto> for TableParquetOptions {
+ fn from_proto(proto: &TableParquetOptionsProto) -> Self {
TableParquetOptions {
global: proto
.global
.as_ref()
- .map(ParquetOptions::from)
+ .map(ParquetOptions::from_proto)
.unwrap_or_default(),
column_specific_options: proto
.column_specific_options
@@ -619,7 +620,7 @@ mod parquet {
.map(|parquet_column_options| {
(
parquet_column_options.column_name.clone(),
- ParquetColumnOptions::from(
+ ParquetColumnOptions::from_proto(
parquet_column_options
.options
.clone()
@@ -688,7 +689,7 @@ mod parquet {
let proto = TableParquetOptionsProto::decode(buf).map_err(|e| {
exec_datafusion_err!("Failed to decode TableParquetOptionsProto: {e:?}")
})?;
- let options: TableParquetOptions = (&proto).into();
+ let options = TableParquetOptions::from_proto(&proto);
Ok(Arc::new(
datafusion_datasource_parquet::file_format::ParquetFormatFactory {
options: Some(options),
@@ -711,7 +712,7 @@ mod parquet {
return exec_err!("Unsupported FileFormatFactory type");
};
- let proto = TableParquetOptionsProto::from_factory(&ParquetFormatFactory {
+ let proto = TableParquetOptionsProto::from_proto(&ParquetFormatFactory {
options: Some(options),
});
diff --git a/datafusion/proto/src/logical_plan/from_proto.rs b/datafusion/proto/src/logical_plan/from_proto.rs
index 78ffd362c8..c68b83964f 100644
--- a/datafusion/proto/src/logical_plan/from_proto.rs
+++ b/datafusion/proto/src/logical_plan/from_proto.rs
@@ -54,10 +54,12 @@ use crate::protobuf::{
},
};
+use crate::convert::{FromProto, TryFromProto};
+
use super::{AsLogicalPlan, LogicalExtensionCodec};
-impl From<&protobuf::UnnestOptions> for UnnestOptions {
- fn from(opts: &protobuf::UnnestOptions) -> Self {
+impl FromProto<&protobuf::UnnestOptions> for UnnestOptions {
+ fn from_proto(opts: &protobuf::UnnestOptions) -> Self {
Self {
preserve_nulls: opts.preserve_nulls,
recursions: opts
@@ -73,8 +75,8 @@ impl From<&protobuf::UnnestOptions> for UnnestOptions {
}
}
-impl From<protobuf::WindowFrameUnits> for WindowFrameUnits {
- fn from(units: protobuf::WindowFrameUnits) -> Self {
+impl FromProto<protobuf::WindowFrameUnits> for WindowFrameUnits {
+ fn from_proto(units: protobuf::WindowFrameUnits) -> Self {
match units {
protobuf::WindowFrameUnits::Rows => Self::Rows,
protobuf::WindowFrameUnits::Range => Self::Range,
@@ -83,10 +85,10 @@ impl From<protobuf::WindowFrameUnits> for WindowFrameUnits {
}
}
-impl TryFrom<protobuf::TableReference> for TableReference {
+impl TryFromProto<protobuf::TableReference> for TableReference {
type Error = Error;
- fn try_from(value: protobuf::TableReference) -> Result<Self, Self::Error> {
+ fn try_from_proto(value: protobuf::TableReference) -> Result<Self, Self::Error> {
use protobuf::table_reference::TableReferenceEnum;
let table_reference_enum = value
.table_reference_enum
@@ -109,8 +111,8 @@ impl TryFrom<protobuf::TableReference> for TableReference {
}
}
-impl From<&protobuf::StringifiedPlan> for StringifiedPlan {
- fn from(stringified_plan: &protobuf::StringifiedPlan) -> Self {
+impl FromProto<&protobuf::StringifiedPlan> for StringifiedPlan {
+ fn from_proto(stringified_plan: &protobuf::StringifiedPlan) -> Self {
Self {
plan_type: match stringified_plan
.plan_type
@@ -152,19 +154,25 @@ impl From<&protobuf::StringifiedPlan> for StringifiedPlan {
}
}
-impl TryFrom<protobuf::WindowFrame> for WindowFrame {
+impl TryFromProto<protobuf::WindowFrame> for WindowFrame {
type Error = Error;
- fn try_from(window: protobuf::WindowFrame) -> Result<Self, Self::Error> {
- let units = protobuf::WindowFrameUnits::try_from(window.window_frame_units)
- .map_err(|_| Error::unknown("WindowFrameUnits", window.window_frame_units))?
- .into();
- let start_bound = window.start_bound.required("start_bound")?;
+ fn try_from_proto(window: protobuf::WindowFrame) -> Result<Self, Self::Error> {
+ 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) => {
- end_bound.try_into()
+ WindowFrameBound::try_from_proto(end_bound)
}
})
.transpose()?
@@ -173,10 +181,10 @@ impl TryFrom<protobuf::WindowFrame> for WindowFrame {
}
}
-impl TryFrom<protobuf::WindowFrameBound> for WindowFrameBound {
+impl TryFromProto<protobuf::WindowFrameBound> for WindowFrameBound {
type Error = Error;
- fn try_from(bound: protobuf::WindowFrameBound) -> Result<Self, Self::Error> {
+ fn try_from_proto(bound: protobuf::WindowFrameBound) -> Result<Self, Self::Error> {
let bound_type =
protobuf::WindowFrameBoundType::try_from(bound.window_frame_bound_type)
.map_err(|_| {
@@ -196,8 +204,8 @@ impl TryFrom<protobuf::WindowFrameBound> for WindowFrameBound {
}
}
-impl From<protobuf::JoinType> for JoinType {
- fn from(t: protobuf::JoinType) -> Self {
+impl FromProto<protobuf::JoinType> for JoinType {
+ fn from_proto(t: protobuf::JoinType) -> Self {
match t {
protobuf::JoinType::Inner => JoinType::Inner,
protobuf::JoinType::Left => JoinType::Left,
@@ -213,8 +221,8 @@ impl From<protobuf::JoinType> for JoinType {
}
}
-impl From<protobuf::JoinConstraint> for JoinConstraint {
- fn from(t: protobuf::JoinConstraint) -> Self {
+impl FromProto<protobuf::JoinConstraint> for JoinConstraint {
+ fn from_proto(t: protobuf::JoinConstraint) -> Self {
match t {
protobuf::JoinConstraint::On => JoinConstraint::On,
protobuf::JoinConstraint::Using => JoinConstraint::Using,
@@ -222,8 +230,8 @@ impl From<protobuf::JoinConstraint> for JoinConstraint {
}
}
-impl From<protobuf::NullEquality> for NullEquality {
- fn from(t: protobuf::NullEquality) -> Self {
+impl FromProto<protobuf::NullEquality> for NullEquality {
+ fn from_proto(t: protobuf::NullEquality) -> Self {
match t {
protobuf::NullEquality::NullEqualsNothing => NullEquality::NullEqualsNothing,
protobuf::NullEquality::NullEqualsNull => NullEquality::NullEqualsNull,
@@ -231,8 +239,8 @@ impl From<protobuf::NullEquality> for NullEquality {
}
}
-impl From<protobuf::dml_node::Type> for WriteOp {
- fn from(t: protobuf::dml_node::Type) -> Self {
+impl FromProto<protobuf::dml_node::Type> for WriteOp {
+ fn from_proto(t: protobuf::dml_node::Type) -> Self {
match t {
protobuf::dml_node::Type::Update => WriteOp::Update,
protobuf::dml_node::Type::Delete => WriteOp::Delete,
@@ -247,8 +255,8 @@ impl From<protobuf::dml_node::Type> for WriteOp {
}
}
-impl From<protobuf::NullTreatment> for NullTreatment {
- fn from(t: protobuf::NullTreatment) -> Self {
+impl FromProto<protobuf::NullTreatment> for NullTreatment {
+ fn from_proto(t: protobuf::NullTreatment) -> Self {
match t {
protobuf::NullTreatment::RespectNulls => NullTreatment::RespectNulls,
protobuf::NullTreatment::IgnoreNulls => NullTreatment::IgnoreNulls,
@@ -304,7 +312,7 @@ pub fn parse_expr(
.window_frame
.as_ref()
.map::<Result<WindowFrame, _>, _>(|window_frame| {
- let window_frame: WindowFrame = window_frame.clone().try_into()?;
+ let window_frame = WindowFrame::try_from_proto(window_frame.clone())?;
window_frame
.regularize_order_bys(&mut order_by)
.map(|_| window_frame)
@@ -322,7 +330,7 @@ pub fn parse_expr(
"Received a WindowExprNode message with unknown NullTreatment {null_treatment}",
))
})?;
- Some(NullTreatment::from(null_treatment))
+ Some(NullTreatment::from_proto(null_treatment))
}
None => None,
};
@@ -371,7 +379,7 @@ pub fn parse_expr(
alias
.relation
.first()
- .map(|r| TableReference::try_from(r.clone()))
+ .map(|r| TableReference::try_from_proto(r.clone()))
.transpose()?,
alias.alias.clone(),
))),
@@ -571,7 +579,10 @@ pub fn parse_expr(
in_list.negated,
))),
ExprType::Wildcard(protobuf::Wildcard { qualifier }) => {
- let qualifier = qualifier.to_owned().map(|x| x.try_into()).transpose()?;
+ let qualifier = qualifier
+ .to_owned()
+ .map(TableReference::try_from_proto)
+ .transpose()?;
#[expect(deprecated)]
Ok(Expr::Wildcard {
qualifier,
@@ -609,7 +620,7 @@ pub fn parse_expr(
"Received an AggregateUdfExprNode message with unknown NullTreatment {null_treatment}",
))
})?;
- Some(NullTreatment::from(null_treatment))
+ Some(NullTreatment::from_proto(null_treatment))
}
None => None,
};
@@ -742,42 +753,11 @@ fn parse_escape_char(s: &str) -> Result<Option<char>> {
}
pub fn from_proto_binary_op(op: &str) -> Result<Operator, Error> {
- match op {
- "And" => Ok(Operator::And),
- "Or" => Ok(Operator::Or),
- "Eq" => Ok(Operator::Eq),
- "NotEq" => Ok(Operator::NotEq),
- "LtEq" => Ok(Operator::LtEq),
- "Lt" => Ok(Operator::Lt),
- "Gt" => Ok(Operator::Gt),
- "GtEq" => Ok(Operator::GtEq),
- "Plus" => Ok(Operator::Plus),
- "Minus" => Ok(Operator::Minus),
- "Multiply" => Ok(Operator::Multiply),
- "Divide" => Ok(Operator::Divide),
- "Modulo" => Ok(Operator::Modulo),
- "IsDistinctFrom" => Ok(Operator::IsDistinctFrom),
- "IsNotDistinctFrom" => Ok(Operator::IsNotDistinctFrom),
- "BitwiseAnd" => Ok(Operator::BitwiseAnd),
- "BitwiseOr" => Ok(Operator::BitwiseOr),
- "BitwiseXor" => Ok(Operator::BitwiseXor),
- "BitwiseShiftLeft" => Ok(Operator::BitwiseShiftLeft),
- "BitwiseShiftRight" => Ok(Operator::BitwiseShiftRight),
- "RegexIMatch" => Ok(Operator::RegexIMatch),
- "RegexMatch" => Ok(Operator::RegexMatch),
- "RegexNotIMatch" => Ok(Operator::RegexNotIMatch),
- "RegexNotMatch" => Ok(Operator::RegexNotMatch),
- "LikeMatch" => Ok(Operator::LikeMatch),
- "ILikeMatch" => Ok(Operator::ILikeMatch),
- "NotLikeMatch" => Ok(Operator::NotLikeMatch),
- "NotILikeMatch" => Ok(Operator::NotILikeMatch),
- "StringConcat" => Ok(Operator::StringConcat),
- "AtArrow" => Ok(Operator::AtArrow),
- "ArrowAt" => Ok(Operator::ArrowAt),
- other => Err(proto_error(format!(
- "Unsupported binary operator '{other:?}'"
- ))),
- }
+ // The proto-string <-> `Operator` mapping is canonically owned by
+ // `datafusion-expr-common` so `datafusion-proto` (logical plans) and
+ // `PhysicalExpr` decoders (e.g. `BinaryExpr`) share one source of truth.
+ Operator::from_proto_name(op)
+ .ok_or_else(|| proto_error(format!("Unsupported binary operator '{op:?}'")))
}
fn parse_optional_expr(
diff --git a/datafusion/proto/src/logical_plan/mod.rs b/datafusion/proto/src/logical_plan/mod.rs
index 8cdd5c5dea..542cae890d 100644
--- a/datafusion/proto/src/logical_plan/mod.rs
+++ b/datafusion/proto/src/logical_plan/mod.rs
@@ -19,13 +19,14 @@ 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,
CustomTableScanNode, DmlNode, SortExprNodeCollection, dml_node,
};
use crate::{
- convert_required, into_required,
+ convert_required,
protobuf::{
self, LogicalExtensionNode, LogicalPlanNode,
listing_table_scan_node::FileFormatType, logical_plan_node::LogicalPlanType,
@@ -39,7 +40,7 @@ use datafusion_catalog::empty::EmptyTable;
use datafusion_common::file_options::file_type::FileType;
use datafusion_common::format::ExplainFormat;
use datafusion_common::{
- Result, TableReference, ToDFSchema, assert_or_internal_err, context,
+ NullEquality, Result, TableReference, ToDFSchema, assert_or_internal_err, context,
internal_datafusion_err, internal_err, not_impl_err, plan_err,
};
use datafusion_datasource::file_format::FileFormat;
@@ -57,11 +58,11 @@ use datafusion_datasource_json::file_format::{
use datafusion_datasource_parquet::file_format::{ParquetFormat, ParquetFormatFactory};
use datafusion_expr::{
AggregateUDF, DmlStatement, FetchType, HigherOrderUDF, RecursiveQuery, SkipType,
- TableSource, Unnest,
+ TableSource, Unnest, WriteOp,
};
use datafusion_expr::{
- DistinctOn, DropView, Expr, LogicalPlan, LogicalPlanBuilder, ScalarUDF, SortExpr,
- Statement, WindowUDF, dml,
+ DistinctOn, DropView, Expr, JoinConstraint, LogicalPlan, LogicalPlanBuilder,
+ ScalarUDF, SortExpr, Statement, WindowUDF, dml,
logical_plan::{
Aggregate, CreateCatalog, CreateCatalogSchema, CreateExternalTable, CreateView,
DdlStatement, Distinct, EmptyRelation, Extension, Join, Prepare, Projection,
@@ -341,7 +342,7 @@ fn from_table_reference(
)
})?;
- Ok(table_ref.clone().try_into()?)
+ Ok(TableReference::try_from_proto(table_ref.clone())?)
}
/// Converts [LogicalPlan::TableScan] to [TableSource]
@@ -900,9 +901,9 @@ impl AsLogicalPlan for LogicalPlanNode {
Arc::new(right),
on,
filter,
- join_type.into(),
- join_constraint.into(),
- null_equality.into(),
+ datafusion_expr::JoinType::from_proto(join_type),
+ JoinConstraint::from_proto(join_constraint),
+ NullEquality::from_proto(null_equality),
join.null_aware,
)?))
}
@@ -1065,7 +1066,13 @@ impl AsLogicalPlan for LogicalPlanNode {
LogicalPlanBuilder::from(input)
.unnest_columns_with_options(
unnest.exec_columns.iter().map(|c| c.into()).collect(),
- into_required!(unnest.options)?,
+ unnest
+ .options
+ .as_ref()
+ .map(datafusion_common::UnnestOptions::from_proto)
+ .ok_or_else(|| {
+ proto_error("Missing required field in protobuf")
+ })?,
)?
.build()
}
@@ -1137,7 +1144,7 @@ impl AsLogicalPlan for LogicalPlanNode {
Ok(LogicalPlan::Dml(datafusion_expr::DmlStatement::new(
from_table_reference(dml_node.table_name.as_ref(), "DML ")?,
to_table_source(&dml_node.target, ctx, extension_codec)?,
- dml_node.dml_type().into(),
+ WriteOp::from_proto(dml_node.dml_type()),
Arc::new(into_logical_plan!(dml_node.input, ctx, extension_codec)?),
)))
}
@@ -1292,7 +1299,9 @@ impl AsLogicalPlan for LogicalPlanNode {
logical_plan_type: Some(LogicalPlanType::ListingScan(
protobuf::ListingTableScanNode {
file_format_type: Some(file_format_type),
- table_name: Some(table_name.clone().into()),
+ table_name: Some(protobuf::TableReference::from_proto(
+ table_name.clone(),
+ )),
collect_stat: options.collect_stat,
file_extension: options.file_extension.clone(),
table_partition_cols: partition_columns,
@@ -1314,7 +1323,9 @@ impl AsLogicalPlan for LogicalPlanNode {
Ok(LogicalPlanNode {
logical_plan_type: Some(LogicalPlanType::ViewScan(Box::new(
protobuf::ViewTableScanNode {
- table_name: Some(table_name.clone().into()),
+ table_name: Some(protobuf::TableReference::from_proto(
+ table_name.clone(),
+ )),
input: Some(Box::new(
LogicalPlanNode::try_from_logical_plan(
view_table.logical_plan(),
@@ -1351,7 +1362,9 @@ impl AsLogicalPlan for LogicalPlanNode {
Ok(LogicalPlanNode {
logical_plan_type: Some(LogicalPlanType::EmptyTableScan(
protobuf::EmptyTableScanNode {
- table_name: Some(table_name.clone().into()),
+ table_name: Some(protobuf::TableReference::from_proto(
+ table_name.clone(),
+ )),
schema: Some(schema),
projection,
filters,
@@ -1365,7 +1378,9 @@ 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(table_name.clone().into()),
+ table_name: Some(protobuf::TableReference::from_proto(
+ table_name.clone(),
+ )),
projection,
schema: Some(schema),
filters,
@@ -1516,11 +1531,11 @@ impl AsLogicalPlan for LogicalPlanNode {
.collect::<Result<Vec<_>, ToProtoError>>()?
.into_iter()
.unzip();
- let join_type: protobuf::JoinType = join_type.to_owned().into();
- let join_constraint: protobuf::JoinConstraint =
- join_constraint.to_owned().into();
- let null_equality: protobuf::NullEquality =
- null_equality.to_owned().into();
+ let join_type = protobuf::JoinType::from_proto(join_type.to_owned());
+ let join_constraint =
+ protobuf::JoinConstraint::from_proto(join_constraint.to_owned());
+ let null_equality =
+ protobuf::NullEquality::from_proto(null_equality.to_owned());
let filter = filter
.as_ref()
.map(|e| serialize_expr(e, extension_codec).map(Box::new))
@@ -1559,7 +1574,9 @@ impl AsLogicalPlan for LogicalPlanNode {
logical_plan_type: Some(LogicalPlanType::SubqueryAlias(Box::new(
protobuf::SubqueryAliasNode {
input: Some(Box::new(input)),
- alias: Some((*alias).clone().into()),
+ alias: Some(protobuf::TableReference::from_proto(
+ (*alias).clone(),
+ )),
},
))),
})
@@ -1690,7 +1707,9 @@ impl AsLogicalPlan for LogicalPlanNode {
Ok(LogicalPlanNode {
logical_plan_type: Some(LogicalPlanType::CreateExternalTable(
protobuf::CreateExternalTableNode {
- name: Some(name.clone().into()),
+ name: Some(protobuf::TableReference::from_proto(
+ name.clone(),
+ )),
location: location.clone(),
file_type: file_type.clone(),
schema: Some(df_schema.try_into()?),
@@ -1717,7 +1736,7 @@ impl AsLogicalPlan for LogicalPlanNode {
})) => Ok(LogicalPlanNode {
logical_plan_type: Some(LogicalPlanType::CreateView(Box::new(
protobuf::CreateViewNode {
- name: Some(name.clone().into()),
+ name: Some(protobuf::TableReference::from_proto(name.clone())),
input: Some(Box::new(LogicalPlanNode::try_from_logical_plan(
input,
extension_codec,
@@ -1889,7 +1908,7 @@ impl AsLogicalPlan for LogicalPlanNode {
.map(|c| *c as u64)
.collect(),
schema: Some(schema.try_into()?),
- options: Some(options.into()),
+ options: Some(protobuf::UnnestOptions::from_proto(options)),
},
))),
})
@@ -1910,7 +1929,7 @@ impl AsLogicalPlan for LogicalPlanNode {
})) => Ok(LogicalPlanNode {
logical_plan_type: Some(LogicalPlanType::DropView(
protobuf::DropViewNode {
- name: Some(name.clone().into()),
+ name: Some(protobuf::TableReference::from_proto(name.clone())),
if_exists: *if_exists,
schema: Some(schema.try_into()?),
},
@@ -1937,7 +1956,7 @@ impl AsLogicalPlan for LogicalPlanNode {
}) => {
let input =
LogicalPlanNode::try_from_logical_plan(input, extension_codec)?;
- let dml_type: dml_node::Type = op.into();
+ let dml_type = dml_node::Type::from_proto(op);
Ok(LogicalPlanNode {
logical_plan_type: Some(LogicalPlanType::Dml(Box::new(DmlNode {
input: Some(Box::new(input)),
@@ -1946,7 +1965,9 @@ impl AsLogicalPlan for LogicalPlanNode {
Arc::clone(target),
extension_codec,
)?)),
- table_name: Some(table_name.clone().into()),
+ table_name: Some(protobuf::TableReference::from_proto(
+ 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 d79107d1d0..71a6bd824a 100644
--- a/datafusion/proto/src/logical_plan/to_proto.rs
+++ b/datafusion/proto/src/logical_plan/to_proto.rs
@@ -50,10 +50,11 @@ use crate::protobuf::{
};
use super::{AsLogicalPlan, LogicalExtensionCodec};
+use crate::convert::{FromProto, TryFromProto};
use crate::protobuf::LogicalPlanNode;
-impl From<&UnnestOptions> for protobuf::UnnestOptions {
- fn from(opts: &UnnestOptions) -> Self {
+impl FromProto<&UnnestOptions> for protobuf::UnnestOptions {
+ fn from_proto(opts: &UnnestOptions) -> Self {
Self {
preserve_nulls: opts.preserve_nulls,
recursions: opts
@@ -69,8 +70,8 @@ impl From<&UnnestOptions> for protobuf::UnnestOptions {
}
}
-impl From<&StringifiedPlan> for protobuf::StringifiedPlan {
- fn from(stringified_plan: &StringifiedPlan) -> Self {
+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 {
@@ -130,8 +131,8 @@ impl From<&StringifiedPlan> for protobuf::StringifiedPlan {
}
}
-impl From<WindowFrameUnits> for protobuf::WindowFrameUnits {
- fn from(units: WindowFrameUnits) -> Self {
+impl FromProto<WindowFrameUnits> for protobuf::WindowFrameUnits {
+ fn from_proto(units: WindowFrameUnits) -> Self {
match units {
WindowFrameUnits::Rows => Self::Rows,
WindowFrameUnits::Range => Self::Range,
@@ -140,10 +141,10 @@ impl From<WindowFrameUnits> for protobuf::WindowFrameUnits {
}
}
-impl TryFrom<&WindowFrameBound> for protobuf::WindowFrameBound {
+impl TryFromProto<&WindowFrameBound> for protobuf::WindowFrameBound {
type Error = Error;
- fn try_from(bound: &WindowFrameBound) -> Result<Self, Self::Error> {
+ fn try_from_proto(bound: &WindowFrameBound) -> Result<Self, Self::Error> {
Ok(match bound {
WindowFrameBound::CurrentRow => Self {
window_frame_bound_type: protobuf::WindowFrameBoundType::CurrentRow
@@ -162,15 +163,18 @@ impl TryFrom<&WindowFrameBound> for protobuf::WindowFrameBound {
}
}
-impl TryFrom<&WindowFrame> for protobuf::WindowFrame {
+impl TryFromProto<&WindowFrame> for protobuf::WindowFrame {
type Error = Error;
- fn try_from(window: &WindowFrame) -> Result<Self, Self::Error> {
+ fn try_from_proto(window: &WindowFrame) -> Result<Self, Self::Error> {
Ok(Self {
- window_frame_units: protobuf::WindowFrameUnits::from(window.units).into(),
- start_bound: Some((&window.start_bound).try_into()?),
+ 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(
- (&window.end_bound).try_into()?,
+ protobuf::WindowFrameBound::try_from_proto(&window.end_bound)?,
)),
})
}
@@ -209,7 +213,7 @@ pub fn serialize_expr(
expr: Some(Box::new(serialize_expr(expr.as_ref(), codec)?)),
relation: relation
.to_owned()
- .map(|r| vec![r.into()])
+ .map(|r| vec![protobuf::TableReference::from_proto(r)])
.unwrap_or(vec![]),
alias: name.to_owned(),
metadata: metadata
@@ -339,8 +343,7 @@ pub fn serialize_expr(
let partition_by = serialize_exprs(partition_by, codec)?;
let order_by = serialize_sorts(order_by, codec)?;
- let window_frame: Option<protobuf::WindowFrame> =
- Some(window_frame.try_into()?);
+ let window_frame = Some(protobuf::WindowFrame::try_from_proto(window_frame)?);
let window_expr = protobuf::WindowExprNode {
exprs: serialize_exprs(args, codec)?,
@@ -354,7 +357,7 @@ pub fn serialize_expr(
None => None,
},
null_treatment: null_treatment
- .map(|nt| protobuf::NullTreatment::from(nt).into()),
+ .map(|nt| protobuf::NullTreatment::from_proto(nt).into()),
fun_definition,
};
protobuf::LogicalExprNode {
@@ -387,7 +390,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(nt).into()),
+ .map(|nt| protobuf::NullTreatment::from_proto(nt).into()),
},
))),
}
@@ -578,7 +581,9 @@ pub fn serialize_expr(
#[expect(deprecated)]
Expr::Wildcard { qualifier, .. } => protobuf::LogicalExprNode {
expr_type: Some(ExprType::Wildcard(protobuf::Wildcard {
- qualifier: qualifier.to_owned().map(|x| x.into()),
+ qualifier: qualifier
+ .to_owned()
+ .map(protobuf::TableReference::from_proto),
})),
},
Expr::ScalarSubquery(subquery) => protobuf::LogicalExprNode {
@@ -681,8 +686,8 @@ where
.collect::<Result<Vec<_>, Error>>()
}
-impl From<TableReference> for protobuf::TableReference {
- fn from(t: TableReference) -> Self {
+impl FromProto<TableReference> for protobuf::TableReference {
+ fn from_proto(t: TableReference) -> Self {
use protobuf::table_reference::TableReferenceEnum;
let table_reference_enum = match t {
TableReference::Bare { table } => {
@@ -713,8 +718,8 @@ impl From<TableReference> for protobuf::TableReference {
}
}
-impl From<JoinType> for protobuf::JoinType {
- fn from(t: JoinType) -> Self {
+impl FromProto<JoinType> for protobuf::JoinType {
+ fn from_proto(t: JoinType) -> Self {
match t {
JoinType::Inner => protobuf::JoinType::Inner,
JoinType::Left => protobuf::JoinType::Left,
@@ -730,8 +735,8 @@ impl From<JoinType> for protobuf::JoinType {
}
}
-impl From<JoinConstraint> for protobuf::JoinConstraint {
- fn from(t: JoinConstraint) -> Self {
+impl FromProto<JoinConstraint> for protobuf::JoinConstraint {
+ fn from_proto(t: JoinConstraint) -> Self {
match t {
JoinConstraint::On => protobuf::JoinConstraint::On,
JoinConstraint::Using => protobuf::JoinConstraint::Using,
@@ -739,8 +744,8 @@ impl From<JoinConstraint> for protobuf::JoinConstraint {
}
}
-impl From<NullEquality> for protobuf::NullEquality {
- fn from(t: NullEquality) -> Self {
+impl FromProto<NullEquality> for protobuf::NullEquality {
+ fn from_proto(t: NullEquality) -> Self {
match t {
NullEquality::NullEqualsNothing => protobuf::NullEquality::NullEqualsNothing,
NullEquality::NullEqualsNull => protobuf::NullEquality::NullEqualsNull,
@@ -748,8 +753,8 @@ impl From<NullEquality> for protobuf::NullEquality {
}
}
-impl From<&WriteOp> for protobuf::dml_node::Type {
- fn from(t: &WriteOp) -> Self {
+impl FromProto<&WriteOp> for protobuf::dml_node::Type {
+ fn from_proto(t: &WriteOp) -> Self {
match t {
WriteOp::Insert(InsertOp::Append) => protobuf::dml_node::Type::InsertAppend,
WriteOp::Insert(InsertOp::Overwrite) => {
@@ -764,8 +769,8 @@ impl From<&WriteOp> for protobuf::dml_node::Type {
}
}
-impl From<NullTreatment> for protobuf::NullTreatment {
- fn from(t: NullTreatment) -> Self {
+impl FromProto<NullTreatment> 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 43ebf04743..f5fd214ef6 100644
--- a/datafusion/proto/src/physical_plan/from_proto.rs
+++ b/datafusion/proto/src/physical_plan/from_proto.rs
@@ -57,19 +57,13 @@ use super::{
DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalPlanDecodeContext,
PhysicalProtoConverterExtension,
};
-use crate::logical_plan::{self};
+use crate::convert::TryFromProto;
use crate::protobuf::physical_expr_node::ExprType;
-use crate::{convert_required, protobuf};
+use crate::{convert_required, convert_required_proto, protobuf};
use datafusion_physical_expr::expressions::{
DynamicFilterInner, DynamicFilterPhysicalExpr,
};
-impl From<&protobuf::PhysicalColumn> for Column {
- fn from(c: &protobuf::PhysicalColumn) -> Column {
- Column::new(&c.name, c.index as usize)
- }
-}
-
/// Parses a physical sort expression from a protobuf.
///
/// # Arguments
@@ -154,7 +148,7 @@ pub fn parse_physical_window_expr(
let window_frame = proto
.window_frame
.as_ref()
- .map(|wf| wf.clone().try_into())
+ .map(|wf| datafusion_expr::WindowFrame::try_from_proto(wf.clone()))
.transpose()
.map_err(|e| internal_datafusion_err!("{e}"))?
.ok_or_else(|| {
@@ -266,57 +260,27 @@ pub fn parse_physical_expr_with_converter(
.as_ref()
.ok_or_else(|| proto_error("Unexpected empty physical expression"))?;
+ // Decoder context handed to per-expression `try_from_proto` constructors.
+ // This is the new shape the codebase is migrating toward (see #21835);
+ // the remaining `ExprType` variants stay matched inline until they migrate.
+ let decoder = ConverterDecoder {
+ ctx,
+ proto_converter,
+ };
+ let decode_ctx =
+ datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx::new(
+ input_schema,
+ &decoder,
+ );
+
let pexpr: Arc<dyn PhysicalExpr> = match expr_type {
- ExprType::Column(c) => {
- let pcol: Column = c.into();
- Arc::new(pcol)
- }
+ // Migrated expressions take the whole `PhysicalExprNode` and unwrap
+ // their own `ExprType` variant — see #21835. This match only routes
+ // to the right constructor.
+ ExprType::Column(_) => Column::try_from_proto(proto, &decode_ctx)?,
ExprType::UnknownColumn(c) => Arc::new(UnKnownColumn::new(&c.name)),
ExprType::Literal(scalar) => Arc::new(Literal::new(scalar.try_into()?)),
- ExprType::BinaryExpr(binary_expr) => {
- let op = logical_plan::from_proto::from_proto_binary_op(&binary_expr.op)?;
- if !binary_expr.operands.is_empty() {
- // New linearized format: reduce the flat operands list back into
- // a nested binary expression tree.
- let operands: Vec<Arc<dyn PhysicalExpr>> = binary_expr
- .operands
- .iter()
- .map(|e| proto_converter.proto_to_physical_expr(e, input_schema, ctx))
- .collect::<Result<Vec<_>>>()?;
-
- if operands.len() < 2 {
- return Err(proto_error(
- "A binary expression must always have at least 2 operands",
- ));
- }
-
- operands
- .into_iter()
- .reduce(|left, right| Arc::new(BinaryExpr::new(left, op, right)))
- .expect(
- "Binary expression could not be reduced to a single expression.",
- )
- } else {
- // Legacy format with l/r fields
- Arc::new(BinaryExpr::new(
- parse_required_physical_expr(
- binary_expr.l.as_deref(),
- ctx,
- "left",
- input_schema,
- proto_converter,
- )?,
- op,
- parse_required_physical_expr(
- binary_expr.r.as_deref(),
- ctx,
- "right",
- input_schema,
- proto_converter,
- )?,
- ))
- }
- }
+ ExprType::BinaryExpr(_) => BinaryExpr::try_from_proto(proto, &decode_ctx)?,
ExprType::AggregateExpr(_) => {
return not_impl_err!(
"Cannot convert aggregate expr node to physical expression"
@@ -694,7 +658,7 @@ pub fn parse_protobuf_file_scan_config(
let file_groups = proto
.file_groups
.iter()
- .map(|f| f.try_into())
+ .map(FileGroup::try_from_proto)
.collect::<Result<Vec<_>, _>>()?;
let object_store_url = match proto.object_store_url.is_empty() {
@@ -763,10 +727,10 @@ pub fn parse_record_batches(buf: &[u8]) -> Result<Vec<RecordBatch>> {
Ok(batches)
}
-impl TryFrom<&protobuf::PartitionedFile> for PartitionedFile {
+impl TryFromProto<&protobuf::PartitionedFile> for PartitionedFile {
type Error = DataFusionError;
- fn try_from(val: &protobuf::PartitionedFile) -> Result<Self, Self::Error> {
+ 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}")))?,
@@ -782,7 +746,7 @@ impl TryFrom<&protobuf::PartitionedFile> for PartitionedFile {
.collect::<Result<Vec<_>, _>>()?,
);
if let Some(range) = val.range.as_ref() {
- let file_range: FileRange = range.try_into()?;
+ 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() {
@@ -792,10 +756,10 @@ impl TryFrom<&protobuf::PartitionedFile> for PartitionedFile {
}
}
-impl TryFrom<&protobuf::FileRange> for FileRange {
+impl TryFromProto<&protobuf::FileRange> for FileRange {
type Error = DataFusionError;
- fn try_from(value: &protobuf::FileRange) -> Result<Self, Self::Error> {
+ fn try_from_proto(value: &protobuf::FileRange) -> Result<Self, Self::Error> {
Ok(FileRange {
start: value.start,
end: value.end,
@@ -803,61 +767,61 @@ impl TryFrom<&protobuf::FileRange> for FileRange {
}
}
-impl TryFrom<&protobuf::FileGroup> for FileGroup {
+impl TryFromProto<&protobuf::FileGroup> for FileGroup {
type Error = DataFusionError;
- fn try_from(val: &protobuf::FileGroup) -> Result<Self, Self::Error> {
+ fn try_from_proto(val: &protobuf::FileGroup) -> Result<Self, Self::Error> {
let files = val
.files
.iter()
- .map(|f| f.try_into())
+ .map(PartitionedFile::try_from_proto)
.collect::<Result<Vec<_>, _>>()?;
Ok(FileGroup::new(files))
}
}
-impl TryFrom<&protobuf::JsonSink> for JsonSink {
+impl TryFromProto<&protobuf::JsonSink> for JsonSink {
type Error = DataFusionError;
- fn try_from(value: &protobuf::JsonSink) -> Result<Self, Self::Error> {
+ fn try_from_proto(value: &protobuf::JsonSink) -> Result<Self, Self::Error> {
Ok(Self::new(
- convert_required!(value.config)?,
+ convert_required_proto!(FileSinkConfig, value.config)?,
convert_required!(value.writer_options)?,
))
}
}
#[cfg(feature = "parquet")]
-impl TryFrom<&protobuf::ParquetSink> for ParquetSink {
+impl TryFromProto<&protobuf::ParquetSink> for ParquetSink {
type Error = DataFusionError;
- fn try_from(value: &protobuf::ParquetSink) -> Result<Self, Self::Error> {
+ fn try_from_proto(value: &protobuf::ParquetSink) -> Result<Self, Self::Error> {
Ok(Self::new(
- convert_required!(value.config)?,
+ convert_required_proto!(FileSinkConfig, value.config)?,
convert_required!(value.parquet_options)?,
))
}
}
-impl TryFrom<&protobuf::CsvSink> for CsvSink {
+impl TryFromProto<&protobuf::CsvSink> for CsvSink {
type Error = DataFusionError;
- fn try_from(value: &protobuf::CsvSink) -> Result<Self, Self::Error> {
+ fn try_from_proto(value: &protobuf::CsvSink) -> Result<Self, Self::Error> {
Ok(Self::new(
- convert_required!(value.config)?,
+ convert_required_proto!(FileSinkConfig, value.config)?,
convert_required!(value.writer_options)?,
))
}
}
-impl TryFrom<&protobuf::FileSinkConfig> for FileSinkConfig {
+impl TryFromProto<&protobuf::FileSinkConfig> for FileSinkConfig {
type Error = DataFusionError;
- fn try_from(conf: &protobuf::FileSinkConfig) -> Result<Self, Self::Error> {
+ fn try_from_proto(conf: &protobuf::FileSinkConfig) -> Result<Self, Self::Error> {
let file_group = FileGroup::new(
conf.file_groups
.iter()
- .map(|f| f.try_into())
+ .map(PartitionedFile::try_from_proto)
.collect::<Result<Vec<_>>>()?,
);
let table_paths = conf
@@ -904,6 +868,33 @@ impl TryFrom<&protobuf::FileSinkConfig> for FileSinkConfig {
}
}
+/// Concrete [`PhysicalExprDecode`] driver that backs
+/// [`PhysicalExprDecodeCtx`] inside `parse_physical_expr_with_converter`.
+///
+/// Today this is a thin wrapper that re-enters the central match through
+/// `proto_to_physical_expr`; once more expressions migrate, the central match
+/// shrinks and a future builder-style decoder can take over.
+///
+/// [`PhysicalExprDecode`]: datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecode
+/// [`PhysicalExprDecodeCtx`]: datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx
+struct ConverterDecoder<'a, 'b> {
+ ctx: &'a PhysicalPlanDecodeContext<'b>,
+ proto_converter: &'a dyn PhysicalProtoConverterExtension,
+}
+
+impl datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecode
+ for ConverterDecoder<'_, '_>
+{
+ fn decode(
+ &self,
+ node: &protobuf::PhysicalExprNode,
+ schema: &Schema,
+ ) -> Result<Arc<dyn PhysicalExpr>> {
+ self.proto_converter
+ .proto_to_physical_expr(node, schema, self.ctx)
+ }
+}
+
#[cfg(test)]
mod tests {
@@ -920,10 +911,10 @@ mod tests {
version: None,
});
- let proto = protobuf::PartitionedFile::try_from(&pf).unwrap();
+ let proto = protobuf::PartitionedFile::try_from_proto(&pf).unwrap();
assert_eq!(proto.path, path_str);
- let pf2 = PartitionedFile::try_from(&proto).unwrap();
+ 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);
@@ -941,7 +932,7 @@ mod tests {
statistics: None,
};
- let err = PartitionedFile::try_from(&proto).unwrap_err();
+ 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/mod.rs b/datafusion/proto/src/physical_plan/mod.rs
index 27284664b0..89871318ee 100644
--- a/datafusion/proto/src/physical_plan/mod.rs
+++ b/datafusion/proto/src/physical_plan/mod.rs
@@ -25,8 +25,10 @@ use arrow::compute::SortOptions;
use arrow::datatypes::{IntervalMonthDayNanoType, Schema, SchemaRef};
use datafusion_catalog::memory::MemorySourceConfig;
use datafusion_common::config::CsvOptions;
+use datafusion_common::display::StringifiedPlan;
use datafusion_common::{
- DataFusionError, Result, internal_datafusion_err, internal_err, not_impl_err,
+ DataFusionError, JoinType, NullEquality, Result, internal_datafusion_err,
+ internal_err, not_impl_err,
};
#[cfg(feature = "parquet")]
use datafusion_datasource::file::FileSource;
@@ -97,6 +99,8 @@ use prost::bytes::BufMut;
use self::from_proto::parse_protobuf_partitioning;
use self::to_proto::serialize_partitioning;
use crate::common::{byte_to_string, str_to_byte};
+use crate::convert::{FromProto, TryFromProto};
+use crate::convert_required;
use crate::physical_plan::from_proto::{
parse_physical_expr_with_converter, parse_physical_sort_expr,
parse_physical_sort_exprs, parse_physical_window_expr,
@@ -114,7 +118,6 @@ use crate::protobuf::{
self, ListUnnest as ProtoListUnnest, SortExprNode, SortMergeJoinExecNode,
proto_error, window_agg_exec_node,
};
-use crate::{convert_required, into_required};
pub mod from_proto;
pub mod to_proto;
@@ -270,8 +273,24 @@ impl AsExecutionPlan for protobuf::PhysicalPlanNode {
}
}
-impl protobuf::PhysicalPlanNode {
- pub fn try_into_physical_plan_with_converter(
+/// Extension methods on [`protobuf::PhysicalPlanNode`].
+///
+/// The prost-generated `PhysicalPlanNode` struct lives in
+/// `datafusion-proto-models`, which is foreign to this crate, so the orphan
+/// rule forbids inherent `impl` blocks here. Instead, all (de)serialization
+/// helpers are exposed through this trait. Callers can bring it in scope with
+/// `use datafusion_proto::physical_plan::PhysicalPlanNodeExt;`.
+///
+/// Method bodies live in the default trait implementation. To make the trait
+/// usable as if it were inherent (i.e. let bodies access fields on `self`),
+/// implementors provide [`PhysicalPlanNodeExt::node`] returning a reference
+/// back to the concrete `protobuf::PhysicalPlanNode`. Default method bodies
+/// then go through `self.node()` to read fields.
+pub trait PhysicalPlanNodeExt: Sized {
+ /// Returns a reference to the underlying [`protobuf::PhysicalPlanNode`].
+ fn node(&self) -> &protobuf::PhysicalPlanNode;
+
+ fn try_into_physical_plan_with_converter(
&self,
ctx: &TaskContext,
codec: &dyn PhysicalExtensionCodec,
@@ -281,14 +300,15 @@ impl protobuf::PhysicalPlanNode {
self.try_into_physical_plan_with_context(&decode_ctx, proto_converter)
}
- pub(crate) fn try_into_physical_plan_with_context(
+ fn try_into_physical_plan_with_context(
&self,
ctx: &PhysicalPlanDecodeContext<'_>,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn ExecutionPlan>> {
- let plan = self.physical_plan_type.as_ref().ok_or_else(|| {
+ let plan = self.node().physical_plan_type.as_ref().ok_or_else(|| {
proto_error(format!(
- "physical_plan::from_proto() Unsupported physical plan '{self:?}'"
+ "physical_plan::from_proto() Unsupported physical plan '{:?}'",
+ self.node(),
))
})?;
match plan {
@@ -415,14 +435,11 @@ impl protobuf::PhysicalPlanNode {
}
}
- pub fn try_from_physical_plan_with_converter(
+ fn try_from_physical_plan_with_converter(
plan: Arc<dyn ExecutionPlan>,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self>
- where
- Self: Sized,
- {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let plan_clone = Arc::clone(&plan);
let plan = plan.as_ref() as &dyn Any;
@@ -695,9 +712,7 @@ impl protobuf::PhysicalPlanNode {
),
}
}
-}
-impl protobuf::PhysicalPlanNode {
fn try_into_explain_physical_plan(
&self,
explain: &protobuf::ExplainExecNode,
@@ -709,7 +724,7 @@ impl protobuf::PhysicalPlanNode {
explain
.stringified_plans
.iter()
- .map(|plan| plan.into())
+ .map(StringifiedPlan::from_proto)
.collect(),
explain.verbose,
)))
@@ -1498,10 +1513,10 @@ impl protobuf::PhysicalPlanNode {
right,
on,
filter,
- &join_type.into(),
+ &JoinType::from_proto(join_type),
projection,
partition_mode,
- null_equality.into(),
+ NullEquality::from_proto(null_equality),
hashjoin.null_aware,
)?;
@@ -1640,8 +1655,8 @@ impl protobuf::PhysicalPlanNode {
right,
on,
filter,
- &join_type.into(),
- null_equality.into(),
+ &JoinType::from_proto(join_type),
+ NullEquality::from_proto(null_equality),
left_sort_exprs,
right_sort_exprs,
partition_mode,
@@ -1713,6 +1728,7 @@ impl protobuf::PhysicalPlanNode {
ctx: &PhysicalPlanDecodeContext<'_>,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn ExecutionPlan>> {
+ let node = self.node();
let input = into_physical_plan(&sort.input, ctx, proto_converter)?;
let exprs = sort
.expr
@@ -1720,7 +1736,7 @@ impl protobuf::PhysicalPlanNode {
.map(|expr| {
let expr = expr.expr_type.as_ref().ok_or_else(|| {
proto_error(format!(
- "physical_plan::from_proto() Unexpected expr {self:?}"
+ "physical_plan::from_proto() Unexpected expr {node:?}"
))
})?;
if let ExprType::Sort(sort_expr) = expr {
@@ -1729,7 +1745,7 @@ impl protobuf::PhysicalPlanNode {
.as_ref()
.ok_or_else(|| {
proto_error(format!(
- "physical_plan::from_proto() Unexpected sort expr {self:?}"
+ "physical_plan::from_proto() Unexpected sort expr {node:?}"
))
})?
.as_ref();
@@ -1746,7 +1762,7 @@ impl protobuf::PhysicalPlanNode {
})
} else {
internal_err!(
- "physical_plan::from_proto() {self:?}"
+ "physical_plan::from_proto() {node:?}"
)
}
})
@@ -1786,6 +1802,7 @@ impl protobuf::PhysicalPlanNode {
ctx: &PhysicalPlanDecodeContext<'_>,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Arc<dyn ExecutionPlan>> {
+ let node = self.node();
let input = into_physical_plan(&sort.input, ctx, proto_converter)?;
let exprs = sort
.expr
@@ -1793,7 +1810,7 @@ impl protobuf::PhysicalPlanNode {
.map(|expr| {
let expr = expr.expr_type.as_ref().ok_or_else(|| {
proto_error(format!(
- "physical_plan::from_proto() Unexpected expr {self:?}"
+ "physical_plan::from_proto() Unexpected expr {node:?}"
))
})?;
if let ExprType::Sort(sort_expr) = expr {
@@ -1802,7 +1819,7 @@ impl protobuf::PhysicalPlanNode {
.as_ref()
.ok_or_else(|| {
proto_error(format!(
- "physical_plan::from_proto() Unexpected sort expr {self:?}"
+ "physical_plan::from_proto() Unexpected sort expr {node:?}"
))
})?
.as_ref();
@@ -1818,7 +1835,7 @@ impl protobuf::PhysicalPlanNode {
},
})
} else {
- internal_err!("physical_plan::from_proto() {self:?}")
+ internal_err!("physical_plan::from_proto() {node:?}")
}
})
.collect::<Result<Vec<_>>>()?;
@@ -1922,7 +1939,7 @@ impl protobuf::PhysicalPlanNode {
left,
right,
filter,
- &join_type.into(),
+ &JoinType::from_proto(join_type),
projection,
)?))
}
@@ -1963,11 +1980,11 @@ impl protobuf::PhysicalPlanNode {
) -> Result<Arc<dyn ExecutionPlan>> {
let input = into_physical_plan(&sink.input, ctx, proto_converter)?;
- let data_sink: JsonSink = sink
- .sink
- .as_ref()
- .ok_or_else(|| proto_error("Missing required field in protobuf"))?
- .try_into()?;
+ let data_sink = JsonSink::try_from_proto(
+ sink.sink
+ .as_ref()
+ .ok_or_else(|| proto_error("Missing required field in protobuf"))?,
+ )?;
let sink_schema = input.schema();
let sort_order = sink
.sort_order
@@ -2000,11 +2017,11 @@ impl protobuf::PhysicalPlanNode {
) -> Result<Arc<dyn ExecutionPlan>> {
let input = into_physical_plan(&sink.input, ctx, proto_converter)?;
- let data_sink: CsvSink = sink
- .sink
- .as_ref()
- .ok_or_else(|| proto_error("Missing required field in protobuf"))?
- .try_into()?;
+ let data_sink = CsvSink::try_from_proto(
+ sink.sink
+ .as_ref()
+ .ok_or_else(|| proto_error("Missing required field in protobuf"))?,
+ )?;
let sink_schema = input.schema();
let sort_order = sink
.sort_order
@@ -2040,11 +2057,11 @@ impl protobuf::PhysicalPlanNode {
{
let input = into_physical_plan(&sink.input, ctx, proto_converter)?;
- let data_sink: ParquetSink = sink
- .sink
- .as_ref()
- .ok_or_else(|| proto_error("Missing required field in protobuf"))?
- .try_into()?;
+ let data_sink = ParquetSink::try_from_proto(
+ sink.sink
+ .as_ref()
+ .ok_or_else(|| proto_error("Missing required field in protobuf"))?,
+ )?;
let sink_schema = input.schema();
let sort_order = sink
.sort_order
@@ -2092,7 +2109,11 @@ impl protobuf::PhysicalPlanNode {
.collect(),
unnest.struct_type_columns.iter().map(|c| *c as _).collect(),
Arc::new(convert_required!(unnest.schema)?),
- into_required!(unnest.options)?,
+ unnest
+ .options
+ .as_ref()
+ .map(datafusion_common::UnnestOptions::from_proto)
+ .ok_or_else(|| proto_error("Missing required field in protobuf"))?,
)?))
}
@@ -2204,9 +2225,9 @@ impl protobuf::PhysicalPlanNode {
right,
on,
filter,
- join_type.into(),
+ JoinType::from_proto(join_type),
sort_options,
- null_equality.into(),
+ NullEquality::from_proto(null_equality),
)?))
}
@@ -2219,7 +2240,9 @@ impl protobuf::PhysicalPlanNode {
let args = match &generate_series.args {
Some(protobuf::generate_series_node::Args::ContainsNull(args)) => {
GenSeriesArgs::ContainsNull {
- name: Self::generate_series_name_to_str(args.name()),
+ name: protobuf::PhysicalPlanNode::generate_series_name_to_str(
+ args.name(),
+ ),
}
}
Some(protobuf::generate_series_node::Args::Int64Args(args)) => {
@@ -2228,7 +2251,9 @@ impl protobuf::PhysicalPlanNode {
end: args.end,
step: args.step,
include_end: args.include_end,
- name: Self::generate_series_name_to_str(args.name()),
+ name: protobuf::PhysicalPlanNode::generate_series_name_to_str(
+ args.name(),
+ ),
}
}
Some(protobuf::generate_series_node::Args::TimestampArgs(args)) => {
@@ -2246,7 +2271,9 @@ impl protobuf::PhysicalPlanNode {
step,
tz: args.tz.as_ref().map(|s| Arc::from(s.as_str())),
include_end: args.include_end,
- name: Self::generate_series_name_to_str(args.name()),
+ name: protobuf::PhysicalPlanNode::generate_series_name_to_str(
+ args.name(),
+ ),
}
}
Some(protobuf::generate_series_node::Args::DateArgs(args)) => {
@@ -2263,7 +2290,9 @@ impl protobuf::PhysicalPlanNode {
end: args.end,
step,
include_end: args.include_end,
- name: Self::generate_series_name_to_str(args.name()),
+ name: protobuf::PhysicalPlanNode::generate_series_name_to_str(
+ args.name(),
+ ),
}
}
None => return internal_err!("Missing args in GenerateSeriesNode"),
@@ -2371,7 +2400,7 @@ impl protobuf::PhysicalPlanNode {
fn try_from_explain_exec(
exec: &ExplainExec,
_codec: &dyn PhysicalExtensionCodec,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
Ok(protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::Explain(
protobuf::ExplainExecNode {
@@ -2379,7 +2408,7 @@ impl protobuf::PhysicalPlanNode {
stringified_plans: exec
.stringified_plans()
.iter()
- .map(|plan| plan.into())
+ .map(protobuf::StringifiedPlan::from_proto)
.collect(),
verbose: exec.verbose(),
},
@@ -2391,7 +2420,7 @@ impl protobuf::PhysicalPlanNode {
exec: &ProjectionExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.input().to_owned(),
codec,
@@ -2424,7 +2453,7 @@ impl protobuf::PhysicalPlanNode {
exec: &AnalyzeExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.input().to_owned(),
codec,
@@ -2452,7 +2481,7 @@ impl protobuf::PhysicalPlanNode {
exec: &FilterExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.input().to_owned(),
codec,
@@ -2484,7 +2513,7 @@ impl protobuf::PhysicalPlanNode {
limit: &GlobalLimitExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
limit.input().to_owned(),
codec,
@@ -2509,7 +2538,7 @@ impl protobuf::PhysicalPlanNode {
limit: &LocalLimitExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
limit.input().to_owned(),
codec,
@@ -2529,7 +2558,7 @@ impl protobuf::PhysicalPlanNode {
exec: &HashJoinExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let left = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.left().to_owned(),
codec,
@@ -2552,8 +2581,8 @@ impl protobuf::PhysicalPlanNode {
})
})
.collect::<Result<_>>()?;
- let join_type: protobuf::JoinType = exec.join_type().to_owned().into();
- let null_equality: protobuf::NullEquality = exec.null_equality().into();
+ let join_type = protobuf::JoinType::from_proto(exec.join_type().to_owned());
+ let null_equality = protobuf::NullEquality::from_proto(exec.null_equality());
let filter = exec
.filter()
.as_ref()
@@ -2619,7 +2648,7 @@ impl protobuf::PhysicalPlanNode {
exec: &SymmetricHashJoinExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let left = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.left().to_owned(),
codec,
@@ -2642,8 +2671,8 @@ impl protobuf::PhysicalPlanNode {
})
})
.collect::<Result<_>>()?;
- let join_type: protobuf::JoinType = exec.join_type().to_owned().into();
- let null_equality: protobuf::NullEquality = exec.null_equality().into();
+ let join_type = protobuf::JoinType::from_proto(exec.join_type().to_owned());
+ let null_equality = protobuf::NullEquality::from_proto(exec.null_equality());
let filter = exec
.filter()
.as_ref()
@@ -2740,7 +2769,7 @@ impl protobuf::PhysicalPlanNode {
exec: &SortMergeJoinExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let left = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.left().to_owned(),
codec,
@@ -2763,8 +2792,8 @@ impl protobuf::PhysicalPlanNode {
})
})
.collect::<Result<_>>()?;
- let join_type: protobuf::JoinType = exec.join_type().to_owned().into();
- let null_equality: protobuf::NullEquality = exec.null_equality().into();
+ let join_type = protobuf::JoinType::from_proto(exec.join_type().to_owned());
+ let null_equality = protobuf::NullEquality::from_proto(exec.null_equality());
let filter = exec
.filter()
.as_ref()
@@ -2827,7 +2856,7 @@ impl protobuf::PhysicalPlanNode {
exec: &CrossJoinExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let left = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.left().to_owned(),
codec,
@@ -2852,7 +2881,7 @@ impl protobuf::PhysicalPlanNode {
exec: &AggregateExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let groups: Vec<bool> = exec
.group_expr()
.groups()
@@ -2955,7 +2984,7 @@ impl protobuf::PhysicalPlanNode {
fn try_from_empty_exec(
empty: &EmptyExec,
_codec: &dyn PhysicalExtensionCodec,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let schema = empty.schema().as_ref().try_into()?;
Ok(protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::Empty(protobuf::EmptyExecNode {
@@ -2967,7 +2996,7 @@ impl protobuf::PhysicalPlanNode {
fn try_from_placeholder_row_exec(
empty: &PlaceholderRowExec,
_codec: &dyn PhysicalExtensionCodec,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let schema = empty.schema().as_ref().try_into()?;
Ok(protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::PlaceholderRow(
@@ -2983,7 +3012,7 @@ impl protobuf::PhysicalPlanNode {
coalesce_batches: &CoalesceBatchesExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
coalesce_batches.input().to_owned(),
codec,
@@ -3004,7 +3033,7 @@ impl protobuf::PhysicalPlanNode {
data_source_exec: &DataSourceExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Option<Self>> {
+ ) -> Result<Option<protobuf::PhysicalPlanNode>> {
let data_source = data_source_exec.data_source();
if let Some(maybe_csv) = data_source.downcast_ref::<FileScanConfig>() {
let source = maybe_csv.file_source();
@@ -3176,7 +3205,7 @@ impl protobuf::PhysicalPlanNode {
exec: &CoalescePartitionsExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.input().to_owned(),
codec,
@@ -3196,7 +3225,7 @@ impl protobuf::PhysicalPlanNode {
exec: &RepartitionExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.input().to_owned(),
codec,
@@ -3221,7 +3250,7 @@ impl protobuf::PhysicalPlanNode {
exec: &SortExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = proto_converter.execution_plan_to_proto(exec.input(), codec)?;
let expr = exec
.expr()
@@ -3268,7 +3297,7 @@ impl protobuf::PhysicalPlanNode {
union: &UnionExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let mut inputs: Vec<protobuf::PhysicalPlanNode> = vec![];
for input in union.inputs() {
inputs.push(
@@ -3290,7 +3319,7 @@ impl protobuf::PhysicalPlanNode {
interleave: &InterleaveExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let mut inputs: Vec<protobuf::PhysicalPlanNode> = vec![];
for input in interleave.inputs() {
inputs.push(
@@ -3312,7 +3341,7 @@ impl protobuf::PhysicalPlanNode {
exec: &SortPreservingMergeExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.input().to_owned(),
codec,
@@ -3350,7 +3379,7 @@ impl protobuf::PhysicalPlanNode {
exec: &NestedLoopJoinExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let left = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.left().to_owned(),
codec,
@@ -3362,7 +3391,7 @@ impl protobuf::PhysicalPlanNode {
proto_converter,
)?;
- let join_type: protobuf::JoinType = exec.join_type().to_owned().into();
+ let join_type = protobuf::JoinType::from_proto(exec.join_type().to_owned());
let filter = exec
.filter()
.as_ref()
@@ -3408,7 +3437,7 @@ impl protobuf::PhysicalPlanNode {
exec: &WindowAggExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.input().to_owned(),
codec,
@@ -3443,7 +3472,7 @@ impl protobuf::PhysicalPlanNode {
exec: &BoundedWindowAggExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.input().to_owned(),
codec,
@@ -3494,7 +3523,7 @@ impl protobuf::PhysicalPlanNode {
exec: &DataSinkExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Option<Self>> {
+ ) -> Result<Option<protobuf::PhysicalPlanNode>> {
let input: protobuf::PhysicalPlanNode =
protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.input().to_owned(),
@@ -3530,7 +3559,7 @@ impl protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::JsonSink(Box::new(
protobuf::JsonSinkExecNode {
input: Some(Box::new(input)),
- sink: Some(sink.try_into()?),
+ sink: Some(protobuf::JsonSink::try_from_proto(sink)?),
sink_schema: Some(exec.schema().as_ref().try_into()?),
sort_order,
},
@@ -3543,7 +3572,7 @@ impl protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::CsvSink(Box::new(
protobuf::CsvSinkExecNode {
input: Some(Box::new(input)),
- sink: Some(sink.try_into()?),
+ sink: Some(protobuf::CsvSink::try_from_proto(sink)?),
sink_schema: Some(exec.schema().as_ref().try_into()?),
sort_order,
},
@@ -3557,7 +3586,7 @@ impl protobuf::PhysicalPlanNode {
physical_plan_type: Some(PhysicalPlanType::ParquetSink(Box::new(
protobuf::ParquetSinkExecNode {
input: Some(Box::new(input)),
- sink: Some(sink.try_into()?),
+ sink: Some(protobuf::ParquetSink::try_from_proto(sink)?),
sink_schema: Some(exec.schema().as_ref().try_into()?),
sort_order,
},
@@ -3573,7 +3602,7 @@ impl protobuf::PhysicalPlanNode {
exec: &UnnestExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.input().to_owned(),
codec,
@@ -3598,7 +3627,7 @@ impl protobuf::PhysicalPlanNode {
.iter()
.map(|c| *c as _)
.collect(),
- options: Some(exec.options().into()),
+ options: Some(protobuf::UnnestOptions::from_proto(exec.options())),
},
))),
})
@@ -3608,7 +3637,7 @@ impl protobuf::PhysicalPlanNode {
exec: &CooperativeExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
exec.input().to_owned(),
codec,
@@ -3632,7 +3661,9 @@ impl protobuf::PhysicalPlanNode {
}
}
- fn try_from_lazy_memory_exec(exec: &LazyMemoryExec) -> Result<Option<Self>> {
+ fn try_from_lazy_memory_exec(
+ exec: &LazyMemoryExec,
+ ) -> Result<Option<protobuf::PhysicalPlanNode>> {
let generators = exec.generators();
// ensure we only have one generator
@@ -3650,7 +3681,9 @@ impl protobuf::PhysicalPlanNode {
target_batch_size: 8192, // Default batch size
args: Some(protobuf::generate_series_node::Args::ContainsNull(
protobuf::GenerateSeriesArgsContainsNull {
- name: Self::str_to_generate_series_name(empty_gen.name())? as i32,
+ name: protobuf::PhysicalPlanNode::str_to_generate_series_name(
+ empty_gen.name(),
+ )? as i32,
},
)),
};
@@ -3674,7 +3707,9 @@ impl protobuf::PhysicalPlanNode {
end: *int_64.end(),
step: *int_64.step(),
include_end: int_64.include_end(),
- name: Self::str_to_generate_series_name(int_64.name())? as i32,
+ name: protobuf::PhysicalPlanNode::str_to_generate_series_name(
+ int_64.name(),
+ )? as i32,
},
)),
};
@@ -3701,7 +3736,9 @@ impl protobuf::PhysicalPlanNode {
nanos: step_value.nanoseconds,
});
let include_end = timestamp_args.include_end();
- let name = Self::str_to_generate_series_name(timestamp_args.name())? as i32;
+ let name = protobuf::PhysicalPlanNode::str_to_generate_series_name(
+ timestamp_args.name(),
+ )? as i32;
let args = match timestamp_args.current().tz_str() {
Some(tz) => protobuf::generate_series_node::Args::TimestampArgs(
@@ -3743,7 +3780,7 @@ impl protobuf::PhysicalPlanNode {
exec: &AsyncFuncExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
Arc::clone(exec.input()),
codec,
@@ -3774,7 +3811,7 @@ impl protobuf::PhysicalPlanNode {
exec: &BufferExec,
extension_codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
Arc::clone(exec.input()),
extension_codec,
@@ -3795,7 +3832,7 @@ impl protobuf::PhysicalPlanNode {
exec: &ScalarSubqueryExec,
codec: &dyn PhysicalExtensionCodec,
proto_converter: &dyn PhysicalProtoConverterExtension,
- ) -> Result<Self> {
+ ) -> Result<protobuf::PhysicalPlanNode> {
let input = protobuf::PhysicalPlanNode::try_from_physical_plan_with_converter(
Arc::clone(exec.input()),
codec,
@@ -3824,6 +3861,12 @@ impl protobuf::PhysicalPlanNode {
}
}
+impl PhysicalPlanNodeExt for protobuf::PhysicalPlanNode {
+ fn node(&self) -> &protobuf::PhysicalPlanNode {
+ self
+ }
+}
+
pub trait AsExecutionPlan: Debug + Send + Sync + Clone {
fn try_decode(buf: &[u8]) -> Result<Self>
where
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs b/datafusion/proto/src/physical_plan/to_proto.rs
index ec8e168178..5181c97401 100644
--- a/datafusion/proto/src/physical_plan/to_proto.rs
+++ b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -36,9 +36,8 @@ use datafusion_physical_expr::scalar_subquery::ScalarSubqueryExpr;
use datafusion_physical_expr::window::{SlidingAggregateWindowExpr, StandardWindowExpr};
use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
use datafusion_physical_plan::expressions::{
- BinaryExpr, CaseExpr, CastExpr, Column, DynamicFilterPhysicalExpr, InListExpr,
- IsNotNullExpr, IsNullExpr, LikeExpr, Literal, NegativeExpr, NotExpr, TryCastExpr,
- UnKnownColumn,
+ CaseExpr, CastExpr, DynamicFilterPhysicalExpr, InListExpr, IsNotNullExpr, IsNullExpr,
+ LikeExpr, Literal, NegativeExpr, NotExpr, TryCastExpr, UnKnownColumn,
};
use datafusion_physical_plan::joins::{HashExpr, HashTableLookupExpr};
use datafusion_physical_plan::udaf::AggregateFunctionExpr;
@@ -49,6 +48,7 @@ use super::{
DefaultPhysicalProtoConverter, PhysicalExtensionCodec,
PhysicalProtoConverterExtension, encode_human_display_alias,
};
+use crate::convert::TryFromProto;
use crate::protobuf::{
self, PhysicalSortExprNode, PhysicalSortExprNodeCollection,
physical_aggregate_expr_node, physical_window_expr_node,
@@ -178,9 +178,7 @@ pub fn serialize_physical_window_expr(
codec,
proto_converter,
)?;
- let window_frame: protobuf::WindowFrame = window_frame
- .as_ref()
- .try_into()
+ let window_frame = protobuf::WindowFrame::try_from_proto(window_frame.as_ref())
.map_err(|e| internal_datafusion_err!("{e}"))?;
Ok(protobuf::PhysicalWindowExprNode {
@@ -253,6 +251,29 @@ pub fn serialize_physical_expr(
)
}
+/// Concrete [`PhysicalExprEncode`] driver used to back
+/// [`PhysicalExprEncodeCtx`] when expressions invoke `PhysicalExpr::to_proto`.
+///
+/// Wraps the existing extension codec + converter pair so individual
+/// expressions can recurse into children without depending on
+/// `datafusion-proto` directly.
+///
+/// [`PhysicalExprEncode`]: datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncode
+/// [`PhysicalExprEncodeCtx`]: datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx
+struct ConverterEncoder<'a> {
+ codec: &'a dyn PhysicalExtensionCodec,
+ proto_converter: &'a dyn PhysicalProtoConverterExtension,
+}
+
+impl datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncode
+ for ConverterEncoder<'_>
+{
+ fn encode(&self, expr: &Arc<dyn PhysicalExpr>) -> Result<protobuf::PhysicalExprNode> {
+ self.proto_converter
+ .physical_expr_to_proto(expr, self.codec)
+ }
+}
+
/// Serialize a `PhysicalExpr` to default protobuf representation.
///
/// If required, a [`PhysicalExtensionCodec`] can be provided which can handle
@@ -266,6 +287,21 @@ pub fn serialize_physical_expr_with_converter(
) -> Result<protobuf::PhysicalExprNode> {
let expr = value.as_ref();
let expr_id = value.expression_id();
+
+ // Give the expression a chance to serialize itself first. Returning
+ // `Ok(Some(node))` lets expressions with private state (e.g.
+ // `DynamicFilterPhysicalExpr`) avoid exposing pub-for-proto accessors.
+ // `Ok(None)` falls through to the downcast chain below — that's the
+ // default for built-in expressions which haven't been migrated yet.
+ let encoder = ConverterEncoder {
+ codec,
+ proto_converter,
+ };
+ let ctx = datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx::new(&encoder);
+ if let Some(node) = expr.try_to_proto(&ctx)? {
+ return Ok(node);
+ }
+
// HashTableLookupExpr is used for dynamic filter pushdown in hash joins.
// It contains an Arc<dyn JoinHashMapType> (the build-side hash table) which
// cannot be serialized - the hash table is a runtime structure built during
@@ -291,17 +327,7 @@ pub fn serialize_physical_expr_with_converter(
});
}
- if let Some(expr) = expr.downcast_ref::<Column>() {
- Ok(protobuf::PhysicalExprNode {
- expr_id,
- expr_type: Some(protobuf::physical_expr_node::ExprType::Column(
- protobuf::PhysicalColumn {
- name: expr.name().to_string(),
- index: expr.index() as u32,
- },
- )),
- })
- } else if let Some(expr) = expr.downcast_ref::<UnKnownColumn>() {
+ if let Some(expr) = expr.downcast_ref::<UnKnownColumn>() {
Ok(protobuf::PhysicalExprNode {
expr_id,
expr_type: Some(protobuf::physical_expr_node::ExprType::UnknownColumn(
@@ -310,46 +336,6 @@ pub fn serialize_physical_expr_with_converter(
},
)),
})
- } else if let Some(expr) = expr.downcast_ref::<BinaryExpr>() {
- // Linearize a nested binary expression tree of the same operator
- // into a flat vector of operands to avoid deep recursion in proto.
- let op = expr.op();
- let mut operand_refs: Vec<&Arc<dyn PhysicalExpr>> = vec![expr.right()];
- let mut current_expr: &BinaryExpr = expr;
- loop {
- match current_expr.left().downcast_ref::<BinaryExpr>() {
- Some(bin) if bin.op() == op => {
- operand_refs.push(bin.right());
- current_expr = bin;
- }
- _ => {
- operand_refs.push(current_expr.left());
- break;
- }
- }
- }
-
- // Reverse so operands are ordered from left innermost to right outermost
- operand_refs.reverse();
-
- let operands = operand_refs
- .iter()
- .map(|e| proto_converter.physical_expr_to_proto(e, codec))
- .collect::<Result<Vec<_>>>()?;
-
- let binary_expr = Box::new(protobuf::PhysicalBinaryExprNode {
- l: None,
- r: None,
- op: format!("{:?}", op),
- operands,
- });
-
- Ok(protobuf::PhysicalExprNode {
- expr_id,
- expr_type: Some(protobuf::physical_expr_node::ExprType::BinaryExpr(
- binary_expr,
- )),
- })
} else if let Some(expr) = expr.downcast_ref::<CaseExpr>() {
Ok(protobuf::PhysicalExprNode {
expr_id,
@@ -642,10 +628,10 @@ fn serialize_when_then_expr(
})
}
-impl TryFrom<&PartitionedFile> for protobuf::PartitionedFile {
+impl TryFromProto<&PartitionedFile> for protobuf::PartitionedFile {
type Error = DataFusionError;
- fn try_from(pf: &PartitionedFile) -> Result<Self> {
+ 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!(
@@ -661,16 +647,20 @@ impl TryFrom<&PartitionedFile> for protobuf::PartitionedFile {
.iter()
.map(|v| v.try_into())
.collect::<Result<Vec<_>, _>>()?,
- range: pf.range.as_ref().map(|r| r.try_into()).transpose()?,
+ range: pf
+ .range
+ .as_ref()
+ .map(protobuf::FileRange::try_from_proto)
+ .transpose()?,
statistics: pf.statistics.as_ref().map(|s| s.as_ref().into()),
})
}
}
-impl TryFrom<&FileRange> for protobuf::FileRange {
+impl TryFromProto<&FileRange> for protobuf::FileRange {
type Error = DataFusionError;
- fn try_from(value: &FileRange) -> Result<Self> {
+ fn try_from_proto(value: &FileRange) -> Result<Self> {
Ok(protobuf::FileRange {
start: value.start,
end: value.end,
@@ -678,14 +668,14 @@ impl TryFrom<&FileRange> for protobuf::FileRange {
}
}
-impl TryFrom<&[PartitionedFile]> for protobuf::FileGroup {
+impl TryFromProto<&[PartitionedFile]> for protobuf::FileGroup {
type Error = DataFusionError;
- fn try_from(gr: &[PartitionedFile]) -> Result<Self, Self::Error> {
+ fn try_from_proto(gr: &[PartitionedFile]) -> Result<Self, Self::Error> {
Ok(protobuf::FileGroup {
files: gr
.iter()
- .map(|f| f.try_into())
+ .map(protobuf::PartitionedFile::try_from_proto)
.collect::<Result<Vec<_>, _>>()?,
})
}
@@ -699,7 +689,7 @@ pub fn serialize_file_scan_config(
let file_groups = conf
.file_groups
.iter()
- .map(|p| p.files().try_into())
+ .map(|p| protobuf::FileGroup::try_from_proto(p.files()))
.collect::<Result<Vec<_>, _>>()?;
let mut output_orderings = vec![];
@@ -797,48 +787,48 @@ pub fn serialize_record_batches(batches: &[RecordBatch]) -> Result<Vec<u8>> {
Ok(buf)
}
-impl TryFrom<&JsonSink> for protobuf::JsonSink {
+impl TryFromProto<&JsonSink> for protobuf::JsonSink {
type Error = DataFusionError;
- fn try_from(value: &JsonSink) -> Result<Self, Self::Error> {
+ fn try_from_proto(value: &JsonSink) -> Result<Self, Self::Error> {
Ok(Self {
- config: Some(value.config().try_into()?),
+ config: Some(protobuf::FileSinkConfig::try_from_proto(value.config())?),
writer_options: Some(value.writer_options().try_into()?),
})
}
}
-impl TryFrom<&CsvSink> for protobuf::CsvSink {
+impl TryFromProto<&CsvSink> for protobuf::CsvSink {
type Error = DataFusionError;
- fn try_from(value: &CsvSink) -> Result<Self, Self::Error> {
+ fn try_from_proto(value: &CsvSink) -> Result<Self, Self::Error> {
Ok(Self {
- config: Some(value.config().try_into()?),
+ config: Some(protobuf::FileSinkConfig::try_from_proto(value.config())?),
writer_options: Some(value.writer_options().try_into()?),
})
}
}
#[cfg(feature = "parquet")]
-impl TryFrom<&ParquetSink> for protobuf::ParquetSink {
+impl TryFromProto<&ParquetSink> for protobuf::ParquetSink {
type Error = DataFusionError;
- fn try_from(value: &ParquetSink) -> Result<Self, Self::Error> {
+ fn try_from_proto(value: &ParquetSink) -> Result<Self, Self::Error> {
Ok(Self {
- config: Some(value.config().try_into()?),
+ config: Some(protobuf::FileSinkConfig::try_from_proto(value.config())?),
parquet_options: Some(value.parquet_options().try_into()?),
})
}
}
-impl TryFrom<&FileSinkConfig> for protobuf::FileSinkConfig {
+impl TryFromProto<&FileSinkConfig> for protobuf::FileSinkConfig {
type Error = DataFusionError;
- fn try_from(conf: &FileSinkConfig) -> Result<Self, Self::Error> {
+ fn try_from_proto(conf: &FileSinkConfig) -> Result<Self, Self::Error> {
let file_groups = conf
.file_group
.iter()
- .map(TryInto::try_into)
+ .map(protobuf::PartitionedFile::try_from_proto)
.collect::<Result<Vec<_>>>()?;
let table_paths = conf
.table_paths
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index c2b62c1745..58b79d641b 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -128,7 +128,7 @@ use datafusion_proto::physical_plan::to_proto::serialize_physical_expr_with_conv
use datafusion_proto::physical_plan::{
AsExecutionPlan, DeduplicatingProtoConverter, DefaultPhysicalExtensionCodec,
DefaultPhysicalProtoConverter, PhysicalExtensionCodec, PhysicalPlanDecodeContext,
- PhysicalProtoConverterExtension,
+ PhysicalPlanNodeExt, PhysicalProtoConverterExtension,
};
use datafusion_proto::protobuf;
use datafusion_proto::protobuf::{PhysicalExprNode, PhysicalPlanNode};
diff --git a/dev/release/rat_exclude_files.txt b/dev/release/rat_exclude_files.txt
index f5ce368df7..77da7db87e 100644
--- a/dev/release/rat_exclude_files.txt
+++ b/dev/release/rat_exclude_files.txt
@@ -52,9 +52,9 @@ Cargo.lock
.history
parquet-testing/*
*rat.txt
-datafusion/proto/src/generated/datafusion_proto_common.rs
-datafusion/proto/src/generated/pbjson.rs
-datafusion/proto/src/generated/prost.rs
+datafusion/proto-models/src/generated/datafusion_proto_common.rs
+datafusion/proto-models/src/generated/pbjson.rs
+datafusion/proto-models/src/generated/prost.rs
datafusion/proto-common/src/generated/pbjson.rs
datafusion/proto-common/src/generated/prost.rs
.github/ISSUE_TEMPLATE/bug_report.yml
diff --git a/licenserc.toml b/licenserc.toml
index 105d969ea5..a1e01a5fd0 100644
--- a/licenserc.toml
+++ b/licenserc.toml
@@ -27,4 +27,5 @@ excludes = [
# generated code
"datafusion/proto/src/generated/",
"datafusion/proto-common/src/generated/",
+ "datafusion/proto-models/src/generated/",
]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org