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>`
that allows the caller to differentiate failure cases.
##########
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> {
Review Comment:
This looks good. But I think that we should go ahead and deprecate the old
constructor, and have all existing callers switch to a method like
`RangePartitioning::try_new_with_samples`. `RangePartitioning::new` will create
something which cannot be scaled, and that's much less useful.
In terms of the number of samples for callers to provide to
`try_new_with_samples`: we might want to recommend something like `K *
target_partitions`, where `K` is some value that we expect will give us enough
flexibility to actually scale things for the relevant number of partitions.
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -205,8 +211,12 @@ impl Display for Partitioning {
pub struct RangePartitioning {
/// Ordered partitioning key.
ordering: LexOrdering,
- /// Boundaries between adjacent partitions.
- split_points: Vec<SplitPoint>,
+ /// Maximum-resolution boundaries used to derive split points.
+ samples: Arc<[SplitPoint]>,
+ /// Effective boundaries for the current partition count.
+ split_points: Arc<[SplitPoint]>,
+ /// Number of effective partitions.
+ partition_count: usize,
Review Comment:
The `partition_count` can be derived from `split_points.len() + 1`, so it
doesn't need to be stored.
##########
datafusion/physical-expr/src/partitioning.rs:
##########
Review Comment:
The fact that `PartialEq` is derived here is going to confuse callers:
`enforce_distribution_relationships` uses equality to check whether two
`RangePartitioning` instances are compatible. The `samples` should not actually
be included in that equality check: only the `split_points`.
There are two schools of though on that, and I don't know which DataFusion
prefers:
* School 1: `PartialEq` should always represent structural equality, and so
should basically always be derived. Other definitions of equality should be
provided by other methods.
* School 2: `PartialEq` should be the most useful definition of equality,
even if that is not structural.
I expect that @gene-bordegaray has an opinion on this one.
--
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]
