gene-bordegaray commented on code in PR #24766:
URL: https://github.com/apache/datafusion/pull/24766#discussion_r4024690131
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -198,48 +204,133 @@ impl Display for Partitioning {
/// partition 2: keys at/after (2023, Allston)
/// ```
///
-/// NOTE: Optimizer and execution behavior for this partitioning is
intentionally
-/// not implemented and will be introduced incrementally. See
-/// <https://github.com/apache/datafusion/issues/22395>.
+/// Equality includes retained samples, since they determine which future
scales
+/// are possible. Use [`Self::has_same_layout`] to compare only the current
layout.
#[derive(Debug, Clone, PartialEq)]
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]>,
+}
+
+/// Why a [`RangePartitioning`] cannot be scaled to a requested partition
count.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RangePartitioningScaleError {
Review Comment:
I dont think eposing a public error type is the best path here. Most likely
could be a datafusion internal error and have some private enum if needed
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -153,15 +153,21 @@ impl Display for Partitioning {
/// Physical range partitioning.
///
-/// [`RangePartitioning`] describes an ordered key space with split points.
+/// [`RangePartitioning`] describes an ordered key space with sampled split
points.
///
/// - `ordering` defines the partitioning key and ordering.
-/// - `split_points` define the boundaries between adjacent partitions.
+/// - `samples` define the maximum-resolution boundaries.
+/// - `partition_count` selects how many ranges to derive from those samples.
Review Comment:
isnt split points still in the struct?
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -153,15 +153,21 @@ impl Display for Partitioning {
/// Physical range partitioning.
///
-/// [`RangePartitioning`] describes an ordered key space with split points.
+/// [`RangePartitioning`] describes an ordered key space with sampled split
points.
///
/// - `ordering` defines the partitioning key and ordering.
-/// - `split_points` define the boundaries between adjacent partitions.
+/// - `samples` define the maximum-resolution boundaries.
+/// - `partition_count` selects how many ranges to derive from those samples.
Review Comment:
I also am not seeing `partition_count` but may be missing
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs:
##########
@@ -1783,3 +1857,83 @@ fn update_children(mut dist_context:
DistributionContext) -> Result<Distribution
}
// See tests in datafusion/core/tests/physical_optimizer
+
Review Comment:
tests for this go in that file above:
`datafusion/core/tests/physical_optimizer`
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -198,48 +204,133 @@ impl Display for Partitioning {
/// partition 2: keys at/after (2023, Allston)
/// ```
///
-/// NOTE: Optimizer and execution behavior for this partitioning is
intentionally
-/// not implemented and will be introduced incrementally. See
-/// <https://github.com/apache/datafusion/issues/22395>.
+/// Equality includes retained samples, since they determine which future
scales
+/// are possible. Use [`Self::has_same_layout`] to compare only the current
layout.
#[derive(Debug, Clone, PartialEq)]
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]>,
+}
+
+/// Why a [`RangePartitioning`] cannot be scaled to a requested partition
count.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RangePartitioningScaleError {
+ /// A partitioning must contain at least one partition.
+ ZeroPartitions,
+ /// The retained samples cannot support the requested number of partitions.
+ /// Callers may retain the current layout or choose another partitioning.
+ InsufficientSamples {
+ /// Requested number of partitions.
+ target_partitions: usize,
+ /// Largest partition count supported by the retained samples.
+ max_partitions: usize,
+ },
+}
+
+impl Display for RangePartitioningScaleError {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::ZeroPartitions => {
+ write!(f, "Range partitioning partition count must be at least
1")
+ }
+ Self::InsufficientSamples {
+ target_partitions,
+ max_partitions,
+ } => write!(
+ f,
+ "Range partitioning partition count {target_partitions}
exceeds maximum {max_partitions}"
+ ),
+ }
+ }
+}
+
+impl std::error::Error for RangePartitioningScaleError {}
+
+impl From<RangePartitioningScaleError> for DataFusionError {
+ fn from(error: RangePartitioningScaleError) -> Self {
+ Self::Plan(error.to_string())
+ }
}
impl RangePartitioning {
/// Creates range partitioning metadata without validating split points.
///
- /// Use [`Self::try_new`] to validate the contract documented on
- /// [`RangePartitioning`].
+ /// Prefer [`Self::try_new_with_samples`] to validate the boundaries and
retain
+ /// additional samples for scaling up. [`Self::try_new`] remains available
for
+ /// validated exact boundaries.
+ #[deprecated(
+ since = "56.0.0",
+ note = "Use RangePartitioning::try_new_with_samples instead"
+ )]
pub fn new(ordering: LexOrdering, split_points: Vec<SplitPoint>) -> Self {
+ let split_points: Arc<[SplitPoint]> = Arc::from(split_points);
Self {
ordering,
+ samples: Arc::clone(&split_points),
split_points,
}
}
/// Creates range partitioning metadata and validates split point shape and
/// ordering.
+ ///
+ /// The exact boundaries are also the retained samples. This allows scaling
Review Comment:
I would clearly state that the `samples` are derived from teh split points
provided
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -541,10 +541,9 @@ ORDER BY l.range_key;
35 350
##########
-# TEST 13: Compatible Range Join Repartitions to Increase Parallelism
-# Co-partitioning satisfaction does not prevent a repartition that increases
-# parallelism. With target_partitions larger than the Range partition count,
-# both sides are hash repartitioned.
+# TEST 13: Compatible Range Join Preserves Unscalable Partitioning
Review Comment:
is this the inteded behavior we want though? What if we would benefit more
from a repartition to 5 for more parallelism. We should make this configurable
to the user in some way. Maybe a threshold of some type
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs:
##########
@@ -1548,37 +1577,78 @@ pub fn ensure_distribution(
&& target_partitions > child_partitions;
// When subset satisfaction is enabled, preserve an
- // already-satisfying partitioning. Otherwise, hash
+ // already-satisfying partitioning. Otherwise,
// repartition may also increase parallelism.
- let needs_hash_repartition = if
allow_subset_satisfy_partitioning {
+ let needs_repartition = if
allow_subset_satisfy_partitioning {
!partitioning_satisfied
} else {
!partitioning_satisfied
|| (target_partitions > child_partitions
&& !preserve_satisfying_file_partitioning)
};
- let should_add_hash_repartition =
- hash_necessary && needs_hash_repartition;
+ let should_add_repartition = hash_necessary &&
needs_repartition;
// See
https://github.com/apache/datafusion/issues/18341#issuecomment-3503238325 for
background
- // When inserting hash is necessary to satisfy hash
requirement, insert hash repartition.
- if should_add_hash_repartition {
+ // Enforce unmet requirements, or increase parallelism
when beneficial.
+ if should_add_repartition {
+ let partitioning = match
child.plan.output_partitioning() {
+ Partitioning::Range(range) if
partitioning_satisfied => {
+ // A satisfying range layout remains useful to
the
+ // co-partitioning pass even when its samples
cannot
+ // support the preferred degree of parallelism.
+ match range.scale(target_partitions) {
+ Ok(range) => {
+ let scaled =
Partitioning::Range(range);
+ // A single partition satisfies any
key requirement,
+ // but scaling it must still use
compatible keys.
+ if scaled
+ .satisfaction(
+ &requirement,
+
child.plan.equivalence_properties(),
+ false,
+ )
+ .is_satisfied()
+ {
+ scaled_native_range =
+
!child.plan.is::<RepartitionExec>();
+ Some(scaled)
+ } else {
+ Some(
+ requirement
+ .clone()
+
.create_partitioning(target_partitions),
+ )
+ }
+ }
+
Err(RangePartitioningScaleError::InsufficientSamples { .. }) => {
+ preserved_unscalable_range = true;
+ None
Review Comment:
this returns None which I have tracked as meaning “keep the current range
partitioning” and this happens even when preserve_file_partitions is disabled
so I think this shouldnt happen
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
Review Comment:
are we intentionally rejecting here?
##########
datafusion/sqllogictest/test_files/range_partitioning.slt:
##########
@@ -557,10 +556,8 @@ JOIN range_partitioned r ON l.range_key = r.range_key;
----
physical_plan
01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(range_key@0,
range_key@0)], projection=[range_key@0, value@1, value@3]
-02)--RepartitionExec: partitioning=Hash([range_key@0], 5), input_partitions=4
Review Comment:
I am very confused here, we are showing that we are not repartitioning but
we are not doing this due to scaling. This was intentional to show that if we
can increase parallelism in the system we should. I think a better test here
would be showing increasing parallelism by repartitioning on range and scaling
the points correctly
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs:
##########
@@ -1548,37 +1577,78 @@ pub fn ensure_distribution(
&& target_partitions > child_partitions;
// When subset satisfaction is enabled, preserve an
- // already-satisfying partitioning. Otherwise, hash
+ // already-satisfying partitioning. Otherwise,
// repartition may also increase parallelism.
- let needs_hash_repartition = if
allow_subset_satisfy_partitioning {
+ let needs_repartition = if
allow_subset_satisfy_partitioning {
!partitioning_satisfied
} else {
!partitioning_satisfied
|| (target_partitions > child_partitions
&& !preserve_satisfying_file_partitioning)
};
- let should_add_hash_repartition =
- hash_necessary && needs_hash_repartition;
+ let should_add_repartition = hash_necessary &&
needs_repartition;
// See
https://github.com/apache/datafusion/issues/18341#issuecomment-3503238325 for
background
- // When inserting hash is necessary to satisfy hash
requirement, insert hash repartition.
- if should_add_hash_repartition {
+ // Enforce unmet requirements, or increase parallelism
when beneficial.
+ if should_add_repartition {
+ let partitioning = match
child.plan.output_partitioning() {
Review Comment:
I think using an option here too is pretty confusing. Could we make it more
explicit what the variants are like Keep or Repartition on Range
##########
datafusion/physical-plan/src/repartition/mod.rs:
##########
Review Comment:
shouldn't we attempt to scale
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs:
##########
@@ -1548,37 +1577,78 @@ pub fn ensure_distribution(
&& target_partitions > child_partitions;
// When subset satisfaction is enabled, preserve an
- // already-satisfying partitioning. Otherwise, hash
+ // already-satisfying partitioning. Otherwise,
// repartition may also increase parallelism.
- let needs_hash_repartition = if
allow_subset_satisfy_partitioning {
+ let needs_repartition = if
allow_subset_satisfy_partitioning {
!partitioning_satisfied
} else {
!partitioning_satisfied
|| (target_partitions > child_partitions
&& !preserve_satisfying_file_partitioning)
};
- let should_add_hash_repartition =
- hash_necessary && needs_hash_repartition;
+ let should_add_repartition = hash_necessary &&
needs_repartition;
// See
https://github.com/apache/datafusion/issues/18341#issuecomment-3503238325 for
background
- // When inserting hash is necessary to satisfy hash
requirement, insert hash repartition.
- if should_add_hash_repartition {
+ // Enforce unmet requirements, or increase parallelism
when beneficial.
+ if should_add_repartition {
+ let partitioning = match
child.plan.output_partitioning() {
+ Partitioning::Range(range) if
partitioning_satisfied => {
+ // A satisfying range layout remains useful to
the
+ // co-partitioning pass even when its samples
cannot
+ // support the preferred degree of parallelism.
+ match range.scale(target_partitions) {
+ Ok(range) => {
+ let scaled =
Partitioning::Range(range);
+ // A single partition satisfies any
key requirement,
+ // but scaling it must still use
compatible keys.
+ if scaled
+ .satisfaction(
+ &requirement,
+
child.plan.equivalence_properties(),
+ false,
+ )
+ .is_satisfied()
+ {
+ scaled_native_range =
+
!child.plan.is::<RepartitionExec>();
+ Some(scaled)
+ } else {
+ Some(
+ requirement
+ .clone()
+
.create_partitioning(target_partitions),
+ )
+ }
+ }
+
Err(RangePartitioningScaleError::InsufficientSamples { .. }) => {
+ preserved_unscalable_range = true;
+ None
Review Comment:
so like a 3 partiitoned range merging with a 8 partitioned hash, I would
think if the user doesn't opt in we should decide to use more parallelism
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -198,48 +204,133 @@ impl Display for Partitioning {
/// partition 2: keys at/after (2023, Allston)
/// ```
///
-/// NOTE: Optimizer and execution behavior for this partitioning is
intentionally
-/// not implemented and will be introduced incrementally. See
-/// <https://github.com/apache/datafusion/issues/22395>.
+/// Equality includes retained samples, since they determine which future
scales
+/// are possible. Use [`Self::has_same_layout`] to compare only the current
layout.
#[derive(Debug, Clone, PartialEq)]
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]>,
+}
+
+/// Why a [`RangePartitioning`] cannot be scaled to a requested partition
count.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RangePartitioningScaleError {
+ /// A partitioning must contain at least one partition.
+ ZeroPartitions,
+ /// The retained samples cannot support the requested number of partitions.
+ /// Callers may retain the current layout or choose another partitioning.
+ InsufficientSamples {
+ /// Requested number of partitions.
+ target_partitions: usize,
+ /// Largest partition count supported by the retained samples.
+ max_partitions: usize,
+ },
+}
+
+impl Display for RangePartitioningScaleError {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match self {
+ Self::ZeroPartitions => {
+ write!(f, "Range partitioning partition count must be at least
1")
+ }
+ Self::InsufficientSamples {
+ target_partitions,
+ max_partitions,
+ } => write!(
+ f,
+ "Range partitioning partition count {target_partitions}
exceeds maximum {max_partitions}"
+ ),
+ }
+ }
+}
+
+impl std::error::Error for RangePartitioningScaleError {}
+
+impl From<RangePartitioningScaleError> for DataFusionError {
+ fn from(error: RangePartitioningScaleError) -> Self {
+ Self::Plan(error.to_string())
+ }
}
impl RangePartitioning {
/// Creates range partitioning metadata without validating split points.
///
- /// Use [`Self::try_new`] to validate the contract documented on
- /// [`RangePartitioning`].
+ /// Prefer [`Self::try_new_with_samples`] to validate the boundaries and
retain
+ /// additional samples for scaling up. [`Self::try_new`] remains available
for
+ /// validated exact boundaries.
+ #[deprecated(
+ since = "56.0.0",
+ note = "Use RangePartitioning::try_new_with_samples instead"
+ )]
pub fn new(ordering: LexOrdering, split_points: Vec<SplitPoint>) -> Self {
+ let split_points: Arc<[SplitPoint]> = Arc::from(split_points);
Self {
ordering,
+ samples: Arc::clone(&split_points),
split_points,
}
}
/// Creates range partitioning metadata and validates split point shape and
/// ordering.
+ ///
+ /// The exact boundaries are also the retained samples. This allows scaling
+ /// down and back up to the original count, but not beyond it. Prefer
+ /// [`Self::try_new_with_samples`] when additional sample points are
available.
pub fn try_new(ordering: LexOrdering, split_points: Vec<SplitPoint>) ->
Result<Self> {
+ let partition_count = split_points.len() + 1;
+ Self::try_new_with_samples(ordering, split_points, partition_count)
+ }
+
+ /// 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.
+ ///
+ /// Retain at least `maximum_expected_partitions - 1` samples to support
that
+ /// many partitions later. For example, supplying `4 * partition_count`
+ /// samples provides capacity for up to `4 * partition_count + 1`
partitions.
+ /// Choose the sampling factor for the workload; small inputs may not have
+ /// enough distinct values, and callers must handle insufficient capacity.
+ pub fn try_new_with_samples(
Review Comment:
I would love to see a concrete example of samles and split points
relationship being shown. Like what an actual scaling would look like and the
contract it enforces, like not exceeding the max samples.
This also makes me think, samples is not the most telling name for a public
facing property. Maybe something like max_partition_bounds?
##########
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:
I don't know if deprecating is needed rather than it jsut being an
invariant. It seems we can alsways derive the samples form split points.
@stuhood do you think there is a use case for having a contructor that does
not return a `Result`? I could see it if a use case vlaidated themselves and is
really trying to squeeze perf
--
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]
