stuhood commented on code in PR #24766:
URL: https://github.com/apache/datafusion/pull/24766#discussion_r4009549520
##########
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:
So, the case where we don't have enough samples to scale up to a particular
`target_partition` count should be treated more like an expected case rather
than a failure. Ideally we would always have enough samples to do this... but
in cases where we don't (small tables, etc), it is definitely a recoverable
case, and callers should convert the `Partitioning::Range` into something
else... e.g. `Partitioning::Unknown`.
So rather than returning `Result<Self>`, maybe this should return
`Option<Self>`? Alternatively, it could return a `Result<Self, ScalingError>`
(name TBD) that allows the caller to differentiate failure cases.
--
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]
