englefly opened a new pull request, #70129:
URL: https://github.com/apache/doris/pull/70129
### What problem does this PR solve?
Issue Number: None
Related PR: #69844
Problem Summary:
Runtime filters currently target scans. Blocks produced before a scan
receives its filter can reach exchange receivers without being filtered. The FE
needs to describe additional receiver targets while keeping each probe
expression bound to the tuple available at that receiver.
This FE-only implementation records the probe expression when runtime-filter
pushdown crosses a `PhysicalDistribute`. Project aliases, join equivalences,
set-operation mappings and pushdown barriers continue to use the existing
visitor. Translation resolves each recorded expression against the exchange's
input tuple IDs, before its optional projection, casts it to the source type,
deduplicates shared exchange targets when merging filters, and registers every
receiving fragment through the existing RF protocol.
For example, `Exchange -> Project(a + 1 AS x) -> Scan(a)` receives an RF on
`x`, while the scan keeps its RF on `a + 1`. A projection attached to the
receiver binds the RF to the receiver's input SlotIds, not its projected output
SlotIds.
- Introduce `enable_runtime_filter_on_exchange`, disabled by default and
forwarded to the planning FE. Enabling it requires matching BE
exchange-consumer support, which is handled separately.
- Preserve scan pruning metadata, existing RF wait policies, default wait
time, non-blocking requirements and whole-filter local/remote routing.
- Keep CTE consumer filters on their existing multicast sink path: the
virtual scan and receiver share a plan ID but use different tuples, so adding
both would overwrite the sink expression in `planId_to_target_expr`.
- Skip distributions eliminated by bucketed aggregation fusion during
translation.
- Targets extend existing scan RFs; exchange-only filters, such as filters
on aggregate results that cannot reach a scan, are outside this change.
BE integration contract: consume `TPlanNode.runtime_filters` on exchange
receivers, select `planId_to_target_expr` by the receiver node ID, and prepare
those expressions against the receiver's row descriptor before projection.
Receiver consumers should apply late arrivals without adding blocking wait
dependencies. No Thrift changes are required by this FE implementation.
### Release note
Add the opt-in session variable `enable_runtime_filter_on_exchange` (default
`false`). With matching BE support, enabling it adds exchange receiver targets
to existing scan runtime filters. Runtime-filter wait defaults and policies are
unchanged.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test
- [ ] No need to test or manual test
- `bash run-fe-ut.sh --run
org.apache.doris.nereids.glue.translator.RuntimeFilterExchangeTest,org.apache.doris.nereids.glue.translator.RuntimeFilterTranslatorBucketPruneTest,org.apache.doris.nereids.postprocess.RuntimeFilterTest`:
**69 tests passed**, zero failures/errors/skips (12 new exchange tests, 7
bucket-pruning tests, 50 existing RF tests).
- `./build.sh --fe -j8`: **passed**, including Checkstyle (zero violations).
- Tests exercise actual optimizer plans, real exchange nodes and fragments:
alias rewriting, projection-before/after tuple binding, multiple receivers,
grouped Union targets, equivalent-target deduplication, joins below receivers,
non-blocking groups, CTE sink isolation, TopN/grouping-set barriers and the
disabled default.
- BE exchange execution and distributed end-to-end performance are deferred
to the companion BE implementation.
- Behavior changed:
- [ ] No.
- [x] Yes. Opt-in exchange receiver targets; disabled by default.
- Does this need documentation?
- [x] No. This FE-only preparation documents its disabled-by-default
flag in `VarAttr`; end-user exchange RF documentation should accompany the
separately delivered BE support.
- [ ] Yes.
Review checkpoints: RF generation and translation share existing pushdown
semantics; both single and grouped translation paths use the same target
helper. Metadata and bindings are query-scoped, with no new threads, locks or
persistent state. The switch is dynamically settable and forwarded; no Thrift
fields or storage formats change. Existing legacy EXPLAIN output exposes
receiver targets. Planning uses recorded boundary expressions instead of
repeated DFS over the translated tree.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
--
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]
