Optimize TopK query of TableModel by RuntimeFilter#18204
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #18204 +/- ##
============================================
+ Coverage 42.39% 42.41% +0.01%
Complexity 343 343
============================================
Files 5354 5357 +3
Lines 379174 379484 +310
Branches 49149 49203 +54
============================================
+ Hits 160759 160964 +205
- Misses 218415 218520 +105 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Signed-off-by: Weihao Li <18110526956@163.com>
Signed-off-by: Weihao Li <18110526956@163.com>
Signed-off-by: Weihao Li <18110526956@163.com>
JackieTien97
left a comment
JackieTien97
left a comment
There was a problem hiding this comment.
Two blocking correctness issues are called out inline.
| return; | ||
| } | ||
| MergeSortKey peek = mergeSortHeap.peek(); | ||
| topKRuntimeFilter.updateThreshold(peek.tsBlock.getTimeByIndex(peek.rowIndex)); |
There was a problem hiding this comment.
[P1] Read the table time value instead of the synthetic TsBlock time column
For table-model blocks, the SQL time column is stored as a value column, while TableTopKOperator constructs the internal TsBlock time column from TIME_COLUMN_TEMPLATE, whose value is always 0. Consequently, this publishes 0 after the heap fills. For ASC queries over positive timestamps, scans can reject all later resources and return wrong rows; for DESC queries over positive timestamps, the filter does not tighten usefully, and negative timestamps can be mis-pruned. Please extract the threshold from the sole ORDER BY time value channel rather than TsBlock#getTimeByIndex.
| } | ||
| if (orderByTimeOnly && hasDirectRawDeviceTableScanChild(node)) { | ||
| // Single region: Output -> TopK -> Scan (no Exchange/root TopK above). | ||
| return FAKE_ROOT_TOPK_ID; |
There was a problem hiding this comment.
[P1] Do not reuse one filter ID for unrelated single-region TopKs
Every single-region TopK -> Scan in a query returns the same constant here. DataNodeQueryContext keys its filter map only by this string, so two independent TopK branches on the same DataNode share one TopKRuntimeFilter. One branch threshold can then prune the other branch; if their directions differ, whichever branch registers first also determines the direction for both. Please use the current TopK node own PlanNodeId for the single-region root, while retaining the actual shared root ID for fragments of the same distributed TopK, and add coverage with two sibling TopKs.
There was a problem hiding this comment.
Pull request overview
This PR adds a TopK runtime filter optimization for table-model ORDER BY time LIMIT k queries. It marks qualifying TopK -> (Device)TableScan plan structures during distributed planning, registers a shared time-threshold filter per query on each DataNode, and lets scans prune TsFiles/chunks/pages/rows that can no longer enter the final TopK result.
Changes:
- Introduce
TopKRuntimeFilterand propagate it from TopK operators to table scans (planner → operator generator → scan options). - Add a distributed-plan optimizer (
TopKRuntimeFilterOptimizer) to tag producer TopK nodes and their scan children with a shared root id for filter lookup/sharing. - Add config + extensive unit/integration tests covering optimizer marking, scan pruning behavior, and external TsFile/table-view paths.
Reviewed changes
Copilot reviewed 27 out of 27 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/planner/node/TopKNode.java | Adds runtime-filter metadata (ascending flag + source id) to TopK plan node, including clone/serde. |
| iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template | Adds enable_topk_runtime_filter configuration template entry (hot reload). |
| iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/QueryDataSourceRuntimeFilterTest.java | Unit tests for per-resource invalidation tracking in QueryDataSource. |
| iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterUtilsTest.java | Unit tests for “ORDER BY time only” detection. |
| iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterTreeViewPlanTest.java | Plan-shape test ensuring TopK and scans are marked for tree-view scans. |
| iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizerTest.java | Unit tests for optimizer marking logic across single/multi-region plans and scan types. |
| iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileTableScanOperatorRuntimeFilterTest.java | Tests runtime-filter stop behavior for external TsFile scanning across devices. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/QueryDataSource.java | Adds runtime-filter pruning tracking and resource-level qualification checks. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterUtils.java | Implements helper to detect time-only ordering schemes. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizer.java | New distributed optimization pass to mark TopK+scan structures and assign shared root ids. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java | Adds runtime-filter source id to scans and includes it in clone/serde. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java | Runs TopK runtime-filter marking after Exchange insertion when enabled by config. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/parameter/SeriesScanOptions.java | Carries TopKRuntimeFilter through scan options (builder + getter). |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java | Prints runtime-filter/TopK optimization markers in plan graph output. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java | Registers/looks up shared runtime filter and attaches it to scan options. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java | Applies runtime filter at file/chunk/page/TsBlock level and supports early exhaustion. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileTableScanOperator.java | Adjusts runtime-filter stop semantics for external TsFile scanning. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractTableScanOperator.java | Stops scan when runtime filter prunes all files; initializes tracking when filter present. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java | Stores shared per-query runtime filters keyed by root TopK id. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java | Loads/hot-reloads the new enable_topk_runtime_filter configuration. |
| iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java | Adds enableTopKRuntimeFilter config flag (volatile, default true). |
| iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/filter/TopKRuntimeFilterTest.java | Unit tests for the TopKRuntimeFilter threshold logic and range checks. |
| iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/plan/planner/TableOperatorGenerator.java | Plumbs runtime filter into TopK operator construction. |
| iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/TopKOperator.java | Updates shared runtime filter threshold from heap-top once TopK heap is full. |
| iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/TableTopKOperator.java | Adds ctor overload to accept and pass through runtime filter. |
| iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/filter/TopKRuntimeFilter.java | New shared filter class using an atomic time threshold for pruning. |
| integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTopKRuntimeFilterScanPathIT.java | Integration tests for ORDER BY time LIMIT on tree-view and read_tsfile paths. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| @Override | ||
| protected boolean shouldStopScanByRuntimeFilter() { | ||
| // Each device uses its own QueryDataSource, exhausting files for the current device must not | ||
| // stop scanning remaining devices in the same external TsFile task. | ||
| return false; | ||
| } |
Signed-off-by: Weihao Li <18110526956@163.com>
Signed-off-by: Weihao Li <18110526956@163.com>
| TopKNode topKNode = (TopKNode) node.clone(); | ||
|
|
||
| boolean orderByTimeOnly = TopKRuntimeFilterUtils.isOrderByTimeOnly(node.getOrderingScheme()); | ||
| String effectiveRootTopKId = resolveEffectiveRootTopKId(node, rootTopKId); |
There was a problem hiding this comment.
what if
outer TopK
-> Union / Join
-> inner TopK A -> Scan
-> inner TopK B -> Scan
| if (currentDeviceIndex < deviceCount - 1) { | ||
| return false; | ||
| } | ||
| return super.shouldStopScanByRuntimeFilter(); |
There was a problem hiding this comment.
super class's queryDataSource belongs to first device, not the last device.
Signed-off-by: Weihao Li <18110526956@163.com>
Signed-off-by: Weihao Li <18110526956@163.com>
Signed-off-by: Weihao Li <18110526956@163.com>
|
Setup: 100,000 devices, 20 measurements per device, 5s sampling interval per device, 100,000 rows written per device.
Theoretical speedup: After the optimization, at most limit devices’ data need to be read, instead of a full scan. Speedup is:
100,000 / limitResult:
