This is an automated email from the ASF dual-hosted git repository. Wei-hao-Li pushed a commit to branch lwh/topK in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b0bf9b38ed0a273d97ec1e917809f53eca035f5f Author: Weihao Li <[email protected]> AuthorDate: Tue Jul 14 17:18:48 2026 +0800 finish Signed-off-by: Weihao Li <[email protected]> --- .../execution/operator/process/TopKOperator.java | 18 ++-- .../calc/plan/planner/TableOperatorGenerator.java | 9 +- .../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 14 +++ .../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 12 +++ .../execution/operator/source/SeriesScanUtil.java | 26 +++--- .../relational/AbstractTableScanOperator.java | 24 ++--- .../planner/DataNodeTableOperatorGenerator.java | 47 ++-------- .../plan/planner/LocalExecutionPlanner.java | 2 - .../plan/planner/TopKRuntimeFilterBinder.java | 60 ------------ .../plan/planner/plan/node/PlanGraphPrinter.java | 9 +- .../distribute/TableDistributedPlanner.java | 6 +- .../planner/node/DeviceTableScanNode.java | 27 ++---- .../optimizations/TopKRuntimeFilterOptimizer.java | 71 +++++++------- .../optimizations/TopKRuntimeFilterUtils.java | 38 -------- .../dataregion/read/QueryDataSource.java | 85 +++++++++++------ .../TopKRuntimeFilterOptimizerTest.java | 102 ++++++++++++++++++--- .../optimizations/TopKRuntimeFilterUtilsTest.java | 60 ++++++++++++ .../read/QueryDataSourceRuntimeFilterTest.java | 76 +++++++++++++++ .../conf/iotdb-system.properties.template | 8 ++ .../plan/relational/planner/node/TopKNode.java | 69 +++++++------- 20 files changed, 442 insertions(+), 321 deletions(-) diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/TopKOperator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/TopKOperator.java index 54d58e85bdd..45ac6f4a7c1 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/TopKOperator.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/TopKOperator.java @@ -249,6 +249,14 @@ public abstract class TopKOperator implements ProcessOperator { return null; } + private void updateTopKRuntimeFilter() { + if (topKRuntimeFilter == null || mergeSortHeap.getHeapSize() < topValue) { + return; + } + MergeSortKey peek = mergeSortHeap.peek(); + topKRuntimeFilter.updateThreshold(peek.tsBlock.getTimeByIndex(peek.rowIndex)); + } + @Override public void close() throws Exception { for (int i = childIndex; i < childrenOperators.size(); i++) { @@ -448,14 +456,4 @@ public abstract class TopKOperator implements ProcessOperator { getOperator(i).close(); childrenOperators.set(i, null); } - - private void updateTopKRuntimeFilter() { - if (topKRuntimeFilter == null || mergeSortHeap.getHeapSize() < topValue) { - return; - } - MergeSortKey peek = mergeSortHeap.peek(); - if (peek != null) { - topKRuntimeFilter.updateThreshold(peek.tsBlock.getTimeByIndex(peek.rowIndex)); - } - } } diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/plan/planner/TableOperatorGenerator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/plan/planner/TableOperatorGenerator.java index e44379c9c5d..e41ef0f2eb9 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/plan/planner/TableOperatorGenerator.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/plan/planner/TableOperatorGenerator.java @@ -889,6 +889,7 @@ public abstract class TableOperatorGenerator< public Operator visitTopK(TopKNode node, C context) { CommonOperatorContext operatorContext = addOperatorContext(context, node.getPlanNodeId(), TableTopKOperator.class.getSimpleName()); + TopKRuntimeFilter topKRuntimeFilter = registerRuntimeFilter(context, node); List<Operator> children = new ArrayList<>(node.getChildren().size()); for (PlanNode child : node.getChildren()) { children.add(this.process(child, context)); @@ -912,17 +913,13 @@ public abstract class TableOperatorGenerator< node.getOrderingScheme().getOrderingList(), sortItemIndexList, sortItemDataTypeList), (int) node.getCount(), node.isChildrenDataInOrder(), - getTopKRuntimeFilter(context, node)); + topKRuntimeFilter); } - protected TopKRuntimeFilter getTopKRuntimeFilter(C context, TopKNode node) { + protected TopKRuntimeFilter registerRuntimeFilter(C context, TopKNode node) { return null; } - protected TopKRuntimeFilter getTopKRuntimeFilter(C context) { - return getTopKRuntimeFilter(context, null); - } - protected List<TSDataType> getOutputColumnTypes(PlanNode node, ITableTypeProvider typeProvider) { return node.getOutputSymbols().stream() .map(s -> getTSDataType(typeProvider.getTableModelType(s))) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java index 7f151e450d2..cf901ebd2db 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java @@ -582,6 +582,12 @@ public class IoTDBConfig { private volatile boolean enableTsFileValidation = false; + /** + * Whether to enable the TopK runtime filter optimization for table-model {@code ORDER BY time + * LIMIT k} queries. Hot-reloadable; set false to fall back to the original execution path. + */ + private volatile boolean enableTopKRuntimeFilter = true; + /** The size of candidate compaction task queue. */ private int candidateCompactionTaskQueueSize = 50; @@ -4418,6 +4424,14 @@ public class IoTDBConfig { this.enableTsFileValidation = enableTsFileValidation; } + public boolean isEnableTopKRuntimeFilter() { + return enableTopKRuntimeFilter; + } + + public void setEnableTopKRuntimeFilter(boolean enableTopKRuntimeFilter) { + this.enableTopKRuntimeFilter = enableTopKRuntimeFilter; + } + public long getInnerCompactionTaskSelectionModsFileThreshold() { return innerCompactionTaskSelectionModsFileThreshold; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index e3862bc0e9b..1ab0c22bbf6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -707,6 +707,11 @@ public class IoTDBDescriptor { properties.getProperty( "enable_tsfile_validation", String.valueOf(conf.isEnableTsFileValidation())))); + conf.setEnableTopKRuntimeFilter( + Boolean.parseBoolean( + properties.getProperty( + "enable_topk_runtime_filter", String.valueOf(conf.isEnableTopKRuntimeFilter())))); + conf.setCandidateCompactionTaskQueueSize( Integer.parseInt( properties.getProperty( @@ -2216,6 +2221,13 @@ public class IoTDBDescriptor { ConfigurationFileUtils.getConfigurationDefaultValue( "enable_tsfile_validation")))); + conf.setEnableTopKRuntimeFilter( + Boolean.parseBoolean( + properties.getProperty( + "enable_topk_runtime_filter", + ConfigurationFileUtils.getConfigurationDefaultValue( + "enable_topk_runtime_filter")))); + // update wal config long prevDeleteWalFilesPeriodInMs = conf.getDeleteWalFilesPeriodInMs(); loadWALHotModifiedProps(properties); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java index e17d0d0fa22..52ddf505a5e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java @@ -207,6 +207,9 @@ public class SeriesScanUtil implements Accountable { * @param dataSource the query data source */ public void initQueryDataSource(QueryDataSource dataSource) { + if (scanOptions.getTopKRuntimeFilter() != null) { + dataSource.initRuntimeFilterTracking(); + } dataSource.fillOrderIndexes(deviceID, orderUtils.getAscending()); this.dataSource = dataSource; @@ -353,11 +356,13 @@ public class SeriesScanUtil implements Accountable { return filter.mayQualifyRange(statistics.getStartTime(), statistics.getEndTime()); } - private void skipByTopKRuntimeFilter( + private boolean skipByTopKRuntimeFilter( Statistics<? extends Serializable> statistics, Runnable skip) { if (!mayQualifyRuntimeFilterRange(statistics)) { skip.run(); + return true; } + return false; } @SuppressWarnings("squid:S3740") @@ -440,8 +445,7 @@ public class SeriesScanUtil implements Accountable { return; } - skipByTopKRuntimeFilter(firstChunkMetadata.getStatistics(), this::skipCurrentChunk); - if (firstChunkMetadata == null) { + if (skipByTopKRuntimeFilter(firstChunkMetadata.getStatistics(), this::skipCurrentChunk)) { return; } @@ -1461,8 +1465,7 @@ public class SeriesScanUtil implements Accountable { return; } - skipByTopKRuntimeFilter(firstPageReader.getStatistics(), this::skipCurrentPage); - if (firstPageReader == null) { + if (skipByTopKRuntimeFilter(firstPageReader.getStatistics(), this::skipCurrentPage)) { return; } @@ -1952,8 +1955,7 @@ public class SeriesScanUtil implements Accountable { return; } - skipByTopKRuntimeFilter(firstTimeSeriesMetadata.getStatistics(), this::skipCurrentFile); - if (firstTimeSeriesMetadata == null) { + if (skipByTopKRuntimeFilter(firstTimeSeriesMetadata.getStatistics(), this::skipCurrentFile)) { return; } @@ -2467,8 +2469,7 @@ public class SeriesScanUtil implements Accountable { if (dataSource.isSeqSatisfied( deviceID, curSeqFileIndex, scanOptions.getGlobalTimeFilter(), false)) { if (filter == null - || dataSource.isSeqSatisfiedByRuntimeFilter( - deviceID, curSeqFileIndex, filter, false)) { + || dataSource.isSeqSatisfiedByRuntimeFilter(curSeqFileIndex, filter, false)) { break; } dataSource.setSeqTsFileResourceInvalidated(curSeqFileIndex); @@ -2493,7 +2494,7 @@ public class SeriesScanUtil implements Accountable { if (dataSource.isUnSeqSatisfied( deviceID, curUnseqFileIndex, scanOptions.getGlobalTimeFilter(), false)) { if (filter == null - || dataSource.isUnSeqSatisfiedByRuntimeFilter(curUnseqFileIndex, filter, false)) { + || dataSource.isUnSeqSatisfiedByRuntimeFilter(curUnseqFileIndex, filter)) { break; } dataSource.setUnseqTsFileResourceInvalidated(curUnseqFileIndex); @@ -2623,8 +2624,7 @@ public class SeriesScanUtil implements Accountable { if (dataSource.isSeqSatisfied( deviceID, curSeqFileIndex, scanOptions.getGlobalTimeFilter(), false)) { if (filter == null - || dataSource.isSeqSatisfiedByRuntimeFilter( - deviceID, curSeqFileIndex, filter, false)) { + || dataSource.isSeqSatisfiedByRuntimeFilter(curSeqFileIndex, filter, false)) { break; } dataSource.setSeqTsFileResourceInvalidated(curSeqFileIndex); @@ -2649,7 +2649,7 @@ public class SeriesScanUtil implements Accountable { if (dataSource.isUnSeqSatisfied( deviceID, curUnseqFileIndex, scanOptions.getGlobalTimeFilter(), false)) { if (filter == null - || dataSource.isUnSeqSatisfiedByRuntimeFilter(curUnseqFileIndex, filter, false)) { + || dataSource.isUnSeqSatisfiedByRuntimeFilter(curUnseqFileIndex, filter)) { break; } dataSource.setUnseqTsFileResourceInvalidated(curUnseqFileIndex); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractTableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractTableScanOperator.java index 383be946624..1c023998042 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractTableScanOperator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractTableScanOperator.java @@ -250,8 +250,11 @@ public abstract class AbstractTableScanOperator extends AbstractSeriesScanOperat @Override public void initQueryDataSource(IQueryDataSource dataSource) { this.queryDataSource = (QueryDataSource) dataSource; - if (currentDeviceIndex < deviceCount) { - setupCurrentDeviceScan(); + if (seriesScanOptions.getTopKRuntimeFilter() != null) { + queryDataSource.initRuntimeFilterTracking(); + } + if (this.seriesScanUtil != null) { + this.seriesScanUtil.initQueryDataSource(queryDataSource); } this.resultTsBlockBuilder = new TsBlockBuilder(getResultDataTypes()); this.resultTsBlockBuilder.setMaxTsBlockLineNumber(this.maxTsBlockLineNum); @@ -266,7 +269,14 @@ public abstract class AbstractTableScanOperator extends AbstractSeriesScanOperat } currentDeviceIndex++; if (currentDeviceIndex < deviceCount) { - setupCurrentDeviceScan(); + // construct AlignedSeriesScanUtil for next device + constructAlignedSeriesScanUtil(); + + // reset QueryDataSource + queryDataSource.reset(); + this.seriesScanUtil.initQueryDataSource(queryDataSource); + this.operatorContext.recordSpecifiedInfo( + CommonOperatorUtils.CURRENT_DEVICE_INDEX_STRING, Integer.toString(currentDeviceIndex)); } } @@ -275,14 +285,6 @@ public abstract class AbstractTableScanOperator extends AbstractSeriesScanOperat return seriesScanOptions.getTopKRuntimeFilter() != null && !queryDataSource.hasValidResource(); } - private void setupCurrentDeviceScan() { - constructAlignedSeriesScanUtil(); - queryDataSource.reset(); - this.seriesScanUtil.initQueryDataSource(queryDataSource); - this.operatorContext.recordSpecifiedInfo( - CommonOperatorUtils.CURRENT_DEVICE_INDEX_STRING, Integer.toString(currentDeviceIndex)); - } - protected void constructAlignedSeriesScanUtil() { if (this.deviceEntries.isEmpty() || currentDeviceIndex >= deviceCount) { // no need to construct SeriesScanUtil, hasNext will return false diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java index e482bf5f9c0..583f0442b41 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java @@ -25,7 +25,6 @@ import org.apache.iotdb.calc.execution.operator.Operator; import org.apache.iotdb.calc.execution.operator.process.FilterAndProjectOperator; import org.apache.iotdb.calc.execution.operator.process.LimitOperator; import org.apache.iotdb.calc.execution.operator.process.OffsetOperator; -import org.apache.iotdb.calc.execution.operator.process.TableTopKOperator; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.LastDescAccumulator; import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.TableAggregator; import org.apache.iotdb.calc.execution.relational.ColumnTransformerBuilder; @@ -188,7 +187,6 @@ import java.util.stream.Collectors; import static com.google.common.base.Preconditions.checkArgument; import static java.util.Objects.requireNonNull; -import static org.apache.iotdb.calc.execution.operator.process.join.merge.MergeSortComparator.getComparatorForTable; import static org.apache.iotdb.calc.plan.relational.planner.ir.GlobalTimePredicateExtractVisitor.isTimeColumn; import static org.apache.iotdb.calc.utils.constant.SqlConstant.AVG; import static org.apache.iotdb.calc.utils.constant.SqlConstant.COUNT; @@ -232,11 +230,14 @@ public class DataNodeTableOperatorGenerator */ private TopKRuntimeFilter registerTopKRuntimeFilterForTopK( TopKNode node, LocalExecutionPlanContext context) { - if (node == null || !node.isUseTopKRuntimeFilter() || context.dataNodeQueryContext == null) { + if (node == null + || node.getTopKRuntimeFilterSourceId() == null + || context.dataNodeQueryContext == null) { return null; } return context.dataNodeQueryContext.registerTopKRuntimeFilter( - node.getPlanNodeId().getId(), new TopKRuntimeFilter(node.isTopKRuntimeFilterAscending())); + node.getTopKRuntimeFilterSourceId().getId(), + new TopKRuntimeFilter(node.isTopKRuntimeFilterAscending())); } /** @@ -247,15 +248,15 @@ public class DataNodeTableOperatorGenerator DeviceTableScanNode scanNode, LocalExecutionPlanContext context) { if (scanNode == null || context.dataNodeQueryContext == null - || !scanNode.getTopKRuntimeFilterSourceId().isPresent()) { + || scanNode.getTopKRuntimeFilterSourceId() == null) { return null; } return context.dataNodeQueryContext.getTopKRuntimeFilter( - scanNode.getTopKRuntimeFilterSourceId().get().getId()); + scanNode.getTopKRuntimeFilterSourceId()); } @Override - protected TopKRuntimeFilter getTopKRuntimeFilter( + protected TopKRuntimeFilter registerRuntimeFilter( LocalExecutionPlanContext context, TopKNode node) { return registerTopKRuntimeFilterForTopK(node, context); } @@ -274,38 +275,6 @@ public class DataNodeTableOperatorGenerator } } - @Override - public Operator visitTopK(TopKNode node, LocalExecutionPlanContext context) { - TopKRuntimeFilter filter = registerTopKRuntimeFilterForTopK(node, context); - - CommonOperatorContext operatorContext = - addOperatorContext(context, node.getPlanNodeId(), TableTopKOperator.class.getSimpleName()); - List<Operator> children = new ArrayList<>(node.getChildren().size()); - for (PlanNode child : node.getChildren()) { - children.add(this.process(child, context)); - } - List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTableTypeProvider()); - int sortItemsCount = node.getOrderingScheme().getOrderBy().size(); - - List<Integer> sortItemIndexList = new ArrayList<>(sortItemsCount); - List<TSDataType> sortItemDataTypeList = new ArrayList<>(sortItemsCount); - genSortInformation( - node.getOutputSymbols(), - node.getOrderingScheme(), - sortItemIndexList, - sortItemDataTypeList, - context.getTableTypeProvider()); - return new TableTopKOperator( - operatorContext, - children, - dataTypes, - getComparatorForTable( - node.getOrderingScheme().getOrderingList(), sortItemIndexList, sortItemDataTypeList), - (int) node.getCount(), - node.isChildrenDataInOrder(), - filter); - } - @Override protected String getSortTmpDir(CommonOperatorContext operatorContext) { OperatorContext dataNodeOperatorContext = (OperatorContext) operatorContext; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java index 0cb2b157051..a093c7832fd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java @@ -112,8 +112,6 @@ public class LocalExecutionPlanner { LocalExecutionPlanContext context = new LocalExecutionPlanContext(types, instanceContext, dataNodeQueryContext); - TopKRuntimeFilterBinder.bind(plan, dataNodeQueryContext); - Operator root = generateOperator(instanceContext, context, plan); PipelineMemoryEstimator memoryEstimator = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TopKRuntimeFilterBinder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TopKRuntimeFilterBinder.java deleted file mode 100644 index 70b2e62abe5..00000000000 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TopKRuntimeFilterBinder.java +++ /dev/null @@ -1,60 +0,0 @@ -/* - * 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.planner; - -import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; -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.execution.fragment.DataNodeQueryContext; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor; - -/** - * Binds {@link TopKRuntimeFilter} runtime objects according to plan marks set by {@link - * org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.TopKRuntimeFilterOptimizer}. - */ -public class TopKRuntimeFilterBinder { - - private TopKRuntimeFilterBinder() {} - - public static void bind(PlanNode planRoot, DataNodeQueryContext dataNodeQueryContext) { - planRoot.accept(new Binder(), dataNodeQueryContext); - } - - private static class Binder implements PlanVisitor<Void, DataNodeQueryContext> { - - @Override - public Void visitPlan(PlanNode node, DataNodeQueryContext context) { - for (PlanNode child : node.getChildren()) { - child.accept(this, context); - } - return null; - } - - @Override - public Void visitTopK(TopKNode node, DataNodeQueryContext context) { - if (node.isUseTopKRuntimeFilter()) { - context.registerTopKRuntimeFilter( - node.getPlanNodeId().getId(), - new TopKRuntimeFilter(node.isTopKRuntimeFilterAscending())); - } - return visitPlan(node, context); - } - } -} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java index 5dd36542656..f0d2bc134d7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java @@ -698,9 +698,10 @@ public class PlanGraphPrinter implements PlanVisitor<List<String>, PlanGraphPrin boxValue.add( String.format( "PushDownLimitToEachDevice: %s", deviceTableScanNode.isPushLimitToEachDevice())); - deviceTableScanNode - .getTopKRuntimeFilterSourceId() - .ifPresent(sourceId -> boxValue.add(String.format("TOPN OPT: %s", sourceId.getId()))); + String topKRuntimeFilterSourceId = deviceTableScanNode.getTopKRuntimeFilterSourceId(); + if (topKRuntimeFilterSourceId != null) { + boxValue.add(String.format("TOPN OPT: %s", topKRuntimeFilterSourceId)); + } } boxValue.add( @@ -1052,7 +1053,7 @@ public class PlanGraphPrinter implements PlanVisitor<List<String>, PlanGraphPrin boxValue.add(String.format("TopK-%s", node.getPlanNodeId().getId())); boxValue.add(String.format("OrderingScheme: %s", node.getOrderingScheme())); boxValue.add(String.format("Count: %s", node.getCount())); - if (node.isUseTopKRuntimeFilter()) { + if (node.getTopKRuntimeFilterSourceId() != null) { boxValue.add("TOPN OPT"); } return render(node, boxValue, context); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java index 893420125d7..a15033af67b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanner.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.OutputNode; import org.apache.iotdb.commons.queryengine.plan.relational.type.InternalTypeManager; import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.execution.exchange.sink.DownStreamChannelLocation; @@ -180,8 +181,9 @@ public class TableDistributedPlanner { PlanNode planWithExchange = new AddExchangeNodes(mppQueryContext).addExchangeNodes(distributedPlan, planContext); - // Mark TopK runtime filter producer/consumer after exchange insertion. - if (analysis.isQuery()) { + // Mark TopK runtime filter after exchange insertion. + if (analysis.isQuery() + && IoTDBDescriptor.getInstance().getConfig().isEnableTopKRuntimeFilter()) { planWithExchange = new TopKRuntimeFilterOptimizer() .optimize( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java index bc7773a50c8..3b7b365c329 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java @@ -80,7 +80,7 @@ public class DeviceTableScanNode extends TableScanNode { protected transient boolean containsNonAlignedDevice; // Id of the TopKNode that produces the runtime filter for this scan; set during optimize. - @Nullable protected PlanNodeId topKRuntimeFilterSourceId; + @Nullable protected String topKRuntimeFilterSourceId; protected DeviceTableScanNode() {} @@ -177,12 +177,7 @@ public class DeviceTableScanNode extends TableScanNode { ReadWriteIOUtils.write(node.pushLimitToEachDevice, byteBuffer); - if (node.topKRuntimeFilterSourceId != null) { - ReadWriteIOUtils.write(true, byteBuffer); - node.topKRuntimeFilterSourceId.serialize(byteBuffer); - } else { - ReadWriteIOUtils.write(false, byteBuffer); - } + ReadWriteIOUtils.write(node.topKRuntimeFilterSourceId, byteBuffer); } protected static void serializeMemberVariables( @@ -212,12 +207,7 @@ public class DeviceTableScanNode extends TableScanNode { ReadWriteIOUtils.write(node.pushLimitToEachDevice, stream); - if (node.topKRuntimeFilterSourceId != null) { - ReadWriteIOUtils.write(true, stream); - node.topKRuntimeFilterSourceId.serialize(stream); - } else { - ReadWriteIOUtils.write(false, stream); - } + ReadWriteIOUtils.write(node.topKRuntimeFilterSourceId, stream); } protected static void deserializeMemberVariables( @@ -248,9 +238,7 @@ public class DeviceTableScanNode extends TableScanNode { node.pushLimitToEachDevice = ReadWriteIOUtils.readBool(byteBuffer); - if (ReadWriteIOUtils.readBool(byteBuffer)) { - node.topKRuntimeFilterSourceId = PlanNodeId.deserialize(byteBuffer); - } + node.topKRuntimeFilterSourceId = ReadWriteIOUtils.readString(byteBuffer); } @Override @@ -331,11 +319,12 @@ public class DeviceTableScanNode extends TableScanNode { this.containsNonAlignedDevice = true; } - public Optional<PlanNodeId> getTopKRuntimeFilterSourceId() { - return Optional.ofNullable(topKRuntimeFilterSourceId); + @Nullable + public String getTopKRuntimeFilterSourceId() { + return topKRuntimeFilterSourceId; } - public void setTopKRuntimeFilterSourceId(PlanNodeId topKRuntimeFilterSourceId) { + public void setTopKRuntimeFilterSourceId(@Nullable String topKRuntimeFilterSourceId) { this.topKRuntimeFilterSourceId = topKRuntimeFilterSourceId; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizer.java index f56000b9f89..25248cb45af 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizer.java @@ -23,13 +23,19 @@ import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; 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 TopK and table scan nodes for TopK runtime filter, similar to Doris {@code TOPN OPT} / - * {@code TOPN OPT: N} plan annotations. + * <p>Marks the {@code TopK + DeviceTableScan} structure for TopK runtime filter. + * + * <p>The topmost TopK establishes the <b>root TopK id</b>. 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 { @@ -41,55 +47,44 @@ public class TopKRuntimeFilterOptimizer implements PlanOptimizer { return plan.accept(new Rewriter(), null); } - private static class Rewriter implements PlanVisitor<PlanNode, Void> { + /** Context carries the root TopK id, or {@code null} until the first (topmost) TopK is seen. */ + private static class Rewriter implements PlanVisitor<PlanNode, PlanNodeId> { @Override - public PlanNode visitPlan(PlanNode node, Void context) { + public PlanNode visitPlan(PlanNode node, PlanNodeId rootTopKId) { PlanNode newNode = node.clone(); for (PlanNode child : node.getChildren()) { - newNode.addChild(child.accept(this, context)); + newNode.addChild(child.accept(this, rootTopKId)); } return newNode; } @Override - public PlanNode visitTopK(TopKNode node, Void context) { + public PlanNode visitTopK(TopKNode node, PlanNodeId rootTopKId) { TopKNode topKNode = (TopKNode) node.clone(); - for (PlanNode child : node.getChildren()) { - topKNode.addChild(child.accept(this, context)); - } - if (TopKRuntimeFilterUtils.isOrderByTimeOnly(topKNode.getOrderingScheme()) - && TopKRuntimeFilterUtils.qualifiesForRuntimeFilter(topKNode)) { - topKNode.setUseTopKRuntimeFilter(true); - topKNode.setTopKRuntimeFilterAscending( - topKNode - .getOrderingScheme() - .getOrdering(topKNode.getOrderingScheme().getOrderBy().get(0)) - .isAscending()); - markScansInSubtree(topKNode, topKNode.getPlanNodeId()); + // The topmost TopK establishes the root id; nested (per-region) TopK nodes inherit it. + PlanNodeId effectiveRootTopKId = rootTopKId == null ? node.getPlanNodeId() : rootTopKId; + + // 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. + boolean orderByTimeOnly = TopKRuntimeFilterUtils.isOrderByTimeOnly(node.getOrderingScheme()); + 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.getId()); + topKNode.addChild(scanNode); + } else { + topKNode.addChild(child.accept(this, effectiveRootTopKId)); + } } return topKNode; } - - private void markScansInSubtree(PlanNode root, PlanNodeId sourceId) { - root.accept( - new PlanVisitor<Void, Void>() { - @Override - public Void visitPlan(PlanNode node, Void unused) { - for (PlanNode child : node.getChildren()) { - child.accept(this, null); - } - return null; - } - - @Override - public Void visitDeviceTableScan(DeviceTableScanNode scanNode, Void unused) { - scanNode.setTopKRuntimeFilterSourceId(sourceId); - return null; - } - }, - null); - } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterUtils.java index a0704a72361..38168f822a6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterUtils.java @@ -19,15 +19,8 @@ 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.OrderingScheme; import org.apache.iotdb.commons.queryengine.plan.relational.planner.Symbol; -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.DeviceTableScanNode; -import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExchangeNode; - -import java.util.concurrent.atomic.AtomicBoolean; public final class TopKRuntimeFilterUtils { @@ -40,35 +33,4 @@ public final class TopKRuntimeFilterUtils { Symbol orderBy = orderingScheme.getOrderBy().get(0); return "time".equalsIgnoreCase(orderBy.getName()); } - - public static boolean qualifiesForRuntimeFilter(TopKNode topKNode) { - AtomicBoolean hasTableScan = new AtomicBoolean(false); - AtomicBoolean hasExchange = new AtomicBoolean(false); - PlanVisitor<Void, Void> subtreeDetector = - new PlanVisitor<Void, Void>() { - @Override - public Void visitPlan(PlanNode node, Void unused) { - for (PlanNode child : node.getChildren()) { - child.accept(this, null); - } - return null; - } - - @Override - public Void visitDeviceTableScan(DeviceTableScanNode scanNode, Void unused) { - hasTableScan.set(true); - return null; - } - - @Override - public Void visitTableExchange(ExchangeNode exchangeNode, Void unused) { - hasExchange.set(true); - return null; - } - }; - for (PlanNode child : topKNode.getChildren()) { - child.accept(subtreeDetector, null); - } - return hasTableScan.get() && !hasExchange.get(); - } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/QueryDataSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/QueryDataSource.java index 09da92f7136..a4184937e2d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/QueryDataSource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/QueryDataSource.java @@ -26,9 +26,9 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.read.filter.basic.Filter; +import org.apache.tsfile.utils.BitMap; import java.util.ArrayList; -import java.util.BitSet; import java.util.Comparator; import java.util.List; import java.util.TreeMap; @@ -74,25 +74,26 @@ public class QueryDataSource implements IQueryDataSource { private String databaseName = null; /** - * Physical seq/unseq TsFile indices pruned by TopK runtime filter at resource level. Shared - * across all devices in this scan. + * Physical seq/unseq TsFile indices pruned by TopK runtime filter at resource level. Initialized + * only when TopK runtime filter is enabled for this scan. */ - private final BitSet seqInvalidatedByRuntimeFilter = new BitSet(); + private BitMap seqInvalidatedByRuntimeFilter; - private final BitSet unseqInvalidatedByRuntimeFilter = new BitSet(); + private BitMap unseqInvalidatedByRuntimeFilter; /** - * Remaining seq + unseq TsFileResources that may still contain RF-qualifying rows. Decremented - * once per newly marked file; when zero, the scan can exit early. + * Remaining seq + unseq TsFileResources that may still contain RF-qualifying rows. Initialized + * together with the bitmaps above. */ private int validSize; + private boolean runtimeFilterTrackingEnabled; + private static final Comparator<Long> descendingComparator = (o1, o2) -> Long.compare(o2, o1); public QueryDataSource(List<TsFileResource> seqResources, List<TsFileResource> unseqResources) { this.seqResources = seqResources; this.unseqResources = unseqResources; - initValidSize(); } public QueryDataSource( @@ -100,7 +101,6 @@ public class QueryDataSource implements IQueryDataSource { this.seqResources = seqResources; this.unseqResources = unseqResources; this.databaseName = databaseName; - initValidSize(); } // used for compaction, because in compaction task(unlike query, each QueryDataSource only serve @@ -110,15 +110,29 @@ public class QueryDataSource implements IQueryDataSource { this.unseqResources = other.unseqResources; this.unSeqFileOrderIndex = other.unSeqFileOrderIndex; this.databaseName = other.databaseName; - this.validSize = other.validSize; - this.seqInvalidatedByRuntimeFilter.or(other.seqInvalidatedByRuntimeFilter); - this.unseqInvalidatedByRuntimeFilter.or(other.unseqInvalidatedByRuntimeFilter); + if (other.runtimeFilterTrackingEnabled) { + this.runtimeFilterTrackingEnabled = true; + this.validSize = other.validSize; + this.seqInvalidatedByRuntimeFilter = other.seqInvalidatedByRuntimeFilter.clone(); + this.unseqInvalidatedByRuntimeFilter = other.unseqInvalidatedByRuntimeFilter.clone(); + } } - private void initValidSize() { + /** Initializes RF tracking state when TopK runtime filter is present for this scan. */ + public void initRuntimeFilterTracking() { + if (runtimeFilterTrackingEnabled) { + return; + } + runtimeFilterTrackingEnabled = true; + seqInvalidatedByRuntimeFilter = new BitMap(getSeqResourcesSize()); + unseqInvalidatedByRuntimeFilter = new BitMap(getUnseqResourcesSize()); validSize = getSeqResourcesSize() + getUnseqResourcesSize(); } + public boolean isRuntimeFilterTrackingEnabled() { + return runtimeFilterTrackingEnabled; + } + @TestOnly public int getValidSize() { return validSize; @@ -126,34 +140,49 @@ public class QueryDataSource implements IQueryDataSource { /** Returns true if any seq/unseq file may still contain RF-qualifying resources. */ public boolean hasValidResource() { + if (!runtimeFilterTrackingEnabled) { + return true; + } return validSize > 0; } - /** Marks the seq file at physical {@code index} as pruned by TopK RF at resource level. */ + /** + * Marks the seq file at physical {@code index} as pruned by TopK RF at resource level. + * + * <p>Caller must skip indices for which {@link #isRuntimeFilterPruned(boolean, int)} is already + * true; this method does not deduplicate marks. + */ public void setSeqTsFileResourceInvalidated(int physicalIndex) { - if (!seqInvalidatedByRuntimeFilter.get(physicalIndex)) { - seqInvalidatedByRuntimeFilter.set(physicalIndex); - validSize--; + if (!runtimeFilterTrackingEnabled) { + return; } + seqInvalidatedByRuntimeFilter.mark(physicalIndex); + validSize--; } /** * Marks the unseq file at traversal {@code orderIndex} as pruned by TopK RF at resource level. + * + * <p>Caller must skip indices for which {@link #isRuntimeFilterPruned(boolean, int)} is already + * true; this method does not deduplicate marks. */ public void setUnseqTsFileResourceInvalidated(int orderIndex) { - int physicalIndex = unSeqFileOrderIndex[orderIndex]; - if (!unseqInvalidatedByRuntimeFilter.get(physicalIndex)) { - unseqInvalidatedByRuntimeFilter.set(physicalIndex); - validSize--; + if (!runtimeFilterTrackingEnabled) { + return; } + unseqInvalidatedByRuntimeFilter.mark(unSeqFileOrderIndex[orderIndex]); + validSize--; } /** Returns true if this file was already pruned by RF at resource level on a prior device. */ public boolean isRuntimeFilterPruned(boolean isSeq, int index) { + if (!runtimeFilterTrackingEnabled) { + return false; + } if (isSeq) { - return seqInvalidatedByRuntimeFilter.get(index); + return seqInvalidatedByRuntimeFilter.isMarked(index); } - return unseqInvalidatedByRuntimeFilter.get(unSeqFileOrderIndex[index]); + return unseqInvalidatedByRuntimeFilter.isMarked(unSeqFileOrderIndex[index]); } public List<TsFileResource> getSeqResources() { @@ -208,7 +237,7 @@ public class QueryDataSource implements IQueryDataSource { } public boolean isSeqSatisfiedByRuntimeFilter( - IDeviceID deviceID, int curIndex, TopKRuntimeFilter filter, boolean debug) { + int curIndex, TopKRuntimeFilter filter, boolean debug) { return isResourceSatisfiedByRuntimeFilter(curIndex, filter, true, debug); } @@ -264,9 +293,8 @@ public class QueryDataSource implements IQueryDataSource { return curUnSeqSatisfied; } - public boolean isUnSeqSatisfiedByRuntimeFilter( - int curIndex, TopKRuntimeFilter filter, boolean debug) { - return isResourceSatisfiedByRuntimeFilter(curIndex, filter, false, debug); + public boolean isUnSeqSatisfiedByRuntimeFilter(int curIndex, TopKRuntimeFilter filter) { + return isResourceSatisfiedByRuntimeFilter(curIndex, filter, false, false); } public long getCurrentUnSeqOrderTime(int curIndex) { @@ -345,9 +373,6 @@ public class QueryDataSource implements IQueryDataSource { } TsFileResource tsFileResource = isSeq ? seqResources.get(curIndex) : unseqResources.get(unSeqFileOrderIndex[curIndex]); - if (tsFileResource == null) { - return false; - } // Resource-level RF uses the TsFile's global time range, not per-device bounds. long startTime = tsFileResource.getFileStartTime(); long endTime = tsFileResource.isClosed() ? tsFileResource.getFileEndTime() : Long.MAX_VALUE; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizerTest.java index c67506f3c0b..1a0f5d5b743 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizerTest.java @@ -19,55 +19,129 @@ 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.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.queryengine.plan.relational.planner.OrderingScheme; import org.apache.iotdb.commons.queryengine.plan.relational.planner.SortOrder; import org.apache.iotdb.commons.queryengine.plan.relational.planner.Symbol; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.LimitNode; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.TopKNode; +import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExchangeNode; +import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.PlanOptimizer.Context; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; import org.junit.Assert; import org.junit.Test; +import org.mockito.Mockito; import java.util.Collections; import java.util.Optional; public class TopKRuntimeFilterOptimizerTest { + /** Single region: the sole TopK sits directly over the scan and uses its own id as root. */ @Test - public void marksTopKAndScanWhenQualified() { + public void marksTopKAndScanWithOwnRootIdInSingleRegion() { PlanNodeId topKId = new PlanNodeId("topk"); - PlanNodeId scanId = new PlanNodeId("scan"); TopKNode topKNode = createTopK(topKId, SortOrder.ASC_NULLS_LAST); - DeviceTableScanNode scanNode = createScan(scanId); - topKNode.addChild(new LimitNode(new PlanNodeId("limit"), scanNode, 10, Optional.empty())); + topKNode.addChild(createScan(new PlanNodeId("scan"))); - TopKNode optimized = (TopKNode) new TopKRuntimeFilterOptimizer().optimize(topKNode, null); + TopKNode optimized = + (TopKNode) new TopKRuntimeFilterOptimizer().optimize(topKNode, queryContext()); - Assert.assertTrue(optimized.isUseTopKRuntimeFilter()); + Assert.assertNotNull(optimized.getTopKRuntimeFilterSourceId()); Assert.assertTrue(optimized.isTopKRuntimeFilterAscending()); - DeviceTableScanNode optimizedScan = - (DeviceTableScanNode) optimized.getChildren().get(0).getChildren().get(0); - Assert.assertEquals(topKId, optimizedScan.getTopKRuntimeFilterSourceId().orElse(null)); + Assert.assertEquals(topKId, optimized.getTopKRuntimeFilterSourceId()); + DeviceTableScanNode optimizedScan = (DeviceTableScanNode) optimized.getChildren().get(0); + Assert.assertEquals(topKId.getId(), optimizedScan.getTopKRuntimeFilterSourceId()); } @Test - public void skipsTopKWithExchangeChild() { + public void marksDescendingWhenOrderByTimeDesc() { PlanNodeId topKId = new PlanNodeId("topk"); TopKNode topKNode = createTopK(topKId, SortOrder.DESC_NULLS_LAST); - topKNode.addChild(new ExchangeNode(new PlanNodeId("exchange"), null, null, null, null)); + topKNode.addChild(createScan(new PlanNodeId("scan"))); + + TopKNode optimized = + (TopKNode) new TopKRuntimeFilterOptimizer().optimize(topKNode, queryContext()); + + Assert.assertNotNull(optimized.getTopKRuntimeFilterSourceId()); + Assert.assertFalse(optimized.isTopKRuntimeFilterAscending()); + } + + /** + * Multi region: the coordinator TopK establishes the root id but is not a producer; each region + * TopK directly above a scan becomes the producer and both it and its scan carry the root id. + */ + @Test + public void regionTopKAndScanShareRootTopKId() { + PlanNodeId rootTopKId = new PlanNodeId("root-topk"); + PlanNodeId regionTopKId = new PlanNodeId("region-topk"); + TopKNode rootTopK = createTopK(rootTopKId, SortOrder.DESC_NULLS_LAST); + ExchangeNode exchange = new ExchangeNode(new PlanNodeId("exchange")); + TopKNode regionTopK = createTopK(regionTopKId, SortOrder.DESC_NULLS_LAST); + regionTopK.addChild(createScan(new PlanNodeId("scan"))); + exchange.addChild(regionTopK); + rootTopK.addChild(exchange); + + TopKNode optimizedRoot = + (TopKNode) new TopKRuntimeFilterOptimizer().optimize(rootTopK, queryContext()); - TopKNode optimized = (TopKNode) new TopKRuntimeFilterOptimizer().optimize(topKNode, null); + // Coordinator TopK is not a producer (its child is an Exchange, not a scan). + Assert.assertNull(optimizedRoot.getTopKRuntimeFilterSourceId()); + + TopKNode optimizedRegion = (TopKNode) optimizedRoot.getChildren().get(0).getChildren().get(0); + Assert.assertNotNull(optimizedRegion.getTopKRuntimeFilterSourceId()); + Assert.assertEquals(rootTopKId, optimizedRegion.getTopKRuntimeFilterSourceId()); + + DeviceTableScanNode optimizedScan = (DeviceTableScanNode) optimizedRegion.getChildren().get(0); + Assert.assertEquals(rootTopKId.getId(), optimizedScan.getTopKRuntimeFilterSourceId()); + } + + /** Scan not a direct child of the TopK (a Limit in between): the structure is not marked. */ + @Test + public void skipsWhenScanNotDirectChild() { + PlanNodeId topKId = new PlanNodeId("topk"); + TopKNode topKNode = createTopK(topKId, SortOrder.ASC_NULLS_LAST); + DeviceTableScanNode scanNode = createScan(new PlanNodeId("scan")); + topKNode.addChild(new LimitNode(new PlanNodeId("limit"), scanNode, 10, Optional.empty())); + + TopKNode optimized = + (TopKNode) new TopKRuntimeFilterOptimizer().optimize(topKNode, queryContext()); + + Assert.assertNull(optimized.getTopKRuntimeFilterSourceId()); + PlanNode limit = optimized.getChildren().get(0); + DeviceTableScanNode optimizedScan = (DeviceTableScanNode) limit.getChildren().get(0); + Assert.assertNull(optimizedScan.getTopKRuntimeFilterSourceId()); + } + + @Test + public void skipsTopKWithoutScan() { + PlanNodeId topKId = new PlanNodeId("topk"); + TopKNode topKNode = createTopK(topKId, SortOrder.ASC_NULLS_LAST); + topKNode.addChild(new ExchangeNode(new PlanNodeId("exchange"))); + + TopKNode optimized = + (TopKNode) new TopKRuntimeFilterOptimizer().optimize(topKNode, queryContext()); + + Assert.assertNull(optimized.getTopKRuntimeFilterSourceId()); + } - Assert.assertFalse(optimized.isUseTopKRuntimeFilter()); + private static Context queryContext() { + Analysis analysis = Mockito.mock(Analysis.class); + Mockito.when(analysis.isQuery()).thenReturn(true); + Context context = Mockito.mock(Context.class); + Mockito.when(context.getAnalysis()).thenReturn(analysis); + return context; } private TopKNode createTopK(PlanNodeId id, SortOrder sortOrder) { Symbol timeSymbol = new Symbol("time"); OrderingScheme orderingScheme = - new OrderingScheme(Collections.singletonList(timeSymbol), sortOrder); + new OrderingScheme(ImmutableList.of(timeSymbol), ImmutableMap.of(timeSymbol, sortOrder)); return new TopKNode(id, orderingScheme, 10, Collections.singletonList(timeSymbol), false); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterUtilsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterUtilsTest.java new file mode 100644 index 00000000000..b3d03822e93 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterUtilsTest.java @@ -0,0 +1,60 @@ +/* + * 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.relational.planner.OrderingScheme; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.SortOrder; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.Symbol; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import org.junit.Assert; +import org.junit.Test; + +public class TopKRuntimeFilterUtilsTest { + + @Test + public void isOrderByTimeOnlyAcceptsSingleTimeColumn() { + Symbol time = new Symbol("time"); + OrderingScheme scheme = + new OrderingScheme( + ImmutableList.of(time), ImmutableMap.of(time, SortOrder.DESC_NULLS_LAST)); + Assert.assertTrue(TopKRuntimeFilterUtils.isOrderByTimeOnly(scheme)); + } + + @Test + public void isOrderByTimeOnlyRejectsNonTimeColumn() { + Symbol tag = new Symbol("tag1"); + OrderingScheme scheme = + new OrderingScheme(ImmutableList.of(tag), ImmutableMap.of(tag, SortOrder.ASC_NULLS_LAST)); + Assert.assertFalse(TopKRuntimeFilterUtils.isOrderByTimeOnly(scheme)); + } + + @Test + public void isOrderByTimeOnlyRejectsMultipleColumns() { + Symbol time = new Symbol("time"); + Symbol tag = new Symbol("tag1"); + OrderingScheme scheme = + new OrderingScheme( + ImmutableList.of(time, tag), + ImmutableMap.of(time, SortOrder.ASC_NULLS_LAST, tag, SortOrder.ASC_NULLS_LAST)); + Assert.assertFalse(TopKRuntimeFilterUtils.isOrderByTimeOnly(scheme)); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/QueryDataSourceRuntimeFilterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/QueryDataSourceRuntimeFilterTest.java new file mode 100644 index 00000000000..e6f1b8787f9 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/read/QueryDataSourceRuntimeFilterTest.java @@ -0,0 +1,76 @@ +/* + * 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.storageengine.dataregion.read; + +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; + +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.Arrays; +import java.util.Collections; + +public class QueryDataSourceRuntimeFilterTest { + + @Test + public void testSeqInvalidatedIndicesAreIndependent() { + TsFileResource f0 = mockResource(10, 10); + TsFileResource f1 = mockResource(5, 5); + TsFileResource f2 = mockResource(20, 20); + QueryDataSource dataSource = + new QueryDataSource(Arrays.asList(f0, f1, f2), Collections.emptyList()); + dataSource.initRuntimeFilterTracking(); + + Assert.assertEquals(3, dataSource.getValidSize()); + + dataSource.setSeqTsFileResourceInvalidated(0); + dataSource.setSeqTsFileResourceInvalidated(2); + + Assert.assertTrue(dataSource.isRuntimeFilterPruned(true, 0)); + Assert.assertFalse(dataSource.isRuntimeFilterPruned(true, 1)); + Assert.assertTrue(dataSource.isRuntimeFilterPruned(true, 2)); + Assert.assertEquals(1, dataSource.getValidSize()); + Assert.assertTrue(dataSource.hasValidResource()); + } + + @Test + public void testAllSeqInvalidatedMeansNoValidResource() { + TsFileResource f0 = mockResource(1, 1); + TsFileResource f1 = mockResource(2, 2); + QueryDataSource dataSource = + new QueryDataSource(Arrays.asList(f0, f1), Collections.emptyList()); + dataSource.initRuntimeFilterTracking(); + + dataSource.setSeqTsFileResourceInvalidated(0); + dataSource.setSeqTsFileResourceInvalidated(1); + + Assert.assertEquals(0, dataSource.getValidSize()); + Assert.assertFalse(dataSource.hasValidResource()); + } + + private static TsFileResource mockResource(long startTime, long endTime) { + TsFileResource resource = Mockito.mock(TsFileResource.class); + Mockito.when(resource.getFileStartTime()).thenReturn(startTime); + Mockito.when(resource.isClosed()).thenReturn(true); + Mockito.when(resource.getFileEndTime()).thenReturn(endTime); + return resource; + } +} diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index c54878bfe1b..fa29e94c9e4 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template @@ -1438,6 +1438,14 @@ recovery_log_interval_in_ms=5000 # Datatype: boolean enable_tsfile_validation=false +# Whether to enable the TopK runtime filter optimization for table-model +# "ORDER BY time LIMIT k" queries. When enabled, the TopK operator feeds the current +# heap-top time back to table scans so scans can skip files/rows that cannot enter the +# final result. Set to false to fall back to the original execution path without restart. +# effectiveMode: hot_reload +# Datatype: boolean +enable_topk_runtime_filter=true + # Default tier TTL. When the survival time of the data exceeds the threshold, it will be migrated to the next tier. # Negative value means the tier TTL is unlimited. # effectiveMode: restart diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/planner/node/TopKNode.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/planner/node/TopKNode.java index 70c866d5e0d..c5195348a51 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/planner/node/TopKNode.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/relational/planner/node/TopKNode.java @@ -32,6 +32,8 @@ import org.apache.iotdb.commons.queryengine.plan.relational.planner.Symbol; import com.google.common.base.Objects; import org.apache.tsfile.utils.ReadWriteIOUtils; +import javax.annotation.Nullable; + import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; @@ -50,12 +52,13 @@ public class TopKNode extends MultiChildProcessNode { private final boolean childrenDataInOrder; - // Marked during distributed optimize when TopK runtime filter should be generated. - private boolean useTopKRuntimeFilter; + private final transient boolean topKRuntimeFilterAscending; - // Sort direction of the runtime filter threshold; meaningful only when useTopKRuntimeFilter is - // true. - private boolean topKRuntimeFilterAscending = true; + // Root TopK plan node id under which the shared runtime filter is registered/looked up; a + // non-null value also marks this TopK as a runtime filter producer (set during distributed + // optimize). All per-region TopK producers and their scan consumers within the same query use + // this id so that fragments of the same query on one DataNode share a single filter instance. + @Nullable private PlanNodeId topKRuntimeFilterSourceId; public TopKNode( PlanNodeId id, @@ -68,6 +71,7 @@ public class TopKNode extends MultiChildProcessNode { this.count = count; this.outputSymbols = outputSymbols; this.childrenDataInOrder = childrenDataInOrder; + this.topKRuntimeFilterAscending = computeTopKRuntimeFilterAscending(scheme); } public TopKNode( @@ -82,14 +86,19 @@ public class TopKNode extends MultiChildProcessNode { this.count = count; this.outputSymbols = outputSymbols; this.childrenDataInOrder = childrenDataInOrder; + this.topKRuntimeFilterAscending = computeTopKRuntimeFilterAscending(scheme); + } + + private static boolean computeTopKRuntimeFilterAscending(OrderingScheme orderingScheme) { + Symbol orderBy = orderingScheme.getOrderBy().get(0); + return orderingScheme.getOrdering(orderBy).isAscending(); } @Override public PlanNode clone() { TopKNode cloned = new TopKNode(getPlanNodeId(), orderingScheme, count, outputSymbols, childrenDataInOrder); - cloned.useTopKRuntimeFilter = useTopKRuntimeFilter; - cloned.topKRuntimeFilterAscending = topKRuntimeFilterAscending; + cloned.topKRuntimeFilterSourceId = topKRuntimeFilterSourceId; return cloned; } @@ -113,8 +122,8 @@ public class TopKNode extends MultiChildProcessNode { Symbol.serialize(symbol, byteBuffer); } ReadWriteIOUtils.write(childrenDataInOrder, byteBuffer); - ReadWriteIOUtils.write(useTopKRuntimeFilter, byteBuffer); - ReadWriteIOUtils.write(topKRuntimeFilterAscending, byteBuffer); + ReadWriteIOUtils.write( + topKRuntimeFilterSourceId == null ? null : topKRuntimeFilterSourceId.getId(), byteBuffer); } @Override @@ -127,8 +136,8 @@ public class TopKNode extends MultiChildProcessNode { Symbol.serialize(symbol, stream); } ReadWriteIOUtils.write(childrenDataInOrder, stream); - ReadWriteIOUtils.write(useTopKRuntimeFilter, stream); - ReadWriteIOUtils.write(topKRuntimeFilterAscending, stream); + ReadWriteIOUtils.write( + topKRuntimeFilterSourceId == null ? null : topKRuntimeFilterSourceId.getId(), stream); } public static TopKNode deserialize(ByteBuffer byteBuffer) { @@ -140,13 +149,12 @@ public class TopKNode extends MultiChildProcessNode { outputSymbols.add(Symbol.deserialize(byteBuffer)); } boolean childrenDataInOrder = ReadWriteIOUtils.readBool(byteBuffer); - boolean useTopKRuntimeFilter = ReadWriteIOUtils.readBool(byteBuffer); - boolean topKRuntimeFilterAscending = ReadWriteIOUtils.readBool(byteBuffer); + String topKRuntimeFilterSourceId = ReadWriteIOUtils.readString(byteBuffer); PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer); TopKNode topKNode = new TopKNode(planNodeId, orderingScheme, count, outputSymbols, childrenDataInOrder); - topKNode.useTopKRuntimeFilter = useTopKRuntimeFilter; - topKNode.topKRuntimeFilterAscending = topKRuntimeFilterAscending; + topKNode.topKRuntimeFilterSourceId = + topKRuntimeFilterSourceId == null ? null : new PlanNodeId(topKRuntimeFilterSourceId); return topKNode; } @@ -162,8 +170,7 @@ public class TopKNode extends MultiChildProcessNode { QueryMessages.EXCEPTION_WRONG_NUMBER_OF_NEW_CHILDREN_817AF800); TopKNode topKNode = new TopKNode(id, newChildren, orderingScheme, count, outputSymbols, childrenDataInOrder); - topKNode.useTopKRuntimeFilter = useTopKRuntimeFilter; - topKNode.topKRuntimeFilterAscending = topKRuntimeFilterAscending; + topKNode.topKRuntimeFilterSourceId = topKRuntimeFilterSourceId; return topKNode; } @@ -179,20 +186,18 @@ public class TopKNode extends MultiChildProcessNode { return childrenDataInOrder; } - public boolean isUseTopKRuntimeFilter() { - return useTopKRuntimeFilter; - } - - public void setUseTopKRuntimeFilter(boolean useTopKRuntimeFilter) { - this.useTopKRuntimeFilter = useTopKRuntimeFilter; - } - public boolean isTopKRuntimeFilterAscending() { return topKRuntimeFilterAscending; } - public void setTopKRuntimeFilterAscending(boolean topKRuntimeFilterAscending) { - this.topKRuntimeFilterAscending = topKRuntimeFilterAscending; + /** A non-null source id marks this TopK as a runtime filter producer. */ + @Nullable + public PlanNodeId getTopKRuntimeFilterSourceId() { + return topKRuntimeFilterSourceId; + } + + public void setTopKRuntimeFilterSourceId(@Nullable PlanNodeId topKRuntimeFilterSourceId) { + this.topKRuntimeFilterSourceId = topKRuntimeFilterSourceId; } @Override @@ -204,19 +209,13 @@ public class TopKNode extends MultiChildProcessNode { return Objects.equal(orderingScheme, sortNode.orderingScheme) && Objects.equal(outputSymbols, sortNode.outputSymbols) && Objects.equal(count, sortNode.count) - && useTopKRuntimeFilter == sortNode.useTopKRuntimeFilter - && topKRuntimeFilterAscending == sortNode.topKRuntimeFilterAscending; + && Objects.equal(topKRuntimeFilterSourceId, sortNode.topKRuntimeFilterSourceId); } @Override public int hashCode() { return Objects.hashCode( - super.hashCode(), - orderingScheme, - outputSymbols, - count, - useTopKRuntimeFilter, - topKRuntimeFilterAscending); + super.hashCode(), orderingScheme, outputSymbols, count, topKRuntimeFilterSourceId); } @Override
