upgrade to datafusion 55 - #540
Conversation
|
TPCH For queries that got faster, I would expect that they have local dynamic filters, which should work in this branch. Also, I could actually try to run individual queries with metrics but I'm not sure if it's worth the time since the overall diff is 1.01x faster.
TPCDS |
|
@gabotechs What do you think about the benchmarks above? Unremarkable? I'm not sure if it's worth analyzing why things are slow or fast because on average, the improvement is 1x. Seems like the variance is high though. Some queries are 30-40% faster, some are 30-40% slower. I ran these with |
|
I see them within the typical noise. You can try running them with Typically, the most relevant bits you want to look at for overall performance are these: So with those benchmarks my read is that there's just no impact on performance. |
|
We are also maintaining a version of this patch near the bottom of our patch stack: https://github.com/paradedb/datafusion-distributed/commits/main/ ... if you'd like to pick it, feel free! |
78b65f1 to
b8eddac
Compare
jayshrivastava
left a comment
jayshrivastava
left a comment
There was a problem hiding this comment.
@asolimando There's some stats changes you may be interested in. Ex. plan_statistics.rs, prepare_dynamic_plan.rs, and stage.rs.
| &self, | ||
| _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>, | ||
| ) -> Result<TreeNodeRecursion> { | ||
| Ok(TreeNodeRecursion::Continue) |
There was a problem hiding this comment.
In my dynamic filtering related work, I have a PR to store pushed down dynamic expressions. I've excluded that change in this PR for simplicity. Also that behavior does not matter in this PR because remote dynamic filtering is not implemented.
More Context:
During execution (surprisingly not planning) the hash join checks for pushed down filters here. During execution, these network boundaries have no children, so the join cannot tell the filter was pushed down and decides not to update its dynamic filter. The idea is to return pushed down expressions here so producers can tell they are used.
| Partitioning::RoundRobinBatch(p) => Partitioning::RoundRobinBatch(f(*p)), | ||
| Partitioning::Hash(hash, p) => Partitioning::Hash(hash.clone(), f(*p)), | ||
| Partitioning::UnknownPartitioning(p) => Partitioning::UnknownPartitioning(f(*p)), | ||
| Partitioning::Range(range) => Partitioning::Range(range.clone()), |
There was a problem hiding this comment.
cc @gene-bordegaray maybe you can advise here
We can't scale up range without changing the split points so I left this as is. There's no test which really exercise range partitioned data today. RepartitionExec doesn't range partition, so I don't think this will ever be called to scale up a `RepartitionExec.
I filed #628 with some open questions we should probably answer. I can look into that issue after the upgrade.
There was a problem hiding this comment.
This is what we ended up doing temporarily: paradedb#67 ... there was one shape in our plans which did result in Partitioning::Range at a network boundary. @mdashti might know more.
There was a problem hiding this comment.
ya I would back what stu and the parade guys did here, most likely safer to not hide unimplemented capabilities 👍
| async fn create_physical_plan( | ||
| &self, | ||
| logical_plan: &LogicalPlan, | ||
| session_state: &SessionState, |
There was a problem hiding this comment.
Migrating to &dyn Session...
| let converted_dynamic = converted.dynamic_expressions_produced(); | ||
| assert_eq!(original_dynamic.len(), 1); | ||
| assert_eq!(converted_dynamic.len(), 1); | ||
| assert!(Arc::ptr_eq(&original_dynamic[0], &converted_dynamic[0])); |
There was a problem hiding this comment.
dynamic_filter_expr() does not exist anymore, so we use the dynamic_expressions_produced() API
|
|
||
| // Detect completed tasks: tasks that were running but disappeared | ||
| for old_task in &self.tasks { | ||
| if old_task.status == TaskStatus::Running as i32 { |
There was a problem hiding this comment.
Clippy stuff
| terminal.draw(|frame| ui::render(frame, app))?; | ||
|
|
||
| // Check for keyboard input (16ms timeout ~ 60fps responsiveness) | ||
| if event::poll(Duration::from_millis(16))? { |
There was a problem hiding this comment.
More clippy stuff
|
benchmarks run tpch/sf100 |
|
benchmarks run tpds/sf10 |
| /// The hashing/comparison itself is performed by the operator — hash-table build, partition | ||
| /// hashing, sort comparators — not by any expression in the plan, and its cost scales with the | ||
| /// key's byte width. Use it for group-by keys, hash-partition keys and sort keys. | ||
| fn hashed_or_sorted_key_complexity(expression: &Arc<dyn PhysicalExpr>) -> Complexity { |
There was a problem hiding this comment.
I don't think the body needs to change maybe just function name now that range uses it?
|
benchmarks run tpch/sf100 |
|
Requested by this comment. Benchmark job 27 failed for |
gabotechs
left a comment
gabotechs
left a comment
There was a problem hiding this comment.
👍 All looks good here, it seems like this is going to be a straightforward one
| // FIXME: because of limitations the the statistics API on DataFusion, we need to resource to | ||
| // this sketchy way of overriding child statistics, as we cannot just provide our own. | ||
| // If we don't do this: | ||
| // 1. we cannot tell nodes to compute statistics based on the ones we provide. | ||
| // 2. we recompute statistics unnecessarily across the plan | ||
| // This is tracked by https://github.com/apache/datafusion/issues/20184 upstream, and until | ||
| // that one is solved, we need to resource to this wrapper. | ||
| fn partition_statistics_with_children_override( |
| │ t0: DataSourceExec: file_groups={1 group: [[/testdata/flights-1m.parquet:<int>..<int>]]}, projection=[1 as Int64(1)], file_type=parquet | ||
| │ t0: DataSourceExec: file_groups={4 groups: [[/testdata/flights-1m.parquet:<int>..<int>], [/testdata/flights-1m.parquet:<int>..<int>], [/testdata/flights-1m.parquet:<int>..<int>], [/testdata/flights-1m.parquet:<int>..<int>]]}, projection=[1 as Int64(1)], file_type=parquet |
There was a problem hiding this comment.
🤔 Why would this change be? maybe apache/datafusion#22439?
There was a problem hiding this comment.
Yes, exactly.
| #[tokio::test] | ||
| async fn exposes_original_leaf_expressions() -> Result<()> { |
There was a problem hiding this comment.
If you ask me, I think we could leave without this test, but up to you
Thanks Jay for the ping and for taking care of the upgrade. I have checked the changes related to the changes in DF I authored and all LGTM, especially happy we could remove I can take a closer look early next week when I am back in case the PR will still be open. |
|
benchmarks run tpch/sf100 |
|
Requested by this comment. Benchmark job 33 failed for |
|
The runner does not seem to work well with git dependencies... 😢 |
|
benchmarks run tpch/sf100 |
|
Requested by this comment. Run metadata
Queue: 1s · Dataset validation: 0s · Total: 21m 19s Capacity: 12 === Comparing tpch/sf100 results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] === TOTAL: prev=67106 ms, new=68270 ms, diff=1.02 slower ✖ Show full query output q1: prev=2506 ms, new=2473 ms, diff=1.01 faster ✔
q2: prev=1600 ms, new=1632 ms, diff=1.02 slower ✖
q3: prev=2648 ms, new=2650 ms, diff=1.00 slower ✖
q4: prev=1182 ms, new=1103 ms, diff=1.07 faster ✔
q5: prev=3831 ms, new=4312 ms, diff=1.13 slower ✖
q6: prev=1210 ms, new=1158 ms, diff=1.04 faster ✔
q7: prev=4452 ms, new=4649 ms, diff=1.04 slower ✖
q8: prev=4573 ms, new=4729 ms, diff=1.03 slower ✖
q9: prev=6251 ms, new=6216 ms, diff=1.01 faster ✔
q10: prev=5814 ms, new=5977 ms, diff=1.03 slower ✖
q11: prev=1174 ms, new=1145 ms, diff=1.03 faster ✔
q12: prev=1736 ms, new=1807 ms, diff=1.04 slower ✖
q13: prev=1802 ms, new=1806 ms, diff=1.00 slower ✖
q14: prev=1547 ms, new=1684 ms, diff=1.09 slower ✖
q15: prev=3170 ms, new=3168 ms, diff=1.00 faster ✔
q16: prev= 820 ms, new= 786 ms, diff=1.04 faster ✔
q17: prev=4696 ms, new=4776 ms, diff=1.02 slower ✖
q18: prev=5632 ms, new=5664 ms, diff=1.01 slower ✖
q19: prev=1987 ms, new=1708 ms, diff=1.16 faster ✔
q20: prev=2600 ms, new=2487 ms, diff=1.05 faster ✔
q21: prev=6922 ms, new=7439 ms, diff=1.07 slower ✖
q22: prev= 953 ms, new= 901 ms, diff=1.06 faster ✔
|
|
The current CI failure (https://github.com/datafusion-contrib/datafusion-distributed/actions/runs/31694133600/job/94427876747?pr=540) is a flaky test that should be fixed by #624 |
| plans and expressions. See | ||
| [Distribute a custom execution plan](../source/user-guide/04-distribute-custom-plan.md) | ||
| for the complete signatures. | ||
|
|
There was a problem hiding this comment.
We need to change this to 4.0 once 3.0 releases.
6 participants
Summary
Closes #530
Changes
src/protobuf/distributed_codec.rswe now use theproto_converterargument during serdeDeduplicatingProtoConverter, so dynamic filters don't necessarily work. I think this is outside the scope of this PR will be addressed in coordinator: display dynamic filters after execution #623, which will be rebased after the upgrade.ExecutionPlan::apply_expressionsis added for every customExecutionPlanin this repoMetricsWrapperExec,WorkUnitFileScanConfig,DistributedLeafExec) delegate to the inner typeTreeNodeRecursion::Continuebecause they have no expressions (ex.SamplerExec)apply_expressionsdoes not need to yield sort or partitioning expressions in the plan propertiespartition_statisticstostatistics_from_inputsfor everyExecutionPlan.src/distributed_planner/statistics/plan_statistics.rscan just usestatistics_from_inputsdirectly instead of doing theStatisticsWrapperworkaround.statistics/complexity_cpu.rs:238.
Peak-memory metrics use the existing gauge wire representation.
DataFusion added MetricValue::PeakMemoryUsage. It is serialized as the existing named-gauge protobuf variant to avoid a wire-format change. See src/protocol/grpc/
metrics_proto.rs:124.
The value and name survive, and aggregation is still additive, but decoding produces a generic Gauge, not PeakMemoryUsage. The practical difference is mainly display formatting: it
may render as a count rather than human-readable bytes. This is the clearest remaining compromise/risk in the upgrade.
File-scan rebalancing changed its discriminator.
DataFusion removed partitioned_by_file_group; output_partitioning.is_some() is now the source of truth. See src/events/defaults/file_scan_config.rs:43. This decides whether files are
round-robin rebalanced or split through FileGroupPartitioner, so it is behavior-sensitive even though it is a one-line migration.
Two previously ignored correctness tests were enabled.
tests/multi_task_collect_join_repros.rsdrop(reporter) was made explicit on the sampler’s empty-input path.
The reporter sends its result on Drop; explicitly dropping it both satisfies the new compiler/lint behavior and guarantees the zero-row EOS report is sent before returning. See src/
execution_plans/sampler.rs:259.
Plan changes
dynamic_rg_pruning=eligibleis now displayed on eligible scans: 1,354 occurrences in TPC-DS, 188 in TPC-H, and 12 in ClickBenchDataSourceExecnow displays its output partitioning. Seetests/join.rs(eventually, someone should delete this test tests on range partitioned data #628)tests/distributed_unions.rsandtests/distributed_aggregation.rs.LocalLimitExec became more common: TPC-DS went from 0 to 20 occurrences and ClickBench from 1 to 21, reflecting additional local limit pushdown.
Subquery/semi-join plans became more distributed:
These are meaningful topology changes: some subqueries now use partitioned left-semi joins and therefore introduce hash shuffles instead of collecting/broadcasting one side.
Minor changes (Ex. tpcds 21)
__common_expr_4became__common_expr_3; that is only an internal alias renumbering.d_dateto__common_expr_2disappeared.d_dateis retained directly in the join output and referenced directly by partial/final aggregates.File-group allocation changed substantially
repartition_file_min_sizefrom 10 MiB to 1 MiB. The PR explicitly calls out TPC-DS SF1 dimension tables. Files may be duplicated across multiple partitions where but each partition reads a different byte range (this is hidden by ...., but we know from the correctness tests that nothing broke). A lot of tpcds queries now split acrosstarget_partitionsinstead of staying under-partitioned. In thetpcdsplan tests, we usetarget_partitions=3.Example: