mbutrovich commented on code in PR #26593:
URL: https://github.com/apache/datafusion/pull/26593#discussion_r4447697157
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -409,7 +486,8 @@ impl RangePartitioning {
}
}
- /// Adapts this range partitioning to the given expressions, preserving
split points and sort options.
+ /// Adapts this range partitioning to the given expressions, preserving
split points, bounds, and sort options.
+ /// The new expressions must describe the same keys for the bounds to
remain valid.
Review Comment:
Should `RangePartitioning::adapt` drop the bounds itself? Its only caller in
the tree is `Partitioning::adapt`, which clears them right after the call,
because the other input may contain keys outside this input's range:
https://github.com/apache/datafusion/blob/67ad9002074c58ed45962c7adcd67c021feaa2d7/datafusion/physical-expr/src/partitioning.rs#L824-L830
`adapt` exists to describe a different input with this input's layout. An
external caller that uses `RangePartitioning::adapt` directly, for example a
distributed planner matching the other side of a join, gets bounds that input
doesn't satisfy. A `RepartitionExec` built from that partitioning now fails
with "outside the declared bounds" on the first key past either end. Neither
method is in a release yet, so changing the semantics here costs nothing
downstream.
If `adapt` sets `start_bound: None, end_bound: None` in the struct it
returns, `Partitioning::adapt` can go back to `.map(Partitioning::Range)`, and
`test_range_partitioning_adapt` would assert `None` for both bounds instead of
the source's bounds:
https://github.com/apache/datafusion/blob/67ad9002074c58ed45962c7adcd67c021feaa2d7/datafusion/physical-expr/src/partitioning.rs#L1950-L1951
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -293,9 +318,58 @@ impl RangePartitioning {
ordering,
samples,
split_points,
+ start_bound: None,
+ end_bound: None,
})
}
+ /// Sets the inclusive start and exclusive end of the covered key space.
+ ///
+ /// `None` leaves that end unbounded. Bounds use the same lexicographic
order
+ /// as the split points, including descending and null ordering. Every
+ /// retained sample must lie between the bounds, inclusive. Equality
permits
+ /// empty first or last partitions; equal start and end bounds describe an
+ /// entirely empty range.
+ ///
+ /// This changes metadata only. The caller must ensure that every row falls
+ /// within the declared bounds and in the partition described by the split
points.
Review Comment:
Could we add a runnable example to `try_with_bounds`? The struct docs show
bounds only as a `text` block. A doctest that splits `[10, 20, 30]` across two
tasks would check the use case from the PR description and also cover
`start_bound` and `end_bound`. Something like:
```suggestion
/// This changes metadata only. The caller must ensure that every row
falls
/// within the declared bounds and in the partition described by the
split points.
///
/// # Example
///
/// Split `[10, 20, 30]` across two tasks without losing the boundary at
`20`:
///
/// ```
/// # use arrow::datatypes::{DataType, Field, Schema};
/// # use datafusion_common::{Result, ScalarValue};
/// # use datafusion_physical_expr::expressions::col;
/// # use datafusion_physical_expr::{LexOrdering, PhysicalSortExpr,
RangePartitioning, SplitPoint};
/// # fn main() -> Result<()> {
/// let schema = Schema::new(vec![Field::new("x", DataType::Int64,
false)]);
/// let ordering: LexOrdering =
/// [PhysicalSortExpr::new_default(col("x", &schema)?)].into();
/// let point = |v| SplitPoint::new(vec![ScalarValue::Int64(Some(v))]);
///
/// // x < 10 and 10 <= x < 20
/// let first = RangePartitioning::try_new(ordering.clone(),
vec![point(10)])?
/// .try_with_bounds(None, Some(point(20)))?;
/// // 20 <= x < 30 and 30 <= x
/// let second = RangePartitioning::try_new(ordering.clone(),
vec![point(30)])?
/// .try_with_bounds(Some(point(20)), None)?;
/// assert_eq!(first.end_bound(), second.start_bound());
/// assert_eq!(first.start_bound(), None);
///
/// // A bound must not fall inside the retained samples.
/// assert!(RangePartitioning::try_new(ordering, vec![point(30)])?
/// .try_with_bounds(Some(point(35)), None)
/// .is_err());
/// # Ok(())
/// # }
/// ```
```
--
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]
