kumarUjjawal commented on code in PR #24003:
URL: https://github.com/apache/datafusion/pull/24003#discussion_r3684264814
##########
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:
Could we make it delegate to the new shared decoder, or use checked
conversion, so direct callers do not panic on oversized counts on 32-bit
targets?
--
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]
