gene-bordegaray commented on code in PR #24501:
URL: https://github.com/apache/datafusion/pull/24501#discussion_r3895822801
##########
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:
this function is always checking:
```text
f(split) vs f(predecessor(split))
```
but for when `SortOptions:descending` the adjacent value should be the
successor (aka the rigtht side).
Example:
```text
Range([timestamp DESC]), split = 01:00
---
The partitions are
partition 0: timestamp > 01:00
partition 1: timestamp <= 01:00
So then the spllit:
partition 0 partition 1
01:00:00.000000001 | 01:00:00.000000000
^
split
---
So the value adjacent to the split in partition 0 is the successor thus
`+1ns`
```
Can we check this and add regression tst for different sort options
##########
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:
this can panic for float NaN and infinite values. If a user decied to make
an non finite float split point it would show up. So pretty niche but maybe we
check finite here
##########
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:
can we ? instead of just continui8ng in this case. I think it isnt always ok
to cntinue if this returnes None and should downgrade
--
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]
