goutamadwant commented on code in PR #24766:
URL: https://github.com/apache/datafusion/pull/24766#discussion_r4011135687
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -234,19 +248,75 @@ impl RangePartitioning {
Ok(Self::new(ordering, split_points))
}
+ /// Creates sample-backed range partitioning and validates the sample
shape,
+ /// ordering, and target partition count.
+ ///
+ /// `partition_count` must be at least one and no larger than
+ /// `samples.len() + 1`. When it is smaller than that maximum, the samples
+ /// are evenly down-sampled to derive the effective split points.
+ pub fn try_new_with_samples(
+ ordering: LexOrdering,
+ samples: Vec<SplitPoint>,
+ partition_count: usize,
+ ) -> Result<Self> {
+ validate_range_split_points(
+ &samples,
+ &ordering
+ .iter()
+ .map(|sort_expr| sort_expr.options)
+ .collect::<Vec<_>>(),
+ )?;
+ validate_range_partition_count(partition_count, samples.len() + 1)?;
+ let samples: Arc<[SplitPoint]> = Arc::from(samples);
+ let split_points = downsample_split_points(&samples, partition_count);
+ Ok(Self {
+ ordering,
+ samples,
+ split_points,
+ partition_count,
+ })
+ }
+
/// Returns the ordering that defines the range key.
pub fn ordering(&self) -> &LexOrdering {
&self.ordering
}
- /// Returns the ordered split points between partitions.
+ /// Returns the maximum-resolution sample points.
+ pub fn samples(&self) -> &[SplitPoint] {
+ &self.samples
+ }
+
+ /// Returns the effective split points between partitions.
pub fn split_points(&self) -> &[SplitPoint] {
&self.split_points
}
/// Returns the number of partitions.
pub fn partition_count(&self) -> usize {
- self.split_points.len() + 1
+ self.partition_count
+ }
+
+ /// Returns the largest partition count supported by the stored samples.
+ pub fn max_partition_count(&self) -> usize {
+ self.samples.len() + 1
+ }
+
+ /// Returns this range partitioning scaled to `target_partitions`.
+ ///
+ /// Scaling retains the original samples, so a range partitioning that was
+ /// scaled down can later be scaled back up to
[`Self::max_partition_count`].
+ pub fn scale(&self, target_partitions: usize) -> Result<Self> {
+ validate_range_partition_count(target_partitions,
self.max_partition_count())?;
+ if target_partitions == self.partition_count {
+ return Ok(self.clone());
+ }
+ Ok(Self {
+ ordering: self.ordering.clone(),
+ samples: Arc::clone(&self.samples),
+ split_points: downsample_split_points(&self.samples,
target_partitions),
+ partition_count: target_partitions,
+ })
}
Review Comment:
@stuhood changed scale to return a typed RangePartitioningScaleError,
distinguishing InsufficientSamples from ZeroPartitions. EnsureRequirements
retains a satisfying range when capacity is insufficient; other callers can
choose their own fallback without parsing error strings.
--
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]
