JackieTien97 commented on code in PR #18204: URL: https://github.com/apache/iotdb/pull/18204#discussion_r3578790516
########## iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizer.java: ########## @@ -0,0 +1,117 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations; + +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.TopKNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode; + +/** + * <b>Optimization phase:</b> Distributed plan planning (after exchange nodes are inserted). + * + * <p>Marks the {@code TopK + DeviceTableScan} structure for TopK runtime filter. + * + * <p>The topmost TopK establishes the <b>root TopK id</b>. In the single-region case ({@code Output + * -> TopK -> Scan}), a static {@link #FAKE_ROOT_TOPK_ID} is used because there is no root TopK + * above Exchange. A per-region TopK that sits directly on top of {@link DeviceTableScanNode}(s) + * becomes the runtime filter producer, and both that TopK and its scan children are tagged with the + * <b>root</b> TopK id (not the region TopK's own id). Because {@code DataNodeQueryContext} is + * shared by all fragment instances of the same query on one DataNode, using the root id lets + * multiple regions on the same DataNode share a single filter. + */ +public class TopKRuntimeFilterOptimizer implements PlanOptimizer { + + /** + * Shared root id for single-region plans ({@code Output -> TopK -> Scan}) without a root TopK. + */ + public static final String FAKE_ROOT_TOPK_ID = "fake-topk-runtime-filter-root"; + + @Override + public PlanNode optimize(PlanNode plan, Context context) { + if (!context.getAnalysis().isQuery()) { + return plan; + } + return plan.accept(new Rewriter(), null); + } + + /** Context carries the root TopK id string, or {@code null} until the first TopK is seen. */ + private static class Rewriter implements PlanVisitor<PlanNode, String> { + + @Override + public PlanNode visitPlan(PlanNode node, String rootTopKId) { + PlanNode newNode = node.clone(); + for (PlanNode child : node.getChildren()) { + newNode.addChild(child.accept(this, rootTopKId)); + } + return newNode; + } + + @Override + public PlanNode visitTopK(TopKNode node, String rootTopKId) { + TopKNode topKNode = (TopKNode) node.clone(); + + boolean orderByTimeOnly = TopKRuntimeFilterUtils.isOrderByTimeOnly(node.getOrderingScheme()); + String effectiveRootTopKId = resolveEffectiveRootTopKId(node, rootTopKId, orderByTimeOnly); + + // A TopK qualifies as a runtime filter producer only when it orders by time and directly + // parents raw DeviceTableScan(s). Detect qualification and tag both the producer TopK and its + // scan children (with the root id) in a single pass over the children. + for (PlanNode child : node.getChildren()) { + boolean isRawDeviceTableScan = + child instanceof DeviceTableScanNode && !(child instanceof AggregationTableScanNode); + if (orderByTimeOnly && isRawDeviceTableScan) { + if (topKNode.getTopKRuntimeFilterSourceId() == null) { + topKNode.setTopKRuntimeFilterSourceId(effectiveRootTopKId); + } + DeviceTableScanNode scanNode = (DeviceTableScanNode) child.clone(); + scanNode.setTopKRuntimeFilterSourceId(effectiveRootTopKId); + topKNode.addChild(scanNode); + } else { + topKNode.addChild(child.accept(this, effectiveRootTopKId)); + } + } + return topKNode; + } + + private static String resolveEffectiveRootTopKId( + TopKNode node, String rootTopKId, boolean orderByTimeOnly) { + if (rootTopKId != null) { + return rootTopKId; + } + if (orderByTimeOnly && hasDirectRawDeviceTableScanChild(node)) { + // Single region: Output -> TopK -> Scan (no Exchange/root TopK above). + return FAKE_ROOT_TOPK_ID; Review 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. ########## iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/TopKOperator.java: ########## @@ -226,6 +249,14 @@ public TsBlock next() throws Exception { return null; } + private void updateTopKRuntimeFilter() { + if (topKRuntimeFilter == null || mergeSortHeap.getHeapSize() < topValue) { + return; + } + MergeSortKey peek = mergeSortHeap.peek(); + topKRuntimeFilter.updateThreshold(peek.tsBlock.getTimeByIndex(peek.rowIndex)); Review 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`. -- 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]
