NGA-TRAN commented on code in PR #24501:
URL: https://github.com/apache/datafusion/pull/24501#discussion_r3937722539
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -252,34 +258,189 @@ impl RangePartitioning {
///
/// Returns `None` if any range key cannot be projected or if projection
/// collapses distinct range keys into duplicate output expressions.
+ ///
+ /// If a projection drops a range key but keeps a monotonic function of it
+ /// (for example `date_bin(interval, timestamp)` or `date_trunc(unit,
timestamp)`
+ /// while range-partitioned on `timestamp`), the range can still be
projected.
+ /// Adjacent partitions stay disjoint only when evaluating the function at
+ /// each split point and its predecessor yields different values, so bins
+ /// do not straddle file groups.
fn project(
&self,
mapping: &ProjectionMapping,
input_eq_properties: &EquivalenceProperties,
) -> Option<Self> {
- let exprs = self
- .ordering
- .iter()
- .map(|sort_expr| Arc::clone(&sort_expr.expr))
- .collect::<Vec<_>>();
- let projected_exprs = input_eq_properties
- .project_expressions(&exprs, mapping)
- .collect::<Option<Vec<_>>>()?;
- let sort_exprs = self
- .ordering
- .iter()
- .zip(projected_exprs)
- .map(|(sort_expr, expr)| PhysicalSortExpr::new(expr,
sort_expr.options))
- .collect::<Vec<_>>();
+ let mut split_points = self.split_points.clone();
+ let mut sort_exprs = Vec::with_capacity(self.ordering.len());
+ for (key_idx, sort_expr) in self.ordering.iter().enumerate() {
+ if let Some(projected) =
+ input_eq_properties.project_expr(&sort_expr.expr, mapping)
+ {
+ sort_exprs.push(PhysicalSortExpr::new(projected,
sort_expr.options));
+ continue;
+ }
+
+ let (target, source) =
+ monotonic_range_key_projection(sort_expr, mapping,
input_eq_properties)
+ .find(|(_, source)| {
+ monotonic_fn_keeps_partitions_disjoint(
+ source,
+ &sort_expr.expr,
+ &split_points,
+ key_idx,
+ )
+ })?;
+ if let Some(updated) = project_split_points_through_fn(
+ &source,
+ &sort_expr.expr,
+ &split_points,
+ key_idx,
+ ) {
+ split_points = updated;
+ }
+ sort_exprs.push(PhysicalSortExpr::new(target, sort_expr.options));
+ }
let ordering = LexOrdering::new(sort_exprs)?;
if ordering.len() != self.ordering.len() {
return None;
}
Some(Self {
ordering,
- split_points: self.split_points.clone(),
+ split_points,
+ })
+ }
+}
+
+/// Yields projection mappings whose source is a same-direction monotonic
+/// transform of `sort_expr`. Callers pick the first candidate that also
+/// keeps adjacent partitions disjoint.
+fn monotonic_range_key_projection<'a>(
+ sort_expr: &'a PhysicalSortExpr,
+ mapping: &'a ProjectionMapping,
+ eq_properties: &'a EquivalenceProperties,
+) -> impl Iterator<Item = (Arc<dyn PhysicalExpr>, Arc<dyn PhysicalExpr>)> + 'a
{
+ mapping
+ .iter()
+ .filter(|(source, _)| {
+ eq_properties.check_monotonic_transform(source, &sort_expr.expr)
+ })
+ .map(|(source, targets)| (Arc::clone(&targets.first().0),
Arc::clone(source)))
+}
+
+/// Adjacent range partitions remain disjoint on `fn_expr` when the function
+/// value at each split differs from the value immediately below the split.
+fn monotonic_fn_keeps_partitions_disjoint(
Review Comment:
Addressed: `monotonic_fn_keeps_partitions_disjoint` now uses the successor
for DESC (predecessor remains for ASC).
`Range([timestamp DESC], [01:00])` is rejected for `date_trunc('hour')` /
`date_bin(60s)` because the 01:00 bin is shared. A split at the last nanosecond
of the hour/minute stays disjoint. Covered by
`test_range_partitioning_descending_uses_successor_for_disjointness`.
##########
datafusion/expr-common/src/interval_arithmetic.rs:
##########
@@ -1243,6 +1243,20 @@ fn prev_value(value: ScalarValue) -> ScalarValue {
value_transition!(MIN, false, value)
}
+/// Returns the previous distinct value of `value`, or `None` if `value` is
+/// null, already at the type minimum, or a type that has no predecessor.
+pub fn checked_predecessor(value: &ScalarValue) -> Option<ScalarValue> {
+ if value.is_null() {
+ return None;
+ }
+ let predecessor = prev_value(value.clone());
Review Comment:
Addressed: `checked_predecessor` (and the new `checked_successor`) return
`None` for `NaN` / ±∞ instead of calling `prev_value` / `next_value`, which
debug-assert finiteness. Covered in `test_checked_predecessor` and
`test_checked_successor`.
##########
datafusion/physical-expr/src/partitioning.rs:
##########
@@ -252,34 +258,189 @@ impl RangePartitioning {
///
/// Returns `None` if any range key cannot be projected or if projection
/// collapses distinct range keys into duplicate output expressions.
+ ///
+ /// If a projection drops a range key but keeps a monotonic function of it
+ /// (for example `date_bin(interval, timestamp)` or `date_trunc(unit,
timestamp)`
+ /// while range-partitioned on `timestamp`), the range can still be
projected.
+ /// Adjacent partitions stay disjoint only when evaluating the function at
+ /// each split point and its predecessor yields different values, so bins
+ /// do not straddle file groups.
fn project(
&self,
mapping: &ProjectionMapping,
input_eq_properties: &EquivalenceProperties,
) -> Option<Self> {
- let exprs = self
- .ordering
- .iter()
- .map(|sort_expr| Arc::clone(&sort_expr.expr))
- .collect::<Vec<_>>();
- let projected_exprs = input_eq_properties
- .project_expressions(&exprs, mapping)
- .collect::<Option<Vec<_>>>()?;
- let sort_exprs = self
- .ordering
- .iter()
- .zip(projected_exprs)
- .map(|(sort_expr, expr)| PhysicalSortExpr::new(expr,
sort_expr.options))
- .collect::<Vec<_>>();
+ let mut split_points = self.split_points.clone();
+ let mut sort_exprs = Vec::with_capacity(self.ordering.len());
+ for (key_idx, sort_expr) in self.ordering.iter().enumerate() {
+ if let Some(projected) =
+ input_eq_properties.project_expr(&sort_expr.expr, mapping)
+ {
+ sort_exprs.push(PhysicalSortExpr::new(projected,
sort_expr.options));
+ continue;
+ }
+
+ let (target, source) =
+ monotonic_range_key_projection(sort_expr, mapping,
input_eq_properties)
+ .find(|(_, source)| {
+ monotonic_fn_keeps_partitions_disjoint(
+ source,
+ &sort_expr.expr,
+ &split_points,
+ key_idx,
+ )
+ })?;
+ if let Some(updated) = project_split_points_through_fn(
+ &source,
+ &sort_expr.expr,
+ &split_points,
+ key_idx,
+ ) {
+ split_points = updated;
Review Comment:
Addressed: `project` now uses `?` on `project_split_points_through_fn`. If
the split cannot be rewritten into the transformed domain, Range is dropped
rather than kept with source-domain bounds.
--
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]
