Hi all,
I'd like to share a query optimization we are working on for the table model:
TopK Runtime Filter for full-table ORDER BY time LIMIT k queries.
Typical queries:
SELECT * FROM t1 ORDER BY time DESC LIMIT 10;
SELECT * FROM t1 ORDER BY time ASC LIMIT 10;
1. Current status
Optimization path (before this change)
Logical plan: MergeLimitWithSort merges Limit + Sort into TopKNode
LIMIT pushdown: PushLimitOffsetIntoTableScan pushes LIMIT into
DeviceTableScanNode; with multiple devices, pushLimitToEachDevice = true
Distributed plan: TableDistributedPlanGenerator splits Scan by Region;
MergeLimitWithMergeSort merges Limit + MergeSort into TopK; AddExchangeNodes
inserts Exchange
Root TopK merge: each DataNode’s local TopK results are exchanged and merged by
the root TopK
Problem
When there are many devices (Tag dimension), each device returns up to k rows,
so Scan may read about k × device_count rows before the global TopK keeps only
k. When the device count is large, Scan I/O and CPU cost grow a lot, even
though the final result only needs k rows.
2. Optimization idea
We adopt a two-phase design, this helps to reduce read load a lot :
Optimize- Add TopKRuntimeFilterOptimizer to mark if TopK/TableScan need create
or set RuntimeFilter.
Execution-When generate operators, create a shared RuntimeFilter; TopK updates
the threshold; Scan prunes read path using RuntimeFilter.
3. Details
3.1 Plan-time pipeline
In TableDistributedPlanner.generateDistributedPlanWithOptimize():
1. TableDistributedPlanGenerator.genResult() // split by Region
2. DistributedOptimizeFactory optimizers // e.g. MergeLimitWithMergeSort
3. AddExchangeNodes() // insert Exchange
4. if enable_topk_runtime_filter == true:
TopKRuntimeFilterOptimizer.optimize() // MUST run after Exchange insertion
5. SubPlanGenerator.splitToSubPlan() // split into Fragments
Trigger conditions (all required):
enable_topk_runtime_filter = true
Single-column ORDER BY time
TopK subtree contains a local DeviceTableScanNode
Typical distributed plan:
Fragment-0 (Root):
TopK ← subtree has only Exchange → NOT marked
├── Exchange ── Fragment-1
└── Exchange ── Fragment-2
Fragment-1:
TopK (TOPN OPT) ← local TopK + Scan → marked
└── Limit
└── DeviceTableScan (TOPN OPT: {topKId})
3.2 Runtime pipeline
1. TableOperatorGenerator -> vistTopK -> produce RuntimeFilter and record it
into context, visitTableScan -> read RuntimeFilter from context and set it into
TableScanOperator
2. TableTopKOperator -> when the matained heap of size k was full, call
updateThreshold(peek.time) of RuntimeFilter
3. TableScanOperator -> use RuntimeFilter to prune read load
** mayQualifyRange(start, end) — skip resources / files / chunks / pages
** mayQualify(time) — row-level pruning
Yours,
Weihao Li
