adriangb commented on code in PR #24003:
URL: https://github.com/apache/datafusion/pull/24003#discussion_r3685606915
##########
datafusion/proto/src/physical_plan/from_proto.rs:
##########
@@ -426,83 +422,20 @@ pub fn parse_protobuf_partitioning(
input_schema: &Schema,
proto_converter: &dyn PhysicalProtoConverterExtension,
) -> Result<Option<Partitioning>> {
- match partitioning {
- Some(protobuf::Partitioning { partition_method }) => match
partition_method {
- Some(protobuf::partitioning::PartitionMethod::RoundRobin(
- partition_count,
- )) => Ok(Some(Partitioning::RoundRobinBatch(
- *partition_count as usize,
- ))),
-
Some(protobuf::partitioning::PartitionMethod::Hash(hash_repartition)) => {
- parse_protobuf_hash_partitioning(
- Some(hash_repartition),
- ctx,
- input_schema,
- proto_converter,
- )
- }
-
Some(protobuf::partitioning::PartitionMethod::Range(range_partitioning)) => {
- Ok(Some(parse_protobuf_range_partitioning(
- range_partitioning,
- ctx,
- input_schema,
- proto_converter,
- )?))
- }
-
Some(protobuf::partitioning::PartitionMethod::Unknown(partition_count)) => {
- Ok(Some(Partitioning::UnknownPartitioning(
- *partition_count as usize,
- )))
- }
- None => Ok(None),
- },
- None => Ok(None),
- }
-}
-
-fn parse_protobuf_range_partitioning(
- range_partitioning: &protobuf::PhysicalRangePartitioning,
- ctx: &PhysicalPlanDecodeContext<'_>,
- input_schema: &Schema,
- proto_converter: &dyn PhysicalProtoConverterExtension,
-) -> Result<Partitioning> {
- let sort_exprs = parse_physical_sort_exprs(
- &range_partitioning.sort_expr,
+ let decoder = ConverterDecoder {
ctx,
- input_schema,
proto_converter,
- )?;
- let sort_expr_count = sort_exprs.len();
- let ordering = LexOrdering::new(sort_exprs).ok_or_else(|| {
- internal_datafusion_err!("Range partitioning requires non-empty
ordering")
- })?;
- if ordering.len() != sort_expr_count {
- return Err(internal_datafusion_err!(
- "Range partitioning ordering must not contain duplicate
expressions"
- ));
- }
- let split_points = range_partitioning
- .split_point
- .iter()
- .map(parse_protobuf_range_split_point)
- .collect::<Result<_>>()?;
- Ok(Partitioning::Range(RangePartitioning::try_new(
- ordering,
- split_points,
- )?))
-}
-
-fn parse_protobuf_range_split_point(
- split_point: &protobuf::PhysicalRangeSplitPoint,
-) -> Result<SplitPoint> {
- let values = split_point
- .value
- .iter()
- .map(|value| ScalarValue::try_from(value).map_err(Into::into))
- .collect::<Result<_>>()?;
- Ok(SplitPoint::new(values))
+ };
+ let decode_ctx =
+
datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx::new(
+ input_schema,
+ &decoder,
+ );
+ partitioning
Review Comment:
Good catch, done. `parse_protobuf_hash_partitioning` now delegates to the
shared `Partitioning::try_from_proto`. I also made the symmetric fix on the
encode side: `Partitioning::try_to_proto` now widens through a checked
`wire_partition_count` helper instead of `as u6`
--
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]
