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 f0075bb9ccaf18a330c8e3c1515dda8615e2d9ec Author: Weihao Li <[email protected]> AuthorDate: Fri Jul 10 13:25:32 2026 +0800 draft Signed-off-by: Weihao Li <[email protected]> --- .../execution/operator/source/SeriesScanUtil.java | 24 ++--- .../dataregion/read/QueryDataSource.java | 100 ++++++++------------- 2 files changed, 49 insertions(+), 75 deletions(-) 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 ad87d994983..e17d0d0fa22 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 @@ -2460,7 +2460,7 @@ public class SeriesScanUtil implements Accountable { public boolean hasNextSeqResource() { TopKRuntimeFilter filter = scanOptions.getTopKRuntimeFilter(); while (dataSource.hasNextSeqResource(curSeqFileIndex, false, deviceID)) { - if (filter != null && dataSource.isRuntimeFilterPruned(true, curSeqFileIndex, false)) { + if (filter != null && dataSource.isRuntimeFilterPruned(true, curSeqFileIndex)) { curSeqFileIndex--; continue; } @@ -2471,12 +2471,11 @@ public class SeriesScanUtil implements Accountable { deviceID, curSeqFileIndex, filter, false)) { break; } - dataSource.truncateSeqValidSizeForRuntimeFilter(curSeqFileIndex, false); + dataSource.setSeqTsFileResourceInvalidated(curSeqFileIndex); if (!dataSource.hasValidResource()) { runtimeFilterExhausted = true; + return false; } - curSeqFileIndex = -1; - break; } curSeqFileIndex--; } @@ -2487,7 +2486,7 @@ public class SeriesScanUtil implements Accountable { public boolean hasNextUnseqResource() { TopKRuntimeFilter filter = scanOptions.getTopKRuntimeFilter(); while (dataSource.hasNextUnseqResource(curUnseqFileIndex, false, deviceID)) { - if (filter != null && dataSource.isRuntimeFilterPruned(false, curUnseqFileIndex, false)) { + if (filter != null && dataSource.isRuntimeFilterPruned(false, curUnseqFileIndex)) { curUnseqFileIndex++; continue; } @@ -2497,9 +2496,10 @@ public class SeriesScanUtil implements Accountable { || dataSource.isUnSeqSatisfiedByRuntimeFilter(curUnseqFileIndex, filter, false)) { break; } - dataSource.decreaseValidSizeForRuntimeFilter(false, curUnseqFileIndex, false); + dataSource.setUnseqTsFileResourceInvalidated(curUnseqFileIndex); if (!dataSource.hasValidResource()) { runtimeFilterExhausted = true; + return false; } } curUnseqFileIndex++; @@ -2616,7 +2616,7 @@ public class SeriesScanUtil implements Accountable { public boolean hasNextSeqResource() { TopKRuntimeFilter filter = scanOptions.getTopKRuntimeFilter(); while (dataSource.hasNextSeqResource(curSeqFileIndex, true, deviceID)) { - if (filter != null && dataSource.isRuntimeFilterPruned(true, curSeqFileIndex, true)) { + if (filter != null && dataSource.isRuntimeFilterPruned(true, curSeqFileIndex)) { curSeqFileIndex++; continue; } @@ -2627,12 +2627,11 @@ public class SeriesScanUtil implements Accountable { deviceID, curSeqFileIndex, filter, false)) { break; } - dataSource.truncateSeqValidSizeForRuntimeFilter(curSeqFileIndex, true); + dataSource.setSeqTsFileResourceInvalidated(curSeqFileIndex); if (!dataSource.hasValidResource()) { runtimeFilterExhausted = true; + return false; } - curSeqFileIndex = dataSource.getSeqResourcesSize(); - break; } curSeqFileIndex++; } @@ -2643,7 +2642,7 @@ public class SeriesScanUtil implements Accountable { public boolean hasNextUnseqResource() { TopKRuntimeFilter filter = scanOptions.getTopKRuntimeFilter(); while (dataSource.hasNextUnseqResource(curUnseqFileIndex, true, deviceID)) { - if (filter != null && dataSource.isRuntimeFilterPruned(false, curUnseqFileIndex, true)) { + if (filter != null && dataSource.isRuntimeFilterPruned(false, curUnseqFileIndex)) { curUnseqFileIndex++; continue; } @@ -2653,9 +2652,10 @@ public class SeriesScanUtil implements Accountable { || dataSource.isUnSeqSatisfiedByRuntimeFilter(curUnseqFileIndex, filter, false)) { break; } - dataSource.decreaseValidSizeForRuntimeFilter(false, curUnseqFileIndex, true); + dataSource.setUnseqTsFileResourceInvalidated(curUnseqFileIndex); if (!dataSource.hasValidResource()) { runtimeFilterExhausted = true; + return false; } } curUnseqFileIndex++; 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 3d5b724754b..09da92f7136 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 @@ -20,6 +20,7 @@ package org.apache.iotdb.db.storageengine.dataregion.read; import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter; +import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -27,6 +28,7 @@ import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.read.filter.basic.Filter; import java.util.ArrayList; +import java.util.BitSet; import java.util.Comparator; import java.util.List; import java.util.TreeMap; @@ -72,13 +74,18 @@ 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. + * Physical seq/unseq TsFile indices pruned by TopK runtime filter at resource level. Shared + * across all devices in this scan. */ - private int seqValidSize; + private final BitSet seqInvalidatedByRuntimeFilter = new BitSet(); - private int unseqValidSize; + private final BitSet unseqInvalidatedByRuntimeFilter = new BitSet(); + + /** + * Remaining seq + unseq TsFileResources that may still contain RF-qualifying rows. Decremented + * once per newly marked file; when zero, the scan can exit early. + */ + private int validSize; private static final Comparator<Long> descendingComparator = (o1, o2) -> Long.compare(o2, o1); @@ -103,83 +110,50 @@ 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; + this.validSize = other.validSize; + this.seqInvalidatedByRuntimeFilter.or(other.seqInvalidatedByRuntimeFilter); + this.unseqInvalidatedByRuntimeFilter.or(other.unseqInvalidatedByRuntimeFilter); } - /** Initializes seq/unseq valid resource counters from list sizes. */ private void initValidSize() { - seqValidSize = getSeqResourcesSize(); - unseqValidSize = getUnseqResourcesSize(); - } - - public int getSeqValidSize() { - return seqValidSize; + validSize = getSeqResourcesSize() + getUnseqResourcesSize(); } - public int getUnseqValidSize() { - return unseqValidSize; + @TestOnly + public int getValidSize() { + return validSize; } - /** Returns true if either seq or unseq may still contain RF-qualifying resources. */ + /** Returns true if any seq/unseq file 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--; - } + return validSize > 0; } - /** - * 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; + /** Marks the seq file at physical {@code index} as pruned by TopK RF at resource level. */ + public void setSeqTsFileResourceInvalidated(int physicalIndex) { + if (!seqInvalidatedByRuntimeFilter.get(physicalIndex)) { + seqInvalidatedByRuntimeFilter.set(physicalIndex); + validSize--; } - 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]}. + * Marks the unseq file at traversal {@code orderIndex} as pruned by TopK RF at resource level. */ - 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); + public void setUnseqTsFileResourceInvalidated(int orderIndex) { + int physicalIndex = unSeqFileOrderIndex[orderIndex]; + if (!unseqInvalidatedByRuntimeFilter.get(physicalIndex)) { + unseqInvalidatedByRuntimeFilter.set(physicalIndex); + validSize--; } } - /** - * 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; + /** 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 (isSeq) { + return seqInvalidatedByRuntimeFilter.get(index); } - // DESC: pruned files form prefix [0, size - validSize) - return index < size - validSize; + return unseqInvalidatedByRuntimeFilter.get(unSeqFileOrderIndex[index]); } public List<TsFileResource> getSeqResources() {
