avantgardnerio commented on code in PR #2241:
URL:
https://github.com/apache/datafusion-ballista/pull/2241#discussion_r3778267985
##########
ballista/scheduler/src/state/task_builder.rs:
##########
@@ -292,6 +292,13 @@ fn select_output_partitions(
Partitioning::UnknownPartitioning(_) => {
Partitioning::UnknownPartitioning(kept.len())
}
+ Partitioning::Range(_) => {
Review Comment:
This arm rewrites the partitioning to `UnknownPartitioning` while still
slicing the partition list, so the reader handed back under-reports its own
partitioning and its split points are gone.
It may also not be strictly unreachable: DataFusion 55 constructs
`Partitioning::Range` in `physical_planner.rs` when lowering a logical range
`Repartition`, and in `ListingTable`. Nothing routes one into a
`ShuffleReaderExec` today, since `DistributedExchangeRule` only ever builds
from `RepartitionExec::Hash`, so it is latent rather than live.
#2295 turns it into an `internal_err!` with a TODO describing what support
would need (slicing the split points alongside the partition slice, the way the
`RangeFilterExec` arm slices `raw_bounds`).
##########
ballista/core/src/execution_plans/sort_shuffle/writer.rs:
##########
@@ -871,6 +873,19 @@ impl ExecutionPlan for SortShuffleWriterExec {
vec![&self.plan]
}
+ /// This writer hashes rows itself rather than relying on an upstream
+ /// `RepartitionExec`, so the partitioning expressions are its own.
+ /// `try_new` rejects anything but `Hash`.
+ fn apply_expressions(
+ &self,
+ f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
+ ) -> Result<TreeNodeRecursion> {
+ match &self.shuffle_output_partitioning {
+ Partitioning::Hash(exprs, _) => apply_expression_roots(exprs, f),
+ _ => Ok(TreeNodeRecursion::Continue),
Review Comment:
The doc comment above is correct that `try_new` rejects anything but `Hash`,
which makes this arm unreachable. As written though, a writer that somehow held
another scheme would silently report no expressions rather than surfacing the
broken invariant.
#2295 swaps it for a let-else returning the same error `try_new` raises.
##########
ballista/core/src/execution_plans/shuffle_reader.rs:
##########
@@ -2226,10 +2247,12 @@ mod tests {
fn broadcast_reader_rejects_out_of_range_partition_index() {
let schema = Arc::new(Schema::new(vec![Field::new("c",
DataType::Int32, false)]));
let reader = ShuffleReaderExec::try_new_broadcast(7, vec![], schema,
3).unwrap();
- let err = reader.partition_statistics(Some(1)).unwrap_err();
+ let err = StatisticsContext::new()
+ .compute(&reader, &StatisticsArgs::new().with_partition(Some(1)))
+ .unwrap_err();
let msg = err.to_string();
assert!(
- msg.contains("invalid partition index 1"),
+ msg.to_lowercase().contains("invalid partition index"),
Review Comment:
Might be worth keeping the index in this assertion. Dropping it hides that
the test quietly changed target: DataFusion bounds-checks the partition index
inside `StatisticsContext::compute` before dispatching to the operator, so the
error is now upstream's (`Invalid partition index: 1, the partition count is
1`) rather than the broadcast guard in `partition_statistics` that this test
was written for.
#2295 asserts on the message that actually fires, index included.
Related, and happy to leave alone: that broadcast guard now looks
unreachable, and `ShuffleReaderExec` is one of the few operators here still
implementing the deprecated `partition_statistics` rather than
`statistics_from_inputs` / `child_stats_requests`. Glad to send a follow-up if
you would like it migrated with the others.
##########
ballista/core/src/execution_plans/runtime_stats.rs:
##########
@@ -296,6 +298,22 @@ impl ExecutionPlan for RuntimeStatsExec {
vec![&self.input]
}
+ /// When sketching, the first ORDER BY expression is evaluated per batch to
+ /// feed the T-Digest. The whole slice is reported: it is carried for serde
+ /// and for the downstream operators that consume it.
+ fn apply_expressions(
+ &self,
+ f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
+ ) -> Result<TreeNodeRecursion> {
+ apply_expression_roots(
+ self.order_by
+ .iter()
+ .flatten()
Review Comment:
`execute` evaluates only the first key here: `routing_expr` is
`order_by.as_ref().and_then(|exprs| exprs.first())`. The remaining keys are
carried so multi-key ORDER BY survives serde for downstream operators, which
makes them ordering metadata.
The trait doc excludes that case: an expression must not be visited solely
because it describes an input or output property such as cached ordering,
partitioning, or equivalence metadata.
No consumer impact today, since `plan_contains_expression_id` is the only
caller in 55 and Ballista disables dynamic filter pushdown anyway. #2295
narrows it to the first key, derived the same way `execute` derives it so the
two cannot drift.
##########
benchmarks/src/bin/tpch.rs:
##########
@@ -1203,7 +1203,7 @@ async fn get_table(
path: &str,
table: &str,
table_format: &str,
- target_partitions: usize,
+ _target_partitions: usize,
Review Comment:
Small one: now that `ListingOptions` has dropped the field, this parameter
is dead rather than temporarily unused, and the callers already set
`with_target_partitions` on the session config.
#2295 removes it. `register_datafusion_tables` existed only to forward it,
so its `partitions` argument goes too.
--
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]
