englefly opened a new pull request, #69844:
URL: https://github.com/apache/doris/pull/69844
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
**In short.** A runtime filter target scan used to wait for its filter from
the moment the filter consumer was created during fragment prepare. The build
side of the join that produces the filter can run for a long time, so the wait
budget (1s by default, adjusted per query by `configRuntimeFilterWaitTime()`)
often expired while the build side was still running. The scan then started
without a filter that was published right after the build finished, so the
filter was wasted exactly when the scan was about to read data.
**Fix 1: start the budget when the pipeline can run.** The BE now starts the
runtime-filter wait clock only after the join build dependency of the target
scan's pipeline is ready (`PipelineTask::_wait_to_start()`). That dependency is
released at build EOS and the filter is published when the build sink closes,
so the short budget is spent right when the filter is about to arrive. The
budget is now a fixed 20ms by default (`runtime_filter_wait_time_ms`) instead
of an estimate based on table row counts, and `configRuntimeFilterWaitTime()`
is removed. A pipeline without a join build side (the scan is in another
fragment) starts the clock when it first runs.
**Fix 2: filters that arrive later are applied at exchanges.** Because the
scan no longer waits long, it can produce blocks before the filter arrives. The
FE therefore attaches the same filter to every exchange node on the plan tree
path from the target scan to the builder join
(`RuntimeFilterTranslator.addExchangeTargets()`), and the BE exchange source
operator applies the filter to the blocks it receives through the existing
late-arrival path (`RuntimeFilterConsumerHelper`, never blocking the exchange
pipeline). The target expression is rebuilt against each exchange's own output
tuple, and an exchange is only used when its output still carries that slot and
the path down to the scan is a single row-preserving chain, so rows from other
branches (joins, set operations, grouping sets, recursive CTEs) are never
filtered with a predicate that does not describe them.
**Implementation notes.**
- `RuntimeFilterDependency` marks the execution dependencies that belong to
runtime filters and forwards the pipeline-ready event to the consumer and its
timers. `RuntimeFilterTimer::should_be_check_timeout()` returns false until
then, so the timer can neither time out nor unblock the scan early.
- The wait clock of the timer and of the consumer (profile/metrics) both
start at that event. `force_wait_timeout` timers (group commit) start
immediately, as before.
- Exchange consumers are created with `wait_for_runtime_filter=false`: they
get no dependency and no timer, they only apply filters that already arrived.
### Release note
A runtime filter target scan now waits at most 20ms
(`runtime_filter_wait_time_ms`) for the filter after the join build side is
ready, instead of a per-query estimated time, and filters that arrive later are
also applied at the exchange nodes between the target scan and the join.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
FE: `RuntimeFilterTranslatorExchangeTest` (new, 3 tests),
`RuntimeFilterTranslatorBucketPruneTest` (7), `RuntimeFilterTest` (50) all
pass; `build.sh --fe` with checkstyle passes. BE: modified files compile with
the BE compile commands and `check-build-hygiene.sh` passes; a full BE
build/regression run was not executed in the development workspace (stale BE
build directory).
- Behavior changed:
- [ ] No.
- [x] Yes. <!-- Explain the behavior change -->
Default `runtime_filter_wait_time_ms` changes from 1000ms to 20ms;
runtime filters are also applied at exchange nodes between the target scan and
the builder join.
- Does this need documentation?
- [ ] No.
- [x] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
The default value of `runtime_filter_wait_time_ms` is user-visible.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should
merge into -->
--
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]
