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 0bb8159720a604bc947343c46647a47c17dc4f6b Author: Weihao Li <[email protected]> AuthorDate: Fri Jul 10 10:51:46 2026 +0800 draft Signed-off-by: Weihao Li <[email protected]> --- .../calc/execution/filter/TopKRuntimeFilter.java | 67 ++++++++ .../operator/process/TableTopKOperator.java | 28 ++- .../execution/operator/process/TopKOperator.java | 33 ++++ .../calc/plan/planner/TableOperatorGenerator.java | 12 +- .../execution/filter/TopKRuntimeFilterTest.java | 73 ++++++++ .../execution/fragment/DataNodeQueryContext.java | 21 +++ .../execution/operator/source/SeriesScanUtil.java | 129 +++++++++++++- .../relational/AbstractTableScanOperator.java | 43 +++-- .../planner/DataNodeTableOperatorGenerator.java | 189 ++++++++++++++++++++- .../plan/planner/LocalExecutionPlanner.java | 2 + .../plan/planner/TopKRuntimeFilterBinder.java | 60 +++++++ .../plan/planner/plan/node/PlanGraphPrinter.java | 6 + .../planner/plan/parameter/SeriesScanOptions.java | 19 ++- .../distribute/TableDistributedPlanner.java | 23 ++- .../planner/node/DeviceTableScanNode.java | 60 +++++-- .../optimizations/TopKRuntimeFilterOptimizer.java | 95 +++++++++++ .../optimizations/TopKRuntimeFilterUtils.java | 74 ++++++++ .../dataregion/read/QueryDataSource.java | 115 +++++++++++++ .../TopKRuntimeFilterOptimizerTest.java | 78 +++++++++ .../plan/relational/planner/node/TopKNode.java | 59 ++++++- 20 files changed, 1143 insertions(+), 43 deletions(-) diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/filter/TopKRuntimeFilter.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/filter/TopKRuntimeFilter.java new file mode 100644 index 00000000000..52d3ccbcaf6 --- /dev/null +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/filter/TopKRuntimeFilter.java @@ -0,0 +1,67 @@ +/* + * 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.calc.execution.filter; + +import java.util.concurrent.atomic.AtomicLong; + +/** + * A dynamic time threshold shared between TopK and Scan operators. + * + * <p>When TopK heap is full, the worst row in the heap becomes the threshold. Scan operators can + * skip rows (and even files) that cannot possibly enter the final TopK result. + */ +public class TopKRuntimeFilter { + private final boolean ascending; + private final AtomicLong threshold; + + public TopKRuntimeFilter(boolean ascending) { + this.ascending = ascending; + threshold = new AtomicLong(ascending ? Long.MAX_VALUE : Long.MIN_VALUE); + } + + public boolean isAscending() { + return ascending; + } + + /** Update threshold with the current heap-top time. Only tightens the bound. */ + public void updateThreshold(long time) { + if (ascending) { + // ASC TopK: keep smallest K rows, threshold is the largest time among them + threshold.updateAndGet(prev -> Math.min(prev, time)); + } else { + // DESC TopK: keep largest K rows, threshold is the smallest time among them + threshold.updateAndGet(prev -> Math.max(prev, time)); + } + } + + public boolean mayQualify(long time) { + long current = threshold.get(); + return ascending ? time < current : time > current; + } + + public boolean mayQualifyRange(long startTime, long endTime) { + long current = threshold.get(); + return ascending ? startTime < current : endTime > current; + } + + public long getThreshold() { + return threshold.get(); + } +} diff --git a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/TableTopKOperator.java b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/TableTopKOperator.java index 2a7d11657b7..aaf270369fe 100644 --- a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/TableTopKOperator.java +++ b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/TableTopKOperator.java @@ -19,6 +19,7 @@ package org.apache.iotdb.calc.execution.operator.process; +import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; import org.apache.iotdb.calc.execution.operator.CommonOperatorContext; import org.apache.iotdb.calc.execution.operator.Operator; import org.apache.iotdb.calc.plan.planner.CommonOperatorUtils; @@ -33,6 +34,24 @@ import java.util.Comparator; import java.util.List; public class TableTopKOperator extends TopKOperator { + public TableTopKOperator( + CommonOperatorContext operatorContext, + List<Operator> childrenOperators, + List<TSDataType> dataTypes, + Comparator<SortKey> comparator, + int topValue, + boolean childrenDataInOrder, + TopKRuntimeFilter topKRuntimeFilter) { + super( + operatorContext, + childrenOperators, + dataTypes, + comparator, + topValue, + childrenDataInOrder, + topKRuntimeFilter); + } + public TableTopKOperator( CommonOperatorContext operatorContext, List<Operator> childrenOperators, @@ -40,7 +59,14 @@ public class TableTopKOperator extends TopKOperator { Comparator<SortKey> comparator, int topValue, boolean childrenDataInOrder) { - super(operatorContext, childrenOperators, dataTypes, comparator, topValue, childrenDataInOrder); + this( + operatorContext, + childrenOperators, + dataTypes, + comparator, + topValue, + childrenDataInOrder, + null); } @Override 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 53ceeee58a0..54d58e85bdd 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 @@ -19,6 +19,7 @@ package org.apache.iotdb.calc.execution.operator.process; +import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; import org.apache.iotdb.calc.execution.operator.CommonOperatorContext; import org.apache.iotdb.calc.execution.operator.Operator; import org.apache.iotdb.calc.i18n.CalcMessages; @@ -85,6 +86,8 @@ public abstract class TopKOperator implements ProcessOperator { // the data of every childOperator is in order private final boolean childrenDataInOrder; + private final TopKRuntimeFilter topKRuntimeFilter; + public static final int OPERATOR_BATCH_UPPER_BOUND = 100000; protected TopKOperator( @@ -94,6 +97,24 @@ public abstract class TopKOperator implements ProcessOperator { Comparator<SortKey> comparator, int topValue, boolean childrenDataInOrder) { + this( + operatorContext, + childrenOperators, + dataTypes, + comparator, + topValue, + childrenDataInOrder, + null); + } + + protected TopKOperator( + CommonOperatorContext operatorContext, + List<Operator> childrenOperators, + List<TSDataType> dataTypes, + Comparator<SortKey> comparator, + int topValue, + boolean childrenDataInOrder, + TopKRuntimeFilter topKRuntimeFilter) { this.operatorContext = operatorContext; this.childrenOperators = childrenOperators; this.dataTypes = dataTypes; @@ -102,6 +123,7 @@ public abstract class TopKOperator implements ProcessOperator { this.tsBlockBuilder = new TsBlockBuilder(topValue, dataTypes); this.topValue = topValue; this.childrenDataInOrder = childrenDataInOrder; + this.topKRuntimeFilter = topKRuntimeFilter; initResultTsBlock(); @@ -209,6 +231,7 @@ public abstract class TopKOperator implements ProcessOperator { if (skipCurrentBatch) { closeOperator(i); } + updateTopKRuntimeFilter(); canCallNext[i] = false; if (System.nanoTime() - startTime > maxRuntime) { @@ -425,4 +448,14 @@ 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 5a003e18d3c..e44379c9c5d 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 @@ -19,6 +19,7 @@ package org.apache.iotdb.calc.plan.planner; +import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; import org.apache.iotdb.calc.execution.operator.CommonOperatorContext; import org.apache.iotdb.calc.execution.operator.Operator; import org.apache.iotdb.calc.execution.operator.process.AssignUniqueIdOperator; @@ -910,7 +911,16 @@ public abstract class TableOperatorGenerator< getComparatorForTable( node.getOrderingScheme().getOrderingList(), sortItemIndexList, sortItemDataTypeList), (int) node.getCount(), - node.isChildrenDataInOrder()); + node.isChildrenDataInOrder(), + getTopKRuntimeFilter(context, node)); + } + + protected TopKRuntimeFilter getTopKRuntimeFilter(C context, TopKNode node) { + return null; + } + + protected TopKRuntimeFilter getTopKRuntimeFilter(C context) { + return getTopKRuntimeFilter(context, null); } protected List<TSDataType> getOutputColumnTypes(PlanNode node, ITableTypeProvider typeProvider) { diff --git a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/filter/TopKRuntimeFilterTest.java b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/filter/TopKRuntimeFilterTest.java new file mode 100644 index 00000000000..b84db8bb4a6 --- /dev/null +++ b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/filter/TopKRuntimeFilterTest.java @@ -0,0 +1,73 @@ +/* + * 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.calc.execution.filter; + +import org.junit.Assert; +import org.junit.Test; + +public class TopKRuntimeFilterTest { + + @Test + public void testDescRuntimeFilter() { + TopKRuntimeFilter filter = new TopKRuntimeFilter(false); + Assert.assertTrue(filter.mayQualify(100L)); + Assert.assertTrue(filter.mayQualifyRange(0L, Long.MAX_VALUE)); + + filter.updateThreshold(50L); + Assert.assertTrue(filter.mayQualify(60L)); + Assert.assertFalse(filter.mayQualify(50L)); + Assert.assertFalse(filter.mayQualify(40L)); + + filter.updateThreshold(55L); + Assert.assertFalse(filter.mayQualify(54L)); + Assert.assertFalse(filter.mayQualify(55L)); + Assert.assertTrue(filter.mayQualify(56L)); + } + + @Test + public void testAscRuntimeFilter() { + TopKRuntimeFilter filter = new TopKRuntimeFilter(true); + Assert.assertTrue(filter.mayQualify(100L)); + Assert.assertTrue(filter.mayQualifyRange(0L, Long.MAX_VALUE - 1)); + + filter.updateThreshold(100L); + Assert.assertTrue(filter.mayQualify(90L)); + Assert.assertFalse(filter.mayQualify(100L)); + Assert.assertFalse(filter.mayQualify(110L)); + + filter.updateThreshold(80L); + Assert.assertFalse(filter.mayQualify(90L)); + Assert.assertTrue(filter.mayQualify(70L)); + } + + @Test + public void testMayQualifyRange() { + TopKRuntimeFilter filter = new TopKRuntimeFilter(false); + filter.updateThreshold(50L); + Assert.assertTrue(filter.mayQualifyRange(60L, 100L)); + Assert.assertFalse(filter.mayQualifyRange(50L, 50L)); + Assert.assertFalse(filter.mayQualifyRange(10L, 40L)); + + TopKRuntimeFilter ascFilter = new TopKRuntimeFilter(true); + ascFilter.updateThreshold(100L); + Assert.assertTrue(ascFilter.mayQualifyRange(10L, 99L)); + Assert.assertFalse(ascFilter.mayQualifyRange(100L, 200L)); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java index ef6b60eec4d..3e3b8beee01 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.queryengine.execution.fragment; +import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; import org.apache.iotdb.commons.client.exception.ClientManagerException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName; @@ -75,6 +76,26 @@ public class DataNodeQueryContext { // unnecessary private final ReentrantLock lock = new ReentrantLock(); + /** + * TopK runtime filters shared across fragment instances on this DataNode for the same query. + * + * <p>Key: root TopK plan node id. Value: filter instance produced by the TopK operator and + * consumed by Scan operators referencing the same id via {@code topKRuntimeFilterSourceId}. + */ + private final Map<String, TopKRuntimeFilter> runtimeFilters = new ConcurrentHashMap<>(); + + /** + * Registers a TopK runtime filter for the given root TopK id. If multiple fragment instances bind + * the same id, they share one filter instance. + */ + public TopKRuntimeFilter registerTopKRuntimeFilter(String filterId, TopKRuntimeFilter filter) { + return runtimeFilters.computeIfAbsent(filterId, id -> filter); + } + + public TopKRuntimeFilter getTopKRuntimeFilter(String filterId) { + return runtimeFilters.get(filterId); + } + public DataNodeQueryContext(int dataNodeFINum) { this.uncachedPathToSeriesScanInfo = new ConcurrentHashMap<>(); this.dataNodeFINum = new AtomicInteger(dataNodeFINum); 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 c6a440b1884..ad87d994983 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 @@ -19,6 +19,7 @@ package org.apache.iotdb.db.queryengine.execution.operator.source; +import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; import org.apache.iotdb.commons.path.IFullPath; import org.apache.iotdb.commons.path.NonAlignedFullPath; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; @@ -139,6 +140,7 @@ public class SeriesScanUtil implements Accountable { protected SeriesScanOptions scanOptions; private final PaginationController paginationController; + private boolean runtimeFilterExhausted; private static final SeriesScanCostMetricSet SERIES_SCAN_COST_METRIC_SET = SeriesScanCostMetricSet.getInstance(); @@ -276,7 +278,7 @@ public class SeriesScanUtil implements Accountable { // Optional.empty(), it needs to return directly to the checkpoint method that checks the operator // execution time slice. public Optional<Boolean> hasNextFile() throws IOException { - if (!paginationController.hasCurLimit()) { + if (runtimeFilterExhausted || !paginationController.hasCurLimit()) { return Optional.of(false); } @@ -339,6 +341,25 @@ public class SeriesScanUtil implements Accountable { && filterAllSatisfy(scanOptions.getPushDownFilter(), firstTimeSeriesMetadata); } + /** + * Returns false when the time range cannot contain any row that may still qualify for TopK, so + * the caller can skip decoding the whole file/chunk/page. + */ + private boolean mayQualifyRuntimeFilterRange(Statistics<? extends Serializable> statistics) { + TopKRuntimeFilter filter = scanOptions.getTopKRuntimeFilter(); + if (filter == null) { + return true; + } + return filter.mayQualifyRange(statistics.getStartTime(), statistics.getEndTime()); + } + + private void skipByTopKRuntimeFilter( + Statistics<? extends Serializable> statistics, Runnable skip) { + if (!mayQualifyRuntimeFilterRange(statistics)) { + skip.run(); + } + } + @SuppressWarnings("squid:S3740") public Statistics currentFileTimeStatistics() { return firstTimeSeriesMetadata.getTimeStatistics(); @@ -369,7 +390,7 @@ public class SeriesScanUtil implements Accountable { * @throws IllegalStateException illegal state */ public Optional<Boolean> hasNextChunk() throws IOException { - if (!paginationController.hasCurLimit()) { + if (runtimeFilterExhausted || !paginationController.hasCurLimit()) { return Optional.of(false); } @@ -419,6 +440,11 @@ public class SeriesScanUtil implements Accountable { return; } + skipByTopKRuntimeFilter(firstChunkMetadata.getStatistics(), this::skipCurrentChunk); + if (firstChunkMetadata == null) { + return; + } + // globalTimeFilter.canSkip() must be FALSE Filter pushDownFilter = scanOptions.getPushDownFilter(); if (pushDownFilter != null && pushDownFilter.canSkip(firstChunkMetadata)) { @@ -542,7 +568,7 @@ public class SeriesScanUtil implements Accountable { @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning public boolean hasNextPage() throws IOException { - if (!paginationController.hasCurLimit()) { + if (runtimeFilterExhausted || !paginationController.hasCurLimit()) { return false; } @@ -919,6 +945,10 @@ public class SeriesScanUtil implements Accountable { } private TsBlock filterAndPaginateCachedBlock(TsBlock tsBlock) { + tsBlock = applyRuntimeFilterToTsBlock(tsBlock); + if (tsBlock == null || tsBlock.isEmpty()) { + return null; + } if (scanOptions.getPushDownFilter() == null) { return paginationController.applyTsBlock(tsBlock); } @@ -937,6 +967,31 @@ public class SeriesScanUtil implements Accountable { paginationController); } + private TsBlock applyRuntimeFilterToTsBlock(TsBlock tsBlock) { + TopKRuntimeFilter filter = scanOptions.getTopKRuntimeFilter(); + if (filter == null) { + return tsBlock; + } + + int positionCount = tsBlock.getPositionCount(); + int keepCount = positionCount; + for (int i = 0; i < positionCount; i++) { + if (!filter.mayQualify(tsBlock.getTimeByIndex(i))) { + keepCount = i; + runtimeFilterExhausted = true; + break; + } + } + + if (keepCount == positionCount) { + return tsBlock; + } + if (keepCount == 0) { + return null; + } + return tsBlock.getRegion(0, keepCount); + } + private TsBlock getTransferedDataTypeTsBlock(TsBlock tsBlock) { Column[] valueColumns = tsBlock.getValueColumns(); int length = tsBlock.getValueColumnCount(); @@ -1406,6 +1461,11 @@ public class SeriesScanUtil implements Accountable { return; } + skipByTopKRuntimeFilter(firstPageReader.getStatistics(), this::skipCurrentPage); + if (firstPageReader == null) { + return; + } + IPageReader pageReader = firstPageReader.getPageReader(); // globalTimeFilter.canSkip() must be FALSE @@ -1892,6 +1952,11 @@ public class SeriesScanUtil implements Accountable { return; } + skipByTopKRuntimeFilter(firstTimeSeriesMetadata.getStatistics(), this::skipCurrentFile); + if (firstTimeSeriesMetadata == null) { + return; + } + // globalTimeFilter.canSkip() must be FALSE Filter pushDownFilter = scanOptions.getPushDownFilter(); if (pushDownFilter != null && pushDownFilter.canSkip(firstTimeSeriesMetadata)) { @@ -2393,9 +2458,24 @@ public class SeriesScanUtil implements Accountable { @Override public boolean hasNextSeqResource() { + TopKRuntimeFilter filter = scanOptions.getTopKRuntimeFilter(); while (dataSource.hasNextSeqResource(curSeqFileIndex, false, deviceID)) { + if (filter != null && dataSource.isRuntimeFilterPruned(true, curSeqFileIndex, false)) { + curSeqFileIndex--; + continue; + } if (dataSource.isSeqSatisfied( deviceID, curSeqFileIndex, scanOptions.getGlobalTimeFilter(), false)) { + if (filter == null + || dataSource.isSeqSatisfiedByRuntimeFilter( + deviceID, curSeqFileIndex, filter, false)) { + break; + } + dataSource.truncateSeqValidSizeForRuntimeFilter(curSeqFileIndex, false); + if (!dataSource.hasValidResource()) { + runtimeFilterExhausted = true; + } + curSeqFileIndex = -1; break; } curSeqFileIndex--; @@ -2405,10 +2485,22 @@ public class SeriesScanUtil implements Accountable { @Override public boolean hasNextUnseqResource() { + TopKRuntimeFilter filter = scanOptions.getTopKRuntimeFilter(); while (dataSource.hasNextUnseqResource(curUnseqFileIndex, false, deviceID)) { + if (filter != null && dataSource.isRuntimeFilterPruned(false, curUnseqFileIndex, false)) { + curUnseqFileIndex++; + continue; + } if (dataSource.isUnSeqSatisfied( deviceID, curUnseqFileIndex, scanOptions.getGlobalTimeFilter(), false)) { - break; + if (filter == null + || dataSource.isUnSeqSatisfiedByRuntimeFilter(curUnseqFileIndex, filter, false)) { + break; + } + dataSource.decreaseValidSizeForRuntimeFilter(false, curUnseqFileIndex, false); + if (!dataSource.hasValidResource()) { + runtimeFilterExhausted = true; + } } curUnseqFileIndex++; } @@ -2522,9 +2614,24 @@ public class SeriesScanUtil implements Accountable { @Override public boolean hasNextSeqResource() { + TopKRuntimeFilter filter = scanOptions.getTopKRuntimeFilter(); while (dataSource.hasNextSeqResource(curSeqFileIndex, true, deviceID)) { + if (filter != null && dataSource.isRuntimeFilterPruned(true, curSeqFileIndex, true)) { + curSeqFileIndex++; + continue; + } if (dataSource.isSeqSatisfied( deviceID, curSeqFileIndex, scanOptions.getGlobalTimeFilter(), false)) { + if (filter == null + || dataSource.isSeqSatisfiedByRuntimeFilter( + deviceID, curSeqFileIndex, filter, false)) { + break; + } + dataSource.truncateSeqValidSizeForRuntimeFilter(curSeqFileIndex, true); + if (!dataSource.hasValidResource()) { + runtimeFilterExhausted = true; + } + curSeqFileIndex = dataSource.getSeqResourcesSize(); break; } curSeqFileIndex++; @@ -2534,10 +2641,22 @@ public class SeriesScanUtil implements Accountable { @Override public boolean hasNextUnseqResource() { + TopKRuntimeFilter filter = scanOptions.getTopKRuntimeFilter(); while (dataSource.hasNextUnseqResource(curUnseqFileIndex, true, deviceID)) { + if (filter != null && dataSource.isRuntimeFilterPruned(false, curUnseqFileIndex, true)) { + curUnseqFileIndex++; + continue; + } if (dataSource.isUnSeqSatisfied( deviceID, curUnseqFileIndex, scanOptions.getGlobalTimeFilter(), false)) { - break; + if (filter == null + || dataSource.isUnSeqSatisfiedByRuntimeFilter(curUnseqFileIndex, filter, false)) { + break; + } + dataSource.decreaseValidSizeForRuntimeFilter(false, curUnseqFileIndex, true); + if (!dataSource.hasValidResource()) { + runtimeFilterExhausted = true; + } } 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 824d46d3a02..383be946624 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 @@ -217,8 +217,17 @@ public abstract class AbstractTableScanOperator extends AbstractSeriesScanOperat @Override public boolean isFinished() throws Exception { - return (retainedTsBlock == null) - && (currentDeviceIndex >= deviceCount || seriesScanOptions.limitConsumedUp()); + if (retainedTsBlock != null) { + return false; + } + if (seriesScanOptions.limitConsumedUp()) { + return true; + } + if (currentDeviceIndex >= deviceCount) { + return true; + } + // QueryDataSource 中 seq/unseq 均无 RF 候选文件时，立即结束（hasNext → false） + return shouldStopScanByRuntimeFilter(); } @Override @@ -241,8 +250,8 @@ public abstract class AbstractTableScanOperator extends AbstractSeriesScanOperat @Override public void initQueryDataSource(IQueryDataSource dataSource) { this.queryDataSource = (QueryDataSource) dataSource; - if (this.seriesScanUtil != null) { - this.seriesScanUtil.initQueryDataSource(queryDataSource); + if (currentDeviceIndex < deviceCount) { + setupCurrentDeviceScan(); } this.resultTsBlockBuilder = new TsBlockBuilder(getResultDataTypes()); this.resultTsBlockBuilder.setMaxTsBlockLineNumber(this.maxTsBlockLineNum); @@ -251,19 +260,29 @@ public abstract class AbstractTableScanOperator extends AbstractSeriesScanOperat } protected void moveToNextDevice() { + if (shouldStopScanByRuntimeFilter()) { + currentDeviceIndex = deviceCount; + return; + } currentDeviceIndex++; if (currentDeviceIndex < deviceCount) { - // 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)); + setupCurrentDeviceScan(); } } + /** Returns true when file-level RF has pruned all seq/unseq files — scan can stop globally. */ + private boolean shouldStopScanByRuntimeFilter() { + 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 a17726d9dc2..e482bf5f9c0 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 @@ -19,11 +19,13 @@ package org.apache.iotdb.db.queryengine.plan.planner; +import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; import org.apache.iotdb.calc.execution.operator.CommonOperatorContext; 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; @@ -46,6 +48,7 @@ import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchem import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName; import org.apache.iotdb.commons.queryengine.plan.relational.planner.Symbol; import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.AggregationNode; +import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.TopKNode; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.FunctionCall; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.LongLiteral; @@ -185,6 +188,7 @@ 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; @@ -222,6 +226,86 @@ public class DataNodeTableOperatorGenerator super(metadata); } + /** + * Producer side: register a shared {@link TopKRuntimeFilter} before visiting TopK children so + * nested Scan nodes can resolve the same instance during operator generation. + */ + private TopKRuntimeFilter registerTopKRuntimeFilterForTopK( + TopKNode node, LocalExecutionPlanContext context) { + if (node == null || !node.isUseTopKRuntimeFilter() || context.dataNodeQueryContext == null) { + return null; + } + return context.dataNodeQueryContext.registerTopKRuntimeFilter( + node.getPlanNodeId().getId(), new TopKRuntimeFilter(node.isTopKRuntimeFilterAscending())); + } + + /** + * Consumer side: lookup the shared filter by root TopK plan node id ({@code + * topKRuntimeFilterSourceId}). + */ + private TopKRuntimeFilter resolveTopKRuntimeFilterForDeviceScan( + DeviceTableScanNode scanNode, LocalExecutionPlanContext context) { + if (scanNode == null + || context.dataNodeQueryContext == null + || !scanNode.getTopKRuntimeFilterSourceId().isPresent()) { + return null; + } + return context.dataNodeQueryContext.getTopKRuntimeFilter( + scanNode.getTopKRuntimeFilterSourceId().get().getId()); + } + + @Override + protected TopKRuntimeFilter getTopKRuntimeFilter( + LocalExecutionPlanContext context, TopKNode node) { + return registerTopKRuntimeFilterForTopK(node, context); + } + + private void applyTopKRuntimeFilter( + SeriesScanOptions.Builder builder, + DeviceTableScanNode scanNode, + LocalExecutionPlanContext context, + TopKRuntimeFilter preResolvedFilter) { + TopKRuntimeFilter filter = + preResolvedFilter != null + ? preResolvedFilter + : resolveTopKRuntimeFilterForDeviceScan(scanNode, context); + if (filter != null) { + builder.withTopKRuntimeFilter(filter); + } + } + + @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; @@ -626,6 +710,7 @@ public class DataNodeTableOperatorGenerator builder.withPushDownOffset( node.isPushLimitToEachDevice() ? 0 : node.getPushDownOffset()); } + applyTopKRuntimeFilter(builder, node, context, null); SeriesScanOptions options = builder.build(); options.setTTLForTableView(viewTTL); seriesScanOptionsList.add(options); @@ -1091,16 +1176,84 @@ public class DataNodeTableOperatorGenerator maxTsBlockLineNum); } + private AbstractTableScanOperator.AbstractTableScanOperatorParameter + constructAbstractTableScanOperatorParameter( + DeviceTableScanNode node, + LocalExecutionPlanContext context, + String className, + Map<String, String> fieldColumnsRenameMap, + long viewTTL, + TopKRuntimeFilter topKRuntimeFilter) { + + CommonTableScanOperatorParameters commonParameter = + new CommonTableScanOperatorParameters(node, fieldColumnsRenameMap, false); + List<IMeasurementSchema> measurementSchemas = commonParameter.measurementSchemas; + List<String> measurementColumnNames = commonParameter.measurementColumnNames; + List<ColumnSchema> columnSchemas = commonParameter.columnSchemas; + int[] columnsIndexArray = commonParameter.columnsIndexArray; + SeriesScanOptions seriesScanOptions = + buildSeriesScanOptions( + context, + commonParameter.columnSchemaMap, + measurementColumnNames, + commonParameter.measurementColumnsIndexMap, + commonParameter.timeColumnName, + node.getTimePredicate(), + node.getPushDownLimit(), + node.getPushDownOffset(), + node.isPushLimitToEachDevice(), + node.getPushDownPredicate(), + node, + topKRuntimeFilter); + seriesScanOptions.setTTLForTableView(viewTTL); + seriesScanOptions.setIsTableViewForTreeModel(node instanceof TreeDeviceViewScanNode); + + OperatorContext operatorContext = addOperatorContext(context, node.getPlanNodeId(), className); + + int maxTsBlockLineNum = TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber(); + if (context.getTypeProvider().getTemplatedInfo() != null) { + maxTsBlockLineNum = + (int) + Math.min( + context.getTypeProvider().getTemplatedInfo().getLimitValue(), maxTsBlockLineNum); + } + + Set<String> allSensors = new HashSet<>(measurementColumnNames); + allSensors.add(""); + + return new AbstractTableScanOperator.AbstractTableScanOperatorParameter( + allSensors, + operatorContext, + node.getPlanNodeId(), + columnSchemas, + columnsIndexArray, + node.getDeviceEntries(), + node.getScanOrder(), + seriesScanOptions, + measurementColumnNames, + measurementSchemas, + maxTsBlockLineNum); + } + // used for TableScanOperator private AbstractTableScanOperator.AbstractTableScanOperatorParameter constructAbstractTableScanOperatorParameter( DeviceTableScanNode node, LocalExecutionPlanContext context) { + return constructAbstractTableScanOperatorParameter(node, context, null); + } + + private AbstractTableScanOperator.AbstractTableScanOperatorParameter + constructAbstractTableScanOperatorParameter( + DeviceTableScanNode node, + LocalExecutionPlanContext context, + TopKRuntimeFilter topKRuntimeFilter) { return constructAbstractTableScanOperatorParameter( node, context, TableScanOperator.class.getSimpleName(), Collections.emptyMap(), - Long.MAX_VALUE); + Long.MAX_VALUE, + topKRuntimeFilter); } @Override @@ -1119,9 +1272,10 @@ public class DataNodeTableOperatorGenerator @Override public Operator visitDeviceTableScan( DeviceTableScanNode node, LocalExecutionPlanContext context) { + TopKRuntimeFilter topKRuntimeFilter = resolveTopKRuntimeFilterForDeviceScan(node, context); AbstractTableScanOperator.AbstractTableScanOperatorParameter parameter = - constructAbstractTableScanOperatorParameter(node, context); + constructAbstractTableScanOperatorParameter(node, context, topKRuntimeFilter); TableScanOperator tableScanOperator = new TableScanOperator(parameter); @@ -1932,6 +2086,34 @@ public class DataNodeTableOperatorGenerator long pushDownOffset, boolean pushLimitToEachDevice, Expression pushDownPredicate) { + return buildSeriesScanOptions( + context, + columnSchemaMap, + measurementColumnNames, + measurementColumnsIndexMap, + timeColumnName, + timePredicate, + pushDownLimit, + pushDownOffset, + pushLimitToEachDevice, + pushDownPredicate, + null, + null); + } + + private SeriesScanOptions buildSeriesScanOptions( + LocalExecutionPlanContext context, + Map<Symbol, ColumnSchema> columnSchemaMap, + List<String> measurementColumnNames, + Map<String, Integer> measurementColumnsIndexMap, + String timeColumnName, + Optional<Expression> timePredicate, + long pushDownLimit, + long pushDownOffset, + boolean pushLimitToEachDevice, + Expression pushDownPredicate, + DeviceTableScanNode scanNode, + TopKRuntimeFilter topKRuntimeFilter) { SeriesScanOptions.Builder scanOptionsBuilder = timePredicate .map(expression -> getSeriesScanOptionsBuilder(context, expression)) @@ -1950,6 +2132,9 @@ public class DataNodeTableOperatorGenerator context.getZoneId(), TimestampPrecisionUtils.currPrecision)); } + if (scanNode != null) { + applyTopKRuntimeFilter(scanOptionsBuilder, scanNode, context, topKRuntimeFilter); + } return scanOptionsBuilder.build(); } 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 a093c7832fd..0cb2b157051 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,6 +112,8 @@ 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 new file mode 100644 index 00000000000..70b2e62abe5 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TopKRuntimeFilterBinder.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.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 06aba49200b..5dd36542656 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,6 +698,9 @@ 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()))); } boxValue.add( @@ -1049,6 +1052,9 @@ 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()) { + boxValue.add("TOPN OPT"); + } return render(node, boxValue, context); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/parameter/SeriesScanOptions.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/parameter/SeriesScanOptions.java index 38335a788f5..e66d4aadc26 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/parameter/SeriesScanOptions.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/parameter/SeriesScanOptions.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.queryengine.plan.planner.plan.parameter; +import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; import org.apache.iotdb.commons.path.AlignedFullPath; import org.apache.iotdb.commons.path.IFullPath; import org.apache.iotdb.commons.path.NonAlignedFullPath; @@ -55,6 +56,7 @@ public class SeriesScanOptions implements Accountable { private PaginationController paginationController; private boolean isTableViewForTreeModel; private long ttlForTableView = Long.MAX_VALUE; + private final TopKRuntimeFilter topKRuntimeFilter; private static final long INSTANCE_SIZE = RamUsageEstimator.shallowSizeOfInstance( TreeNonAlignedDeviceViewAggregationScanOperator.class); @@ -66,7 +68,8 @@ public class SeriesScanOptions implements Accountable { long pushDownOffset, Set<String> allSensors, boolean pushLimitToEachDevice, - boolean isTableViewForTreeModel) { + boolean isTableViewForTreeModel, + TopKRuntimeFilter topKRuntimeFilter) { this.globalTimeFilter = globalTimeFilter; this.originalTimeFilter = globalTimeFilter; this.pushDownFilter = pushDownFilter; @@ -75,6 +78,7 @@ public class SeriesScanOptions implements Accountable { this.allSensors = allSensors; this.pushLimitToEachDevice = pushLimitToEachDevice; this.isTableViewForTreeModel = isTableViewForTreeModel; + this.topKRuntimeFilter = topKRuntimeFilter; } public static SeriesScanOptions getDefaultSeriesScanOptions(IFullPath seriesPath) { @@ -186,6 +190,10 @@ public class SeriesScanOptions implements Accountable { && (paginationController != null && !paginationController.hasCurLimit()); } + public TopKRuntimeFilter getTopKRuntimeFilter() { + return topKRuntimeFilter; + } + public static class Builder { private Filter globalTimeFilter = null; @@ -197,6 +205,7 @@ public class SeriesScanOptions implements Accountable { private boolean pushLimitToEachDevice = true; private boolean isTableViewForTreeModel = false; + private TopKRuntimeFilter topKRuntimeFilter; public Builder withGlobalTimeFilter(Filter globalTimeFilter) { this.globalTimeFilter = globalTimeFilter; @@ -228,6 +237,11 @@ public class SeriesScanOptions implements Accountable { return this; } + public Builder withTopKRuntimeFilter(TopKRuntimeFilter topKRuntimeFilter) { + this.topKRuntimeFilter = topKRuntimeFilter; + return this; + } + public void withAllSensors(Set<String> allSensors) { this.allSensors = allSensors; } @@ -240,7 +254,8 @@ public class SeriesScanOptions implements Accountable { pushDownOffset, allSensors, pushLimitToEachDevice, - isTableViewForTreeModel); + isTableViewForTreeModel, + topKRuntimeFilter); } } } 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 fd9bc35fb4c..893420125d7 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 @@ -47,6 +47,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExchangeNode import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.DataNodeLocationSupplierFactory; import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.DistributedOptimizeFactory; import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.PlanOptimizer; +import org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations.TopKRuntimeFilterOptimizer; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExplainOutputFormat; import java.util.Collections; @@ -176,7 +177,27 @@ public class TableDistributedPlanner { .forEach((k, v) -> mppQueryContext.getTypeProvider().putTableModelType(k, v)); // add exchange node for distributed plan - return new AddExchangeNodes(mppQueryContext).addExchangeNodes(distributedPlan, planContext); + PlanNode planWithExchange = + new AddExchangeNodes(mppQueryContext).addExchangeNodes(distributedPlan, planContext); + + // Mark TopK runtime filter producer/consumer after exchange insertion. + if (analysis.isQuery()) { + planWithExchange = + new TopKRuntimeFilterOptimizer() + .optimize( + planWithExchange, + new PlanOptimizer.Context( + mppQueryContext.getSession(), + analysis, + metadata, + mppQueryContext, + new SymbolAllocator(), + mppQueryContext.getQueryId(), + NOOP, + PlanOptimizersStatsCollector.createPlanOptimizersStatsCollector())); + } + + return planWithExchange; } private DistributedQueryPlan generateDistributedPlan( 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 c02c74058af..bc7773a50c8 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 @@ -79,6 +79,9 @@ 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; + protected DeviceTableScanNode() {} public DeviceTableScanNode( @@ -129,20 +132,23 @@ public class DeviceTableScanNode extends TableScanNode { @Override public DeviceTableScanNode clone() { - return new DeviceTableScanNode( - getPlanNodeId(), - qualifiedObjectName, - outputSymbols, - assignments, - deviceEntries, - tagAndAttributeIndexMap, - scanOrder, - timePredicate, - pushDownPredicate, - pushDownLimit, - pushDownOffset, - pushLimitToEachDevice, - containsNonAlignedDevice); + DeviceTableScanNode cloned = + new DeviceTableScanNode( + getPlanNodeId(), + qualifiedObjectName, + outputSymbols, + assignments, + deviceEntries, + tagAndAttributeIndexMap, + scanOrder, + timePredicate, + pushDownPredicate, + pushDownLimit, + pushDownOffset, + pushLimitToEachDevice, + containsNonAlignedDevice); + cloned.topKRuntimeFilterSourceId = topKRuntimeFilterSourceId; + return cloned; } protected static void serializeMemberVariables( @@ -170,6 +176,13 @@ 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); + } } protected static void serializeMemberVariables( @@ -198,6 +211,13 @@ 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); + } } protected static void deserializeMemberVariables( @@ -227,6 +247,10 @@ public class DeviceTableScanNode extends TableScanNode { } node.pushLimitToEachDevice = ReadWriteIOUtils.readBool(byteBuffer); + + if (ReadWriteIOUtils.readBool(byteBuffer)) { + node.topKRuntimeFilterSourceId = PlanNodeId.deserialize(byteBuffer); + } } @Override @@ -307,6 +331,14 @@ public class DeviceTableScanNode extends TableScanNode { this.containsNonAlignedDevice = true; } + public Optional<PlanNodeId> getTopKRuntimeFilterSourceId() { + return Optional.ofNullable(topKRuntimeFilterSourceId); + } + + public void setTopKRuntimeFilterSourceId(PlanNodeId topKRuntimeFilterSourceId) { + this.topKRuntimeFilterSourceId = topKRuntimeFilterSourceId; + } + public String toString() { return "DeviceTableScanNode-" + this.getPlanNodeId(); } 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 new file mode 100644 index 00000000000..f56000b9f89 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizer.java @@ -0,0 +1,95 @@ +/* + * 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.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.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. + */ +public class TopKRuntimeFilterOptimizer implements PlanOptimizer { + + @Override + public PlanNode optimize(PlanNode plan, Context context) { + if (!context.getAnalysis().isQuery()) { + return plan; + } + return plan.accept(new Rewriter(), null); + } + + private static class Rewriter implements PlanVisitor<PlanNode, Void> { + + @Override + public PlanNode visitPlan(PlanNode node, Void context) { + PlanNode newNode = node.clone(); + for (PlanNode child : node.getChildren()) { + newNode.addChild(child.accept(this, context)); + } + return newNode; + } + + @Override + public PlanNode visitTopK(TopKNode node, Void context) { + 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()); + } + 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 new file mode 100644 index 00000000000..a0704a72361 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterUtils.java @@ -0,0 +1,74 @@ +/* + * 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.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 { + + private TopKRuntimeFilterUtils() {} + + public static boolean isOrderByTimeOnly(OrderingScheme orderingScheme) { + if (orderingScheme.getOrderBy().size() != 1) { + return false; + } + 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 0c86c887fe8..3d5b724754b 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 @@ -19,6 +19,7 @@ package org.apache.iotdb.db.storageengine.dataregion.read; +import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -70,11 +71,21 @@ public class QueryDataSource implements IQueryDataSource { private String databaseName = null; + /** + * Remaining seq/unseq TsFileResources that may still contain rows qualifying for TopK runtime + * filter. Initialized to list sizes and decremented when a file is skipped at resource level. + * When both counters reach zero, the scan can exit early. + */ + private int seqValidSize; + + private int unseqValidSize; + 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( @@ -82,6 +93,7 @@ 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 @@ -91,6 +103,83 @@ public class QueryDataSource implements IQueryDataSource { this.unseqResources = other.unseqResources; this.unSeqFileOrderIndex = other.unSeqFileOrderIndex; this.databaseName = other.databaseName; + this.seqValidSize = other.seqValidSize; + this.unseqValidSize = other.unseqValidSize; + } + + /** Initializes seq/unseq valid resource counters from list sizes. */ + private void initValidSize() { + seqValidSize = getSeqResourcesSize(); + unseqValidSize = getUnseqResourcesSize(); + } + + public int getSeqValidSize() { + return seqValidSize; + } + + public int getUnseqValidSize() { + return unseqValidSize; + } + + /** Returns true if either seq or unseq may still contain RF-qualifying resources. */ + public boolean hasValidResource() { + return seqValidSize > 0 || unseqValidSize > 0; + } + + /** + * Decrements the remaining file count when a TsFileResource is pruned by TopK runtime filter at + * resource level. Counters are shared across all devices in this scan; when both reach zero, the + * entire scan can stop without switching to subsequent devices. + */ + public void decreaseValidSize(boolean isSeq) { + if (isSeq) { + if (seqValidSize > 0) { + seqValidSize--; + } + } else if (unseqValidSize > 0) { + unseqValidSize--; + } + } + + /** + * Decrements validSize when a TsFileResource is first pruned by TopK runtime filter at resource + * level. File lists are time-ordered, so RF pruning is monotonic from one end: ASC prunes a + * suffix, DESC prunes a prefix. {@link #seqValidSize} / {@link #unseqValidSize} then double as + * the cross-device scan watermark. + */ + public void decreaseValidSizeForRuntimeFilter(boolean isSeq, int index, boolean ascending) { + if (isRuntimeFilterPruned(isSeq, index, ascending)) { + return; + } + decreaseValidSize(isSeq); + } + + /** + * Truncates seq validSize when RF fails at {@code index}. Seq files are time-ordered, so all + * files on the pruned side of {@code index} can be excluded at once: ASC suffix {@code [index, + * size)}, DESC prefix {@code [0, index]}. + */ + public void truncateSeqValidSizeForRuntimeFilter(int index, boolean ascending) { + int size = getSeqResourcesSize(); + if (ascending) { + seqValidSize = Math.min(seqValidSize, index); + } else { + seqValidSize = Math.min(seqValidSize, size - index - 1); + } + } + + /** + * Returns true if this file index was already pruned by RF at resource level on a prior device. + */ + public boolean isRuntimeFilterPruned(boolean isSeq, int index, boolean ascending) { + int validSize = isSeq ? seqValidSize : unseqValidSize; + int size = isSeq ? getSeqResourcesSize() : getUnseqResourcesSize(); + if (ascending) { + // ASC: pruned files form suffix [validSize, size) + return index >= validSize; + } + // DESC: pruned files form prefix [0, size - validSize) + return index < size - validSize; } public List<TsFileResource> getSeqResources() { @@ -144,6 +233,11 @@ public class QueryDataSource implements IQueryDataSource { return curSeqSatisfied; } + public boolean isSeqSatisfiedByRuntimeFilter( + IDeviceID deviceID, int curIndex, TopKRuntimeFilter filter, boolean debug) { + return isResourceSatisfiedByRuntimeFilter(curIndex, filter, true, debug); + } + public long getCurrentSeqOrderTime(int curIndex) { if (curIndex != this.curSeqIndex) { throw new IllegalArgumentException( @@ -196,6 +290,11 @@ public class QueryDataSource implements IQueryDataSource { return curUnSeqSatisfied; } + public boolean isUnSeqSatisfiedByRuntimeFilter( + int curIndex, TopKRuntimeFilter filter, boolean debug) { + return isResourceSatisfiedByRuntimeFilter(curIndex, filter, false, debug); + } + public long getCurrentUnSeqOrderTime(int curIndex) { if (curIndex != this.curUnSeqIndex) { throw new IllegalArgumentException( @@ -265,6 +364,22 @@ public class QueryDataSource implements IQueryDataSource { curUnSeqSatisfied = null; } + private boolean isResourceSatisfiedByRuntimeFilter( + int curIndex, TopKRuntimeFilter filter, boolean isSeq, boolean debug) { + if (filter == null) { + return true; + } + 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; + return filter.mayQualifyRange(startTime, endTime); + } + public String getDatabaseName() { if (databaseName == null) { List<TsFileResource> resources = !seqResources.isEmpty() ? seqResources : unseqResources; 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 new file mode 100644 index 00000000000..c67506f3c0b --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizerTest.java @@ -0,0 +1,78 @@ +/* + * 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.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.planner.node.DeviceTableScanNode; +import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExchangeNode; + +import org.junit.Assert; +import org.junit.Test; + +import java.util.Collections; +import java.util.Optional; + +public class TopKRuntimeFilterOptimizerTest { + + @Test + public void marksTopKAndScanWhenQualified() { + 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 optimized = (TopKNode) new TopKRuntimeFilterOptimizer().optimize(topKNode, null); + + Assert.assertTrue(optimized.isUseTopKRuntimeFilter()); + Assert.assertTrue(optimized.isTopKRuntimeFilterAscending()); + DeviceTableScanNode optimizedScan = + (DeviceTableScanNode) optimized.getChildren().get(0).getChildren().get(0); + Assert.assertEquals(topKId, optimizedScan.getTopKRuntimeFilterSourceId().orElse(null)); + } + + @Test + public void skipsTopKWithExchangeChild() { + 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 optimized = (TopKNode) new TopKRuntimeFilterOptimizer().optimize(topKNode, null); + + Assert.assertFalse(optimized.isUseTopKRuntimeFilter()); + } + + private TopKNode createTopK(PlanNodeId id, SortOrder sortOrder) { + Symbol timeSymbol = new Symbol("time"); + OrderingScheme orderingScheme = + new OrderingScheme(Collections.singletonList(timeSymbol), sortOrder); + return new TopKNode(id, orderingScheme, 10, Collections.singletonList(timeSymbol), false); + } + + private DeviceTableScanNode createScan(PlanNodeId id) { + return new DeviceTableScanNode( + id, null, Collections.emptyList(), Collections.emptyMap(), Collections.emptyMap()); + } +} 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 34376dcdbd5..70c866d5e0d 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 @@ -50,6 +50,13 @@ public class TopKNode extends MultiChildProcessNode { private final boolean childrenDataInOrder; + // Marked during distributed optimize when TopK runtime filter should be generated. + private boolean useTopKRuntimeFilter; + + // Sort direction of the runtime filter threshold; meaningful only when useTopKRuntimeFilter is + // true. + private boolean topKRuntimeFilterAscending = true; + public TopKNode( PlanNodeId id, OrderingScheme scheme, @@ -79,7 +86,11 @@ public class TopKNode extends MultiChildProcessNode { @Override public PlanNode clone() { - return new TopKNode(getPlanNodeId(), orderingScheme, count, outputSymbols, childrenDataInOrder); + TopKNode cloned = + new TopKNode(getPlanNodeId(), orderingScheme, count, outputSymbols, childrenDataInOrder); + cloned.useTopKRuntimeFilter = useTopKRuntimeFilter; + cloned.topKRuntimeFilterAscending = topKRuntimeFilterAscending; + return cloned; } @Override @@ -102,6 +113,8 @@ public class TopKNode extends MultiChildProcessNode { Symbol.serialize(symbol, byteBuffer); } ReadWriteIOUtils.write(childrenDataInOrder, byteBuffer); + ReadWriteIOUtils.write(useTopKRuntimeFilter, byteBuffer); + ReadWriteIOUtils.write(topKRuntimeFilterAscending, byteBuffer); } @Override @@ -114,6 +127,8 @@ public class TopKNode extends MultiChildProcessNode { Symbol.serialize(symbol, stream); } ReadWriteIOUtils.write(childrenDataInOrder, stream); + ReadWriteIOUtils.write(useTopKRuntimeFilter, stream); + ReadWriteIOUtils.write(topKRuntimeFilterAscending, stream); } public static TopKNode deserialize(ByteBuffer byteBuffer) { @@ -125,8 +140,14 @@ 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); PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer); - return new TopKNode(planNodeId, orderingScheme, count, outputSymbols, childrenDataInOrder); + TopKNode topKNode = + new TopKNode(planNodeId, orderingScheme, count, outputSymbols, childrenDataInOrder); + topKNode.useTopKRuntimeFilter = useTopKRuntimeFilter; + topKNode.topKRuntimeFilterAscending = topKRuntimeFilterAscending; + return topKNode; } @Override @@ -139,7 +160,11 @@ public class TopKNode extends MultiChildProcessNode { checkArgument( children.size() == newChildren.size(), QueryMessages.EXCEPTION_WRONG_NUMBER_OF_NEW_CHILDREN_817AF800); - return new TopKNode(id, newChildren, orderingScheme, count, outputSymbols, childrenDataInOrder); + TopKNode topKNode = + new TopKNode(id, newChildren, orderingScheme, count, outputSymbols, childrenDataInOrder); + topKNode.useTopKRuntimeFilter = useTopKRuntimeFilter; + topKNode.topKRuntimeFilterAscending = topKRuntimeFilterAscending; + return topKNode; } public OrderingScheme getOrderingScheme() { @@ -154,6 +179,22 @@ 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; + } + @Override public boolean equals(Object o) { if (this == o) return true; @@ -162,12 +203,20 @@ public class TopKNode extends MultiChildProcessNode { TopKNode sortNode = (TopKNode) o; return Objects.equal(orderingScheme, sortNode.orderingScheme) && Objects.equal(outputSymbols, sortNode.outputSymbols) - && Objects.equal(count, sortNode.count); + && Objects.equal(count, sortNode.count) + && useTopKRuntimeFilter == sortNode.useTopKRuntimeFilter + && topKRuntimeFilterAscending == sortNode.topKRuntimeFilterAscending; } @Override public int hashCode() { - return Objects.hashCode(super.hashCode(), orderingScheme, outputSymbols, count); + return Objects.hashCode( + super.hashCode(), + orderingScheme, + outputSymbols, + count, + useTopKRuntimeFilter, + topKRuntimeFilterAscending); } @Override
