You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "avantgardnerio (via GitHub)" <gi...@apache.org> on 2026/06/25 19:21:13 UTC
[I] Add AQE to DataFusion [datafusion]
avantgardnerio opened a new issue, #23194:
URL: https://github.com/apache/datafusion/issues/23194
**Is your feature request related to a problem or challenge?**
DataFusion makes all execution-plan choices at planning time from static statistics, which are frequently `Inexact` or `Absent` (e.g. after a `GROUP BY`, a selective filter, or anything involving derived columns). A wide set of optimizations is blocked by this — the planner has to guess at runtime cardinality and a bad guess costs orders of magnitude.
Concrete decisions that need runtime stats:
- **Build-side swap for hash joins** — JoinSelection picks build/probe from estimates; when wrong, the build side is too big.
- **Dynamic range repartitioning** — parallel sort and parallel window need range boundaries chosen from the actual data distribution, not estimates.
- **Parallel window functions** — partitioning a window operator across threads only pays when partitions are roughly balanced; runtime property.
- **Parallel sort** — pick the right degree of parallelism once we know how much data there is.
- **Post-shuffle partition coalescing** — collapse many tiny partitions (over-estimated repartition) into a smaller number of right-sized ones.
- **Skew handling** — detect a hot partition and split it across workers.
- **Adaptive aggregation strategy** — pick between hash and sort aggregation, or between `Partial` and `Single`, based on observed cardinality.
- **Dynamic filter pushdown from build to probe** — generate a filter from the materialized build side and push it into the probe-side scan.
Each of these needs the same primitive: read runtime stats from a completed sub-plan, then re-run a planning decision against the up-to-date plan.
**Describe the solution you'd like**
PoC in #23167.
The shape of the solution:
- **`StageBoundaryBuffer`** — an `ExecutionPlan` operator that materializes its input, counts rows as the drain task ferries batches through, and exposes the count via `runtime_row_count` (long-term, fold into `partition_statistics` returning `Precision::Exact`). It IS a pipeline breaker by construction — no need to find a natural one inside the operator being adapted.
- **Per-decision insertion rule** — `InsertHashJoinBoundaries` (the first, in the PoC) wraps each HashJoin input. Future adaptive optimizations each get their own targeted insertion rule (`InsertWindowBoundaries`, `InsertRangePartitioningBoundaries`, etc.) — they don't need to know about each other.
- **`RuntimeOptimizerExec`** — single always-on wrapper at the plan root. On each stage-completion event, fires registered runtime rules against the current plan, replaces the plan with each rule's output, then releases the stage's boundaries so downstream consumers can drain.
- **`RuntimeRule` trait with identical signature to `PhysicalOptimizerRule::optimize`.** They live in separate traits today only because `physical-plan` cannot depend on `physical-optimizer` (the dependency runs the other way). End-state: a single trait. Existing rules become runtime-aware by reading boundary state from the plan tree they receive — `JoinSelection` itself can run against a post-breaker plan with no logic change, only fresher numbers.
- **Stages execute sequentially; boundaries within a stage execute in parallel** via spawned drain tasks (matching `RepartitionExec`'s spawn model).
- **Memory bounded by the materialized side, with a spill-or-stream fallback** — under `MemoryPool` pressure, spill to disk if available; otherwise release the boundary early and let the rule see "overflowed" — the query still runs, we just lose the adaptive decision (no worse than the pre-AQE baseline).
The PoC demonstrates the build-side-swap case end-to-end. Every other item in the problem list is a ~30-line targeted insertion rule on the same infrastructure.
**Describe alternatives you've considered**
- **Status quo (static stats only).** Decisions are made at plan time from `Inexact` estimates. Bad estimates produce slow queries. No path to win on the optimizations listed above.
- **Spark-style full re-planning at shuffle boundaries.** Run the full optimizer on the unexecuted portion of the plan after each shuffle finishes. Works in Spark because shuffle is a heavyweight stage barrier. In a streaming/in-process engine like DataFusion, full re-planning at every potential decision point is too coarse. Inserting a targeted pipeline breaker exactly where stats are needed is finer-grained and matches the streaming model.
- **Stat updates without explicit boundaries.** Have every operator update its `partition_statistics` as data flows through; rules poll and re-decide. Possible but requires contract changes across every operator and a polling-based fire mechanism. The boundary approach gives the same epistemic guarantee with a single new operator and event-driven firing.
**Additional context**
PoC PR: #23167. The runtime swap on a real query is observable via `RUST_LOG=info`:
```
RTO: stage 0 ready (2 boundaries); firing 1 rule(s) before release
SwapBuildSideIfInverted: flipping HashJoinExec — current build (left) = 100 rows, probe (right) = 5 rows.
RTO: stage 0 released; downstream consumers can now drain ...
```
Spark AQE for context: https://spark.apache.org/docs/latest/sql-performance-tuning.html#adaptive-query-execution
--
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: github-unsubscribe@datafusion.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4857444576
> We (Datadog), will gladly donate https://github.com/datafusion-contrib/datafusion-distributed to Apache if that implies hosting it as a new crate in https://github.com/apache/datafusion. We've built the crate leaving that door open from the beginning, both from a philosophy and code standpoint.
>
> We've discussed this in the past, but at that moment it was not the right time. Now that we've been running it in production at a huge scale for a while and we can no longer break things and move fast, the door is open.
Thanks @gabotechs . I don't think it would be appropriate to host a complete distributed query engine (ballista or datafusion-distributed) in the core datafusion repo. Donating it to the project as a standalone repo is a different discussion (I'd be happy to comment on that too, if that is useful).
My proposal was to add a minimal amount of code to help catch regressions that break the distributed model.
I'm going to study datafusion-distributed over the next couple of days to understand the architecture and how it varies with Ballista so that I can make sure that I am not proposing something that would help with Ballista and not with datafusion-distributed.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4848793652
> > Now that AQE/AQP has been fleshed out with an implementation and substantive discussion, I'm curious to know if you still prefer:
>
> My personal preference is to see AQP implemented outside the core datafusion crate / repo, be validated that it works, and is broadly applicable, and then potentially bring it into the core to be maintained along with the other code here
It makes sense to prove this out in real distributed consumers first. We are already doing that in Ballista, and @gabotechs is in datafusion-distributed.
The friction for Ballista isn't the AQE logic. It is that core DataFusion increasingly assumes a plan runs in one process, and that breaks the moment a plan is split at a pipeline breaker, serialized, and executed one partition per task. Three we hit just upgrading to DF 54 (apache/datafusion-ballista#1906):
- `DataSourceExec`'s shared scan work queue is only divided correctly when all partitions are polled together. A task polling its one partition drained the queue and read the whole table (https://github.com/apache/datafusion-ballista/issues/1907).
- An uncorrelated scalar subquery's `ScalarSubqueryExpr` only decodes inside its `ScalarSubqueryExec`. Once stage splitting separates them, the stage no longer round trips (https://github.com/apache/datafusion-ballista/issues/1909).
- Runtime dynamic filter pushdown delivers a filter from build side to probe scan within one plan instance. Split across a stage boundary it never arrives and the probe stalls.
I'll start filing issues in core DF when I hit things like this, that are regressions from Ballista's point of view, to help (hopefully) with the conversation.
My own preference is that modeling some of this in core (eventually) would help, because it lets core catch these regressions before a release rather than after. All three shipped in DF 54 and only surfaced once we upgraded. Perhaps this is a new crate in core that models a distributed-style approach, but still in-process, jut to catch these kind of issues.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "edmondop (via GitHub)" <gi...@apache.org>.
edmondop commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-5936343425
I have a small PR (#25798) proposing the stage-boundary interface as a first, standalone piece: an execution-plan capability to prime input partitions, observe when each has completed, and release output. It includes a test/example implementation and examples for deterministic pause/resume and caller-controlled stage admission. It does not add AQE orchestration, runtime optimizer rules, or statistics. I’m sharing it here because it may be a useful core seam for the broader directions in this issue. Feedback on whether this interface belongs in core—and whether its per-partition readiness/release contract is the right scope—would be appreciated.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4809454985
In `datafusion-distributed`, once this EPIC is finished:
- https://github.com/datafusion-contrib/datafusion-distributed/issues/377 (driven by me)
And once the worker protocol has been fully abstracted out of gRPC so that there can be an in-memory implementation:
- https://github.com/datafusion-contrib/datafusion-distributed/pull/508 (driven by ParadeDB folks)
There should be a fully usable single-node AQE implementation that works pretty much out of the box in non-distributed contexts, without requiring materialization to disk. It still requires all the coordination machinery from `datafusion-distributed` though, so I don't think it can be easily factored out.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4855994992
More thoughts on this, and I apologize if I am repeating points already made, but I think it makes total sense to have specialized physical plans for in-process vs distributed. The challenge currently is that each distributed project is starting with the DF in-process plan and converting it and adding hacks to work around the differences in assumptions for in-process vs distributed. I would very much like DF to be a good foundation for distributed query engines (that was always the goal).
To facilitate this, I think it makes sense to have a new optional crate in DF core specifically for this purpose. Initially, it can be based on the existing approach of taking the DF in-process plan and converting it to a distributed plan, but over time this could evolve into having specialized versions of operators for distributed use.
I am going to POC this today and create a draft PR to demonstrate this and see what people think.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4847675625
> Now that AQE/AQP has been fleshed out with an implementation and substantive discussion, I'm curious to know if you still prefer:
My personal preference is to see AQP implemented outside the core datafusion crate / repo, be validated that it works, and is broadly applicable, and then potentially bring it into the core to be maintained along with the other code here
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.
avantgardnerio commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4844950636
Hi @gabotechs , I think your list of concerns is really helpful. I'm usually working with Claude on things, and he provided this helpful mapping of each of them to what's in the PR:
<img width="1792" height="400" alt="Image" src="https://github.com/user-attachments/assets/d2d65ead-7ee8-48eb-906b-b73cb0cfb593" />
I think that makes the picture clearer than it's been previously. I've not been at it long, but I think AQE typically gets some baggage associated with it:
1. "AQE is only for batch processes" - I think your work is really innovative and shows this is untrue
2. "AQE is a distribution concern" - I think Andy's work on Datafusion/Ballista (and this PR) show what is good for the distributed goose is also generally good for the local gander.
I like what you said about idempotent rules being a keystone, and I see folks have been putting a lot of effort into making them so in upstream, in order to benefit downstream repos. This puts Datafusion in an interesting position of having to maintain the code (and the invariant), but not having it easily testable or reap any direct performance benefits. (yes, there can be unit tests, but contributors must know to write and maintain them and why).
What this PR hopes to offer the downstream community is:
1. An operational whitelist of AQE ready optimizer rules, built gradually, over time (idempotent ones might be ready to go!)
2. operators that are AQE compatible
3. _and most importantly in-repo benefit_ so Datafusion contributors have incentive to test & maintain them (performance gains)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "Samyak2 (via GitHub)" <gi...@apache.org>.
Samyak2 commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-5858808212
> * **Post-shuffle partition coalescing** — collapse many tiny partitions (over-estimated repartition) into a smaller number of right-sized ones. Needs per-partition row counts.
> * **Adaptive aggregation strategy** — pick between hash and sort aggregation, or between `Partial` and `Single`, based on observed cardinality.
To add to these points, one "adaptive execution" method we had seen the need for was this: https://github.com/apache/datafusion/issues/20847
Being able to dynamically decide whether we need to partitioned aggregate/partitioned join is something that DataFusion doesn't naturally support. The output partitions (and their nature, like unknown/hash) are fixed at plan time.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.
avantgardnerio commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4812330190
@asolimando
> WDYT?
I forgot about your comment, then went and re-invented most of it in the reply above ☝️ . I'm glad we independently arrived at similar conclusions.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4842226498
Answering now to https://github.com/apache/datafusion/issues/23194#issuecomment-4838546624:
> I'm trying to compile a list of the ongoing AQE-like efforts, inside the existing DF framework, including my own:
> - adaptive agg ...
> - parallel sort ...
> - halo rows ...
> - prefix scan ...
All those PRs ship some form of adaptation local to individual nodes, without reshaping the topology of the execution plan at runtime (correct me if I'm wrong), which I guess they still fit the definition of "adaptive", but it's a different topic from what `ballista` and `datafusion-distributed` in their parallel efforts towards AQE aim for: reshaping the plan topology at execution time for fragments of the plan that still have not began execution.
For reshaping the plan topology at execution time, good idempotent `PhysicalOptimizerRule` implementations that can be re-run on an already optimized plan are key for any system that has a way of collecting runtime `datafusion::physical_plan::Statistics`, this being `ballista`, `datafusion-distributed`, or `datafusion` itself. I bet that having those in `datafusion` would be nice.
Now, the mechanism that splits the plan in fragments and collects runtime `datafusion::physical_plan::Statistics` for the already executed fragments... I don't see how we can come up with an abstraction general enough that it's worth hosting in `datafusion`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4859387785
The way I see it is, we could add the rules that make sense for running on a single machine, distributed engines could extend this with extra rules (just like the current logical / physical optimizer rule list).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4856809829
We (Datadog), will gladly donate https://github.com/datafusion-contrib/datafusion-distributed to Apache if that implies hosting it as a new crate in https://github.com/apache/datafusion. We've built the crate leaving that door open from the beginning, both from a philosophy and code standpoint.
We've discussed this in the past, but at that moment it was not the right time. Now that we've been running it in production at a huge scale for a while and we can no longer break things and move fast, the door is open.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.
avantgardnerio commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4858814434
@comphead if you click through into the (working) [reference PoC](https://github.com/apache/datafusion/pull/23167), you'll find numbered stages:
```
01)RuntimeOptimizerExec
02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, group_key@0)], projection=[group_key@2, sum_payload@3, payload@1]
03)----CoalescePartitionsExec
04)------DataSourceExec: partitions=4, partition_sizes=[1, 0, 0, 0]
05)----ProjectionExec: expr=[group_key@0 as group_key, sum(big.payload)@1 as sum_payload]
06)------PipelineBreakerBuffer
07)--------AggregateExec: mode=FinalPartitioned, gby=[group_key@0 as group_key], aggr=[sum(big.payload)]
08)----------RepartitionExec: partitioning=Hash([group_key@0], 4), input_partitions=4
09)------------PipelineBreakerBuffer
10)--------------AggregateExec: mode=Partial, gby=[group_key@0 as group_key], aggr=[sum(big.payload)]
11)----------------DataSourceExec: partitions=4, partition_sizes=[4, 3, 3, 3]
```
They just execute in an existing `Arc<dyn PhysicalPlanNode>`, being re-written at boundaries (obviously not the in-progress ones).
> some coordinator that knows about all hash sizes, allocations, stats
See `RuntimeOptimizerExec`
> For distributed the stats synchronization
For Ballista, I assumed this would just be 1. recognize the stage boundary, 2. send the info to the scheduler on the stage completion gRPC call (admittedly this breaks down with streaming or large responses). I was hoping we could start with a dedicated stats collector node outside any operator (`PipelineBreakerBuffer` does this with row_count in the PR, `SortExtrema` in another, no reason not to add more - KLL, etc), but it's an interesting question if we can crack open existing operators and extract their state, or if these concerns are intractable.
> Is the scope to come up with some universal trait/approach that would shape the AQE for single machine and distributed?
That is what I interpreted I was being asked to do by Andrew, yes. This adds true AQE to DF without much disruption (comparatively speaking :), and defers the distribution concerns (and distribution only) to downstream implementations.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4859362533
One point I want to make is that dynamic optimization of different stages (a.k.a. AQE in Spark) definitely can be used in a single process / single node env as well.
We still several pipeline breaking operators, e.g. sorts and hash join build side, which basically needs to load the full input before it can make progress.
Currently this could be done ad-hoc inside each operator, but a (small) framework to mark pipeline breaking stages and dynamically re-optimizing the plan sounds like a more principled way of doing things.
Some obvious examples include:
* Join reordering
* Push down dynamic filters (currently it is done inside operators)
* Update statistics (for join, aggregations or parallel merge)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "Dandandan (via GitHub)" <gi...@apache.org>.
Dandandan commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4874134861
> > > If we could somehow migrate Ballista to use datafusion-distributed I think that would also be amazing (and would remove all doubt about moving datafusion-distributed into the core datafusion crate)
> >
> >
> > I would be eager to learn what would be the benefit of doing so?
>
> in my (biased) view it would
>
> 1. Validate that datafusion-distributed is general enough that it can be used with an existing system like Ballista
> 2. It would potentially let us pool development efforts
>
> To be clear I am not saying it is required or even a good idea -- I was mostly dreaming that if it was to come to pass it would be cool
I wouldn't really support directly adding datafusion-distributed as a "vetted" distributed solution / library.
Rather I would support building / extending the building blocks necessary for
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4874456368
> in my (biased) view it would
> 1. Validate that datafusion-distributed is general enough that it can be used with an existing system like Ballista
> 2. It would potentially let us pool development efforts
>
> To be clear I am not saying it is required or even a good idea -- I was mostly dreaming that if it was to come to pass it would be cool
One thing to note here is that the goal of `datafusion-distributed` is to be general for end users, allowing them to build their own distributed engine for their own datasources, for example:
- `datafusion` extension points allow people to provide their own data sources, and customize how they are executed
- `datafusion-distributed` extension points allow people to customize how those data sources should be distributed, what workers should be involved, how work is split, where should network boundaries be placed, etc... all while maintaining an almost exact 1:1 execution model as normal `datafusion`, just happening that some nodes move data over the wire rather than in-memory.
So, the design goal of `datafusion-distributed` is to satisfy end users directly, which I'm not sure if Ballista would fit in that definition. There do are success stories of other systems, like ParadeDB, that are currently built on top of `datafusion-distributed` though, but I don't think it's in the same position as Ballista.
The market gap `datafusion-distributed` fills is also different from Ballista:
- Customizable In-memory engines: Trino is the closest here, but it's not nearly as customizable as the `datafusion` + `datafusion-distributed` combo, and we are confident that a system built on top `datafusion-distributed` is faster and cheaper to run than their Trino equivalent.
- Customizable disk-based shuffling engines: Ballista would fall in this bucket, but there are already really big players doing a very good job in this space (Spark, Comet), so `datafusion-distributed` is not interested in competing here.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4860004054
> If we could somehow migrate Ballista to use datafusion-distributed I think that would also be amazing (and would remove all doubt about moving datafusion-distributed into the core datafusion crate)
I would be eager to learn what would be the benefit of doing so?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4858824752
> We (Datadog), will gladly donate https://github.com/datafusion-contrib/datafusion-distributed to Apache if that implies hosting it as a new crate in https://github.com/apache/datafusion. We've built the crate leaving that door open from the beginning, both from a philosophy and code standpoint.
>
> We've discussed this in the past, but at that moment it was not the right time. Now that we've been running it in production at a huge scale for a while and we can no longer break things and move fast, the door is open.
I am much more in favor of adopting datafusion-distributed compared to starting with a new crate in the datafusion repo -- `datafusion-distributed` has had a lot of thought in it and I have heard from many users they are using it / plan to use it (as are we at InfluxData). This gives me confidence that its APIs are broadly applicable and we can maintain it.
To be clear, datafusion-distributed is not a complete distributed query engine, it is more like "the common pieces needed to build one" -- like shuffles and network communication traits.
> I'm going to study datafusion-distributed over the next couple of days to understand the architecture and how it varies with Ballista so that I can make sure that I am not proposing something that would help with Ballista and not with datafusion-distributed.
If we could somehow migrate Ballista to use datafusion-distributed I think that would also be amazing (and would remove all doubt about moving datafusion-distributed into the core datafusion crate)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "asolimando (via GitHub)" <gi...@apache.org>.
asolimando commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4807999502
Similar work for distributed-dafafusion from @gabotechs: https://github.com/datafusion-contrib/datafusion-distributed/pull/486
I only looked at the issue/PR description for this issue, but I can already see many commonalities:
- pipeline breaker/boundaries as a good place to accumulate runtime statistics
- re-use of the built-in statistics propagation mechanism (great for re-use), only fueled with runtime statistics
- runtime statistics must be sampled as we are in a streaming computational model (idea behind `SamplerExec` in the above PR, and the similar buffer node here)
I wonder how much can be re-used across core DF/distributed DF/ballista, there are different challenges and the same logical concept has different forms in the three cases, but the mechanism seems very similar, if not identical.
WDYT?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4811570206
> The key architectural trait I'd like to highlight is the original "isomorphic scaling"
I found this terminology somewhat complicated to grok at first -- in my words this means that the work is divided across DataFusion partitions and additional parallelization is implemented explicitly in terms of partitions (rather than more cores for a particular operator, for example).
The rationale is that ballista and datafusion distributed reason about and can distirbuted partitions across cores and nodes
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4810551376
One thing that I'd expect to be shared across all approaches and probably very key to all of them, is to ensure we can reuse the current `PhysicalOptimizerRule` implementations that swap joins, etc... for the during-execution passes that adapt the query at runtime.
I imagine this implies that these `PhysicalOptimizerRule`s should be idempotent among other things.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.
avantgardnerio commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4812304433
@gabotechs , I've reviewed dfd's architecture and your above proposal, and I think we can come to a common proposal that satisfies DF, dfd, and Ballista.
Ideas:
1. The PoC's StageBoundary becomes the same shape as SamplerExec, parameterized by a single capacity N. Small N = sampler-shaped (bounded peek with NDV/null-pct/velocity); N=∞ = pipeline breaker (exact stats, gated on spill). Same operator, one knob: streaming-vs-eager collapses into a scalar (with other possible tricks down the road like memory-pressure awareness, spill, and upstream pipeline breaker awareness)
2. I think `SamplerExec` provides truly richer stats, I think we can build stats that are a superset of `LoadInfo` and the `PartitionStatistics` of today (adding PartitionSortExtrema while we are at it)
3. The StageBoundary/SamplerExec takes a listener: Option<Arc<dyn StatsListener>>, in the constructor. DF would use None, but upstream projects could get async notifications
This way dfd's coordination layer stays completely intact — your oneshot→gRPC delivery becomes the StatsListener impl — and DF picks up the observation primitive natively. Do you think this would be a fruitful direction? If you'd like to merge in SamplerExec then DF would have the perfect slot for it. Or if you prefer, the work would be lifted directly with attribution.
As an aside, I would like to correct a confusing sentence from before though - when I said:
> What this PR hopes to do is add the necessary primitives within Datafusion
I meant primitives + intra-DF-AQ-execution. This way multiple projects don't need to redo the work of AQE aware optimizations, the entire DF community would help in the AQE project. And what were previously single-core operations (windows without partitionby) within DF become parallelized within a worker/executor.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4810277098
Just for context, https://www.cs.cmu.edu/~15721-f24/papers/AQP_in_Lakehouse.pdf
I'm not sure we can make one common AQE but we should definitely try. If we could make it as pluggable and flexible as planning rules are maybe it will work.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4858283179
Thanks @avantgardnerio, what is the scope for this AQE? Currently discussing datafusion and distributed variants.
I have some feeling that support AQE for datafusion single machine engine and distributed variants are totally different amount of work.
For distributed the stats synchronization is supposed to be more complicated, it would require some coordinator that knows about all hash sizes, allocations, stats, etc.
Also to insert pipeline breakers the distributed prob should support some sort of stages like Spark.
Is the scope to come up with some universal trait/approach that would shape the AQE for single machine and distributed?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.
avantgardnerio commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4846178779
@alamb — circling back to your #23093 [comment](https://github.com/apache/datafusion/issues/23093#issuecomment-4770428368) :
> handle at a higher level — for example building some sort of more 'adaptive query processing' into DataFusion.
Now that AQE/AQP has been fleshed out with an implementation and substantive discussion, I'm curious to know if you still prefer:
1. the AQE direction in this issue
2. or the in-operator method like the 4 examples above (halo rows, prefix scan, adaptive agg, parallel sort)
I'd like to proceed with whichever one is the most viable route, because presently we only utilize 1/4000th of our cores while executing unpartitioned window functions.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4858260852
Created draft PR https://github.com/apache/datafusion/pull/23282 - obviously all AI-generated - I can't build that fast 😅 . I am going to start reviewing it myself now. Looking forward to feedback.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4804230071
Note there is a related ticket / work in balllista
- https://github.com/apache/datafusion-ballista/issues/1359
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4809266105
How about starting in datafusion-contrib/datafusion-aqe?
I think this can be implemented almost entirely outside of the core crates, where we are already struggling with a large umber of features.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.
avantgardnerio commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4810571111
> PhysicalOptimizerRule
The `RuntimeRule` trait only exists in the PoC because of a circular dependency issue that is easily resolved if we agree on the direction. The intent is to use `PhysicalOptimizerRule` - though each one would have to be made "runtime aware" and added to the `RuntimeOptimizer` white list of rules; but, this approach can happen incrementally by converting a rule at a time, not a "big bang" where we deliver no value until every impl changes.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4810633997
I'm on a vacation will help once I'm back
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4810039046
> There should be a fully usable single-node AQE implementation that works pretty much out of the box in non-distributed contexts,
FYI @milenkovicm and @avantgardnerio -- see @gabotechs comments above
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.
avantgardnerio commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4810556878
> I'm not sure we can make one common AQE but we should definitely try.
I'd like to highlight that Coralogix runs Ballista internally for distribution, so this PR was definitely written with distribution in mind.
The key architectural trait I'd like to highlight is the original "isomorphic scaling" approach that Datafusion & Ballista took from inception: the same mechanisms Datafusion uses to spread load across cores is the same mechanism that Ballista uses to distribute across executors.
The intent here was to follow that pattern: Stage boundaries inserted by these rules would become shuffle writes in Ballista. (The additional parallel window function PRs all follow this pattern as well).
So I don't think this is a competing approach, and a `datafusion-aqe` trait would just end up being a 3rd effort would diverge rather than converge existing approaches. What this PR hopes to do is add the necessary primitives within Datafusion to make distributed AQE easier.
I'm not as familiar with `datafusion-distributed`, but I'm going to spend the rest of my day comparing the other two implementations and roadmaps with what is provided here - the goal being to find commonality between them and move it in-repo.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "zhuqi-lucas (via GitHub)" <gi...@apache.org>.
zhuqi-lucas commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4815834188
> Adaptive aggregation strategy — pick between hash and sort aggregation, or between Partial and Single, based on observed cardinality.
https://github.com/apache/datafusion/pull/22518
Related PR in progress , and we have good benchmark result.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4841887365
Answering @avantgardnerio here:
> its oneshot→gRPC delivery becomes the StatsListener impl — and DF picks up the observation primitive natively. Do you think this would be a fruitful direction? If you'd like to merge in SamplerExec then DF would have the perfect slot for it. Or if you prefer, the work would be lifted directly with attribution.
`SamplerExec` is really just the tip of the iceberg, it's only useful in the context of the coordination logic of `datafusion-distributed`, as a coordinating entity needs to be responsible of:
1) Kicking it off eagerly so that runtime sampling starts before execution ([done here](https://github.com/datafusion-contrib/datafusion-distributed/blob/2203b3bfdd93b1ad258e60a6dc668c9911b7b9ff/src/worker/impl_coordinator_channel.rs#L104-L106))
2) Collecting the runtime samples, reconciling the different samples from the different partitions, and building a `datafusion::physical_plan::Statistics` out of the reconciled samples ([done here](https://github.com/datafusion-contrib/datafusion-distributed/blob/2203b3bfdd93b1ad258e60a6dc668c9911b7b9ff/src/coordinator/prepare_dynamic_plan.rs#L195-L195)).
3) Re-arranging the yet un-sampled remaining part of the plan based on the runtime collected `datafusion::physical_plan::Statistics` ([done here](https://github.com/datafusion-contrib/datafusion-distributed/blob/0ac54f00f70d6d9c56ffb43534227635eadc08a3/src/coordinator/prepare_dynamic_plan.rs#L70-L70))
So unless it's decided that `datafusion-distributed` coordinating machinery is general enough to also be ported to `datafusion`, I think there's not a lot to gain from moving just `SamplerExec` to `datafusion`, as I don't think it will generalize well to other coordinating machineries.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4860174791
Rather than prescribing AQE implementation should the focus be on providing building blocks to implement AQE planners?
From perspective of ballista AQE the hard part was to make "event driven" core which would re-plan stages based on statistics collected at the stage boundaries.
Significant discrepancy from datafusion execution is that datafusion makes quite a lot decisions before plan starts (eg. partitioning, or join implementation), AQE would benefit from differing such decisions for later in the flow. https://www.cs.cmu.edu/~15721-f24/papers/AQP_in_Lakehouse.pdf mentions that spark converts parts of logical to physical plan when needed rather than on execution start, I believe similar approach would simplify AQE implementation (ballista implements DynamicJoin to defer some partitioning and join decisions, which i see as a hack)
IMHO having a planner which could act on execution events would be the first step
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "avantgardnerio (via GitHub)" <gi...@apache.org>.
avantgardnerio commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4847085245
BTW, here's the affect on parallel windows:
<img width="1560" height="600" alt="Image" src="https://github.com/user-attachments/assets/5c2e9724-9743-48c1-87d4-7d1d3d705303" />
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [I] Add AQE to DataFusion [datafusion]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on issue #23194:
URL: https://github.com/apache/datafusion/issues/23194#issuecomment-4874870525
Datafusion's first concern, rightly, was not AQE; thus, trying to add one at this moment may be more of a hack rather than a systematic approach to it.
I agree with @Dandandan comment https://github.com/apache/datafusion/issues/23194#issuecomment-4859362533 making a plan-observation/runtime-adaptation framework could be a step in right direction, and would help to make some separation of concerns between operators and planner(s). IMHO, making a case how core datafusion would benefit from this would be a good first step.
I'd suggest we focus on requirements/gaps rather than prescribing solutions. Also, it would be great if we refrain from speculative comparisons and claims, as they do not really help.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org