ApsaraDB for SelectDB:Runtime filter

How it works

In a hash join, the right table is loaded first to build a hash table. A runtime filter captures the key values from that hash table and sends the filter to the OlapScanNode scanning the left table. The OlapScanNode discards non-matching rows before they reach the HashJoinNode.

The following example uses two tables: T1 (a fact table with 1,000,000 rows) and T2 (a dimension table with 200 rows).

Without a runtime filter — all 1,000,000 rows from T1 flow up to the join:

HashJoinNode
     |                    |
     | 1,000,000          | 200
     |                    |
OlapScanNode         OlapScanNode
     ^                    ^
     | 1,000,000          | 200
    T1 (fact)            T2 (dimension)

With a runtime filter applied at the scan layer — only 6,000 rows reach the join:

HashJoinNode
     |                    |
     | 6,000              | 200
     |                    |
OlapScanNode         OlapScanNode
     ^                    ^
     | 1,000,000          | 200
    T1 (fact)            T2 (dimension)

With the filter pushed further down to the storage engine — indexes prune the data before it is read:

HashJoinNode
     |                    |
     | 6,000              | 200
     |                    |
OlapScanNode         OlapScanNode
     ^                    ^
     | 6,000              | 200
    T1 (fact)            T2 (dimension)

Unlike predicate pushdown or partition pruning, the filter condition is not known at query planning time. It is computed from the actual data in the right table during execution, then broadcast to the OlapScanNode reading the left table.

Configure runtime filters

Use the following session variables to tune runtime filter behavior for specific queries.

-- By name (comma-separated, quoted)
SET runtime_filter_type = "BLOOM_FILTER,IN,MIN_MAX";

-- Equivalent numeric form (1=IN, 2=BLOOM_FILTER, 4=MIN_MAX)
SET runtime_filter_type = 7;

Selectivity = HashJoinNode cardinality / HashJoinNode left child cardinality

Verify that runtime filters are applied

Run EXPLAIN to view the query plan and confirm filters are generated and consumed on the expected columns.

• Join side (filter generated): runtime filters: RF000[in] <- table.column

• Scan side (filter applied): runtime filters: RF000[in] -> table.column

Example:

CREATE TABLE test (t1 INT) DISTRIBUTED BY HASH (t1) BUCKETS 2;
INSERT INTO test VALUES (1), (2), (3), (4);

CREATE TABLE test2 (t2 INT) DISTRIBUTED BY HASH (t2) BUCKETS 2;
INSERT INTO test2 VALUES (3), (4), (5);

EXPLAIN SELECT t1 FROM test JOIN test2 WHERE test.t1 = test2.t2;

+-------------------------------------------------------------------+
| Explain String                                                    |
+-------------------------------------------------------------------+
| PLAN FRAGMENT 0                                                   |
|  OUTPUT EXPRS:`t1`                                                |
|                                                                   |
|   4:EXCHANGE                                                      |
|                                                                   |
| PLAN FRAGMENT 1                                                   |
|  OUTPUT EXPRS:                                                    |
|   PARTITION: HASH_PARTITIONED: `default_cluster:ssb`.`test`.`t1`  |
|                                                                   |
|   2:HASH JOIN                                                     |
|   |  join op: INNER JOIN (BUCKET_SHUFFLE)                         |
|   |  equal join conjunct: `test`.`t1` = `test2`.`t2`              |
|   |  runtime filters: RF000[in] <- `test2`.`t2`                   |
|   |                                                               |
|   |----3:EXCHANGE                                                 |
|   |                                                               |
|   0:OlapScanNode                                                  |
|      TABLE: test                                                  |
|      runtime filters: RF000[in] -> `test`.`t1`                    |
|                                                                   |
| PLAN FRAGMENT 2                                                   |
|  OUTPUT EXPRS:                                                    |
|   PARTITION: HASH_PARTITIONED: `default_cluster:ssb`.`test2`.`t2` |
|                                                                   |
|   1:OlapScanNode                                                  |
|      TABLE: test2                                                 |
+-------------------------------------------------------------------+

Join side — 2:HASH JOIN in PLAN FRAGMENT 1 generates an IN predicate (RF000) from test2.t2. The values are only known at runtime.

Scan side — 0:OlapScanNode applies RF000 to filter test.t1 before rows reach the join.

Running the query returns [3, 4] — only the two rows where t1 = t2.

Check filter effectiveness with a profile:

Enable profiling, then run the query:

SET enable_profile = true;

In the profile, look for the RuntimeFilter section under OLAP_SCAN_NODE:

RuntimeFilter:in:
    HasPushDownToEngine: true    -- filter reached the storage engine
    AWaitTimeCost: 0ns           -- no wait; filter arrived before scan started
    EffectTimeCost: 2.76ms       -- time spent applying the filter

And the filtering result:

RowsVectorPredFiltered: 9,320,008   -- rows discarded by the filter
VectorPredEvalTime: 364.39ms        -- time spent on filter evaluation

A high RowsVectorPredFiltered count confirms the filter is effective. HasPushDownToEngine: true confirms it reached the storage layer.

Planning rules

Runtime filters are generated and applied according to the following rules. Violations of these rules cause filters to be skipped or produce incorrect results.

Generation rules:

• Filters are generated only for equality conditions in JOIN ON clauses. NULL-safe equality (<=>) is excluded because null values in the left table could be incorrectly filtered.

• The source expression type cannot be HLL or BITMAP.

• Source and target expressions cannot be constants.

• Source and target expressions cannot be the same expression.

• Types of the source and target expressions must match (bloom filters are hash-based). If types differ, the system attempts to cast the target expression to the source type.

• Filters from PlanNode.Conjuncts are not pushed down — these can produce incorrect results. For example, when an IN subquery is rewritten as a join, the system stores the auto-generated JOIN condition in PlanNode.Conjuncts; applying a runtime filter there may cause rows to be dropped from the results.

Pushdown rules:

• Filters can only be pushed down to OlapScanNode. Other scan node types are not supported.

• Filters cannot be pushed to the left table of left outer joins, full outer joins, or anti-joins.

• The target expression must reference a column that exists in the original base table.

• The target expression cannot contain null-checking expressions such as COALESCE, IFNULL, or CASE.

Column conduction rules:

• If the JOIN ON clause contains A.k = B.k AND B.k = C.k, the filter for C.k can be pushed down to B.k only — not propagated further to A.k.

• If the JOIN ON clause contains A.a + B.b = C.c and A.a is equivalent to B.a, A.a can be replaced with B.a and the filter pushed to B. If A.a and B.a are not equivalent, the filter cannot be pushed to B because the target expression must be bound to a single left table.