Copilot commented on code in PR #24003:
URL: https://github.com/apache/datafusion/pull/24003#discussion_r3684224228
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -515,6 +519,138 @@ impl Partitioning {
}
}
+/// Protobuf conversions for [`Partitioning`].
+///
+/// Child expressions (hash keys, range orderings) and `ScalarValue` split
+/// points are (de)serialized through the expression-level context, so this is
+/// the single copy of the partitioning wire format: `RepartitionExec`,
+/// `FileScanConfig` and `datafusion-proto`'s central serializer all route
+/// through it.
+///
+/// [`protobuf::Partitioning`]: datafusion_proto_models::protobuf::Partitioning
+#[cfg(feature = "proto")]
+impl Partitioning {
+ /// Serialize this partitioning into its protobuf representation.
+ pub fn try_to_proto(
+ &self,
+ ctx:
&datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
+ ) -> Result<datafusion_proto_models::protobuf::Partitioning> {
+ use datafusion_proto_models::protobuf;
+
+ let partition_method = match self {
+ Partitioning::RoundRobinBatch(n) => {
+ protobuf::partitioning::PartitionMethod::RoundRobin(*n as u64)
+ }
+ Partitioning::Hash(exprs, n) => {
+ protobuf::partitioning::PartitionMethod::Hash(
+ protobuf::PhysicalHashRepartition {
+ hash_expr: ctx.encode_children_expressions(exprs)?,
+ partition_count: *n as u64,
+ },
+ )
+ }
Review Comment:
The protobuf encoding path casts `usize` partition counts to `u64` with
`as`, which can silently truncate on platforms where `usize` is wider than
`u64`. Since decode now explicitly errors on out-of-range counts, encode should
similarly validate (e.g., `u64::try_from(*n)` /
`u64::try_from(partition_count)` with an internal error) to avoid generating
lossy/invalid wire data.
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -515,6 +519,138 @@ impl Partitioning {
}
}
+/// Protobuf conversions for [`Partitioning`].
+///
+/// Child expressions (hash keys, range orderings) and `ScalarValue` split
+/// points are (de)serialized through the expression-level context, so this is
+/// the single copy of the partitioning wire format: `RepartitionExec`,
+/// `FileScanConfig` and `datafusion-proto`'s central serializer all route
+/// through it.
Review Comment:
This comment states that `FileScanConfig` already routes through this
implementation, but the diff shown in this PR doesn’t update `FileScanConfig`
yet. Consider rewording to avoid claiming current behavior (e.g., remove
`FileScanConfig` here or phrase it as an intended/next-step consumer) so the
documentation matches the code in this PR.
##########
datafusion/physical-plan/src/proto.rs:
##########
@@ -190,6 +196,25 @@ impl<'a> ExecutionPlanEncodeCtx<'a> {
pub fn encode_udwf(&self, udwf: &WindowUDF) -> Result<Option<Vec<u8>>> {
self.encoder.encode_udwf(udwf)
}
+
+ /// An expression-level encode context backed by this plan context.
+ ///
+ /// Lets a plan hand `ctx` to expression-level conversions that own their
own
+ /// wire logic — e.g.
+ ///
[`Partitioning::try_to_proto`](datafusion_physical_expr::Partitioning::try_to_proto)
+ /// and
+ ///
[`PhysicalSortExpr::try_to_proto`](datafusion_physical_expr::PhysicalSortExpr::try_to_proto).
+ pub fn expr_ctx(&self) -> PhysicalExprEncodeCtx<'_> {
+ PhysicalExprEncodeCtx::new(self)
+ }
+}
+
+/// Lets [`ExecutionPlanEncodeCtx`] back an [`PhysicalExprEncodeCtx`], so
+/// expression-level conversions can be reused from plan hooks.
Review Comment:
Grammar: use 'a' rather than 'an' before `PhysicalExprEncodeCtx` /
`PhysicalExprDecodeCtx` ('a Physical…').
##########
datafusion/physical-plan/src/proto.rs:
##########
@@ -286,6 +311,28 @@ impl<'a> ExecutionPlanDecodeCtx<'a> {
) -> Result<Arc<WindowUDF>> {
self.decoder.decode_udwf(name, payload)
}
+
+ /// An expression-level decode context backed by this plan context, bound
to
+ /// `input_schema`.
+ ///
+ /// The decode counterpart of
+ /// [`ExecutionPlanEncodeCtx::expr_ctx`], for calling conversions such as
+ ///
[`Partitioning::try_from_proto`](datafusion_physical_expr::Partitioning::try_from_proto).
+ pub fn expr_ctx<'s>(&'s self, input_schema: &'s Schema) ->
PhysicalExprDecodeCtx<'s> {
+ PhysicalExprDecodeCtx::new(input_schema, self)
+ }
+}
+
+/// Lets [`ExecutionPlanDecodeCtx`] back a [`PhysicalExprDecodeCtx`], so
+/// expression-level conversions can be reused from plan hooks.
Review Comment:
Grammar: use 'a' rather than 'an' before `PhysicalExprEncodeCtx` /
`PhysicalExprDecodeCtx` ('a Physical…').
##########
datafusion/physical-plan/src/proto.rs:
##########
@@ -190,6 +196,25 @@ impl<'a> ExecutionPlanEncodeCtx<'a> {
pub fn encode_udwf(&self, udwf: &WindowUDF) -> Result<Option<Vec<u8>>> {
self.encoder.encode_udwf(udwf)
}
+
+ /// An expression-level encode context backed by this plan context.
+ ///
+ /// Lets a plan hand `ctx` to expression-level conversions that own their
own
+ /// wire logic — e.g.
+ ///
[`Partitioning::try_to_proto`](datafusion_physical_expr::Partitioning::try_to_proto)
+ /// and
+ ///
[`PhysicalSortExpr::try_to_proto`](datafusion_physical_expr::PhysicalSortExpr::try_to_proto).
Review Comment:
The intra-doc link for `PhysicalSortExpr::try_to_proto` appears to point at
`datafusion_physical_expr::PhysicalSortExpr`, but the method is introduced on
`PhysicalSortExpr` in `datafusion_physical_expr_common::sort_expr`. If
`PhysicalSortExpr` isn’t re-exported from `datafusion_physical_expr`, this will
create a broken intra-doc link (often CI-failing under
`rustdoc::broken_intra_doc_links`). Update the link target to the correct path
or ensure the type is re-exported where the link points.
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -515,6 +519,138 @@ impl Partitioning {
}
}
+/// Protobuf conversions for [`Partitioning`].
+///
+/// Child expressions (hash keys, range orderings) and `ScalarValue` split
+/// points are (de)serialized through the expression-level context, so this is
+/// the single copy of the partitioning wire format: `RepartitionExec`,
+/// `FileScanConfig` and `datafusion-proto`'s central serializer all route
+/// through it.
+///
+/// [`protobuf::Partitioning`]: datafusion_proto_models::protobuf::Partitioning
+#[cfg(feature = "proto")]
+impl Partitioning {
+ /// Serialize this partitioning into its protobuf representation.
+ pub fn try_to_proto(
+ &self,
+ ctx:
&datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
+ ) -> Result<datafusion_proto_models::protobuf::Partitioning> {
+ use datafusion_proto_models::protobuf;
+
+ let partition_method = match self {
+ Partitioning::RoundRobinBatch(n) => {
+ protobuf::partitioning::PartitionMethod::RoundRobin(*n as u64)
+ }
+ Partitioning::Hash(exprs, n) => {
+ protobuf::partitioning::PartitionMethod::Hash(
+ protobuf::PhysicalHashRepartition {
+ hash_expr: ctx.encode_children_expressions(exprs)?,
+ partition_count: *n as u64,
+ },
+ )
+ }
+ Partitioning::Range(range) => {
+ let sort_expr =
sort_exprs_try_to_proto(range.ordering().iter(), ctx)?;
+ let split_point = range
+ .split_points()
+ .iter()
+ .map(|split_point| {
+ let value = split_point
+ .values()
+ .iter()
+ .map(|value| value.try_into().map_err(Into::into))
+ .collect::<Result<Vec<_>>>()?;
+ Ok(protobuf::PhysicalRangeSplitPoint { value })
+ })
+ .collect::<Result<Vec<_>>>()?;
+ protobuf::partitioning::PartitionMethod::Range(
+ protobuf::PhysicalRangePartitioning {
+ sort_expr,
+ split_point,
+ },
+ )
+ }
+ Partitioning::UnknownPartitioning(n) => {
+ protobuf::partitioning::PartitionMethod::Unknown(*n as u64)
+ }
+ };
Review Comment:
The protobuf encoding path casts `usize` partition counts to `u64` with
`as`, which can silently truncate on platforms where `usize` is wider than
`u64`. Since decode now explicitly errors on out-of-range counts, encode should
similarly validate (e.g., `u64::try_from(*n)` /
`u64::try_from(partition_count)` with an internal error) to avoid generating
lossy/invalid wire data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
