[Enhancement] pushdown join runtime filter into storage layer #55124
+1,237
−19
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
silverbullet233
force-pushed
the
rf_pushdown_internal
branch
from
f43f35f to
d306e69
Compare
silverbullet233
changed the title
Signed-off-by: silverbullet233 <[email protected]>
silverbullet233
force-pushed
the
rf_pushdown_internal
branch
from
d306e69 to
2753e51
Compare
[BE Incremental Coverage Report]❌ fail : 57 / 219 (26.03%) file detail
|
Signed-off-by: silverbullet233 <[email protected]>
silverbullet233
force-pushed
the
rf_pushdown_internal
branch
from
74234bc to
d62440d
Compare
Signed-off-by: silverbullet233 <[email protected]>
[FE Incremental Coverage Report]✅ pass : 0 / 0 (0%) |
[Java-Extensions Incremental Coverage Report]✅ pass : 0 / 0 (0%) |
satanson
reviewed
Jan 16, 2025
| @@ -459,6 +471,155 @@ struct WithModuloArg { | |||
| } | |||
| } | |||
| }; | |||
|
|
|||
| template <TRuntimeFilterLayoutMode::type M> | |||
| struct HashValueComputeWithSelection { | |||
There was a problem hiding this comment.
This code is tedious and error-prone, seems duplicate, suggest to unite it with HashValueCompute into one.
| protected: | ||
| RuntimeFilterProbeDescriptor* _rf_desc; | ||
| const JoinRuntimeFilter* _rf = nullptr; | ||
| ColumnId _column_id; |
There was a problem hiding this comment.
do we support runtime filter whose probe_expr is other Expr except ColumnRef and/or multi-column partition exprs?
stdpain
reviewed
Jan 16, 2025
| @@ -1554,4 +1554,8 @@ CONF_mBool(avro_ignore_union_type_tag, "false"); | |||
| // default batch size for simdjson lib | |||
| CONF_mInt32(json_parse_many_batch_size, "1000000"); | |||
| CONF_mBool(enable_dynamic_batch_size_for_json_parse_many, "true"); | |||
| CONF_mBool(enable_rf_pushdown, "true"); | |||
There was a problem hiding this comment.
Is it possiable add a session variable to control the behaviour?
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Why I'm doing:
Currently, the runtime filter of our scan operator is executed after the storage layer has read all the data. This means that we lose the opportunity to use storage layer optimization, such as late materialization.
What I'm doing:
In this PR, I support pushing the runtime filter down to the storage layer. When querying, it will be treated as a specific predicate and evaluated in the first stage of late materialization, so that we have the opportunity to use late materialization to reduce the amount of data read.
In addition, since each scan operator will generate multiple parallel io tasks, after pushing the runtime filter down, the number of threads evaluating the runtime filters in parallel will increase, which will also improve the computing efficiency to a certain extent.
Design points
somewhat different from ordinary single-column predicates, multiple runtime filter predicates may be related.
For example, when there are multiple runtime filters, in order to balance the execution efficiency, we do not need to execute each runtime filter once, but only select a few runtime filters with good filterability. Similarly, when their filtering effects are not good, we can choose to skip all of them.
our storage layer currently only supports single-column predicates, which is not suitable for the above scenario. so I added
RuntimeFilterPredicates, which is responsible for collecting all relevant runtime filters and selecting the appropriate runtime filters to execute based on the selectivity.RuntimeFilterPredicateswill be executed in the first stage of late materialization. If the filterability is good, we can read less data.When applying late materialization, we will only choose to read the columns with predicates first, and the columns required by the runtime filters may not have predicates. The implementation here is a bit complicated. In order to reduce the cost of modification, I used a tricky method and introduced
ColumnPlaceHolderPredicate, which is only used as a placeholder and will not produce any filtering effect.In the ChunkSource init stage, ColumnPlaceHolderPredicates are generated for the columns required by the runtime filter and placed in the PredicateTree. In this way, the existing column selection strategy can ensure that these columns will be read out in the first stage of delayed materialization.
Then, in SegmentIterator init stage, we remove the ColumnPlaceHolderPredicate by
ColumnPredicateRewriter, so that it will not generate additional overhead.Test Result
I tested on a tpch 1TB data set, using four 16c 64g BEs, each with two disks(each have 5000 iops and 140MB/s throughput), and the baseline code is
main-9a6445a.Below are the test results, and the query time unit in the table is milliseconds.
TODO: in this pr, I only support this enhancement in internal olap table under shared_nothing mode, will support other tables in other pr.
What type of PR is this:
Does this PR entail a change in behavior?
If yes, please specify the type of change:
Checklist:
Bugfix cherry-pick branch check: