avantgardnerio commented on code in PR #26593:
URL: https://github.com/apache/datafusion/pull/26593#discussion_r4470887816
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -293,9 +318,92 @@ 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.
+ ///
+ /// # 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(), Some(&point(20)));
+ /// assert_eq!(first.end_bound(), second.start_bound());
+ /// assert_eq!(first.start_bound(), None);
+ /// assert_eq!(second.end_bound(), None);
+ ///
+ /// // A bound must not exclude any retained samples.
+ /// assert!(RangePartitioning::try_new(ordering, vec![point(30)])?
+ /// .try_with_bounds(Some(point(35)), None)
+ /// .is_err());
+ /// # Ok(())
+ /// # }
+ /// ```
+ pub fn try_with_bounds(
+ mut self,
+ start_bound: Option<SplitPoint>,
+ end_bound: Option<SplitPoint>,
+ ) -> Result<Self> {
+ // Samples are already ordered, so their endpoints suffice to validate
+ // the bounds, including samples omitted from the effective split
points.
+ let endpoints = start_bound
+ .iter()
+ .chain(self.samples.first())
+ .chain(self.samples.last())
+ .chain(end_bound.iter())
+ .cloned()
+ .collect::<Vec<_>>();
+ validate_range_split_points(
Review Comment:
> require each bound to lie strictly outside the retained samples
Ballista needs this one non-strict too. A task's bounds are the cuts on
either side of its slice, so with repeated cuts `start_bound` can equal its
first split point, and a task whose only partition sits between two equal cuts
has `start_bound == end_bound`.
--
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]
