This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 66e4b98ebd8 Optimize TopK query of TableModel by RuntimeFilter (#18204)
66e4b98ebd8 is described below
commit 66e4b98ebd8e9300c9e81c656e7e9eec4108b7f9
Author: Weihao Li <[email protected]>
AuthorDate: Thu Jul 16 09:06:34 2026 +0800
Optimize TopK query of TableModel by RuntimeFilter (#18204)
---
.../recent/IoTDBTopKRuntimeFilterScanPathIT.java | 353 +++++++++++++++++++++
.../org/apache/iotdb/calc/i18n/CalcMessages.java | 2 +
.../org/apache/iotdb/calc/i18n/CalcMessages.java | 2 +
.../calc/execution/filter/TopKRuntimeFilter.java | 67 ++++
.../operator/process/TableTopKOperator.java | 40 ++-
.../execution/operator/process/TopKOperator.java | 36 +++
.../calc/plan/planner/TableOperatorGenerator.java | 25 ++
.../execution/filter/TopKRuntimeFilterTest.java | 73 +++++
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 14 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 12 +
.../execution/fragment/DataNodeQueryContext.java | 21 ++
.../execution/operator/source/SeriesScanUtil.java | 133 +++++++-
.../relational/AbstractTableScanOperator.java | 23 +-
.../ExternalTsFileTableScanOperator.java | 13 +-
.../planner/DataNodeTableOperatorGenerator.java | 158 ++++++++-
.../plan/planner/plan/node/PlanGraphPrinter.java | 7 +
.../planner/plan/parameter/SeriesScanOptions.java | 19 +-
.../distribute/TableDistributedPlanGenerator.java | 11 +-
.../distribute/TableDistributedPlanner.java | 5 +-
.../planner/node/DeviceTableScanNode.java | 49 ++-
.../optimizations/LogicalOptimizeFactory.java | 1 +
.../optimizations/TopKRuntimeFilterOptimizer.java | 83 +++++
.../dataregion/read/QueryDataSource.java | 114 +++++++
...alTsFileTableScanOperatorRuntimeFilterTest.java | 127 ++++++++
.../planner/assertions/PlanMatchPattern.java | 25 ++
.../TopKRuntimeFilterDistributedPlanTest.java | 200 ++++++++++++
.../TopKRuntimeFilterTreeViewPlanTest.java | 119 +++++++
.../read/QueryDataSourceRuntimeFilterTest.java | 76 +++++
.../conf/iotdb-system.properties.template | 8 +
.../plan/relational/planner/node/TopKNode.java | 55 +++-
30 files changed, 1831 insertions(+), 40 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTopKRuntimeFilterScanPathIT.java
b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTopKRuntimeFilterScanPathIT.java
new file mode 100644
index 00000000000..7697bc81649
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTopKRuntimeFilterScanPathIT.java
@@ -0,0 +1,353 @@
+/*
+ * 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.relational.it.query.recent;
+
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableClusterIT;
+import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
+import org.apache.iotdb.itbase.env.BaseEnv;
+
+import org.apache.tsfile.enums.ColumnCategory;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.exception.write.WriteProcessException;
+import org.apache.tsfile.file.metadata.TableSchema;
+import org.apache.tsfile.write.TsFileWriter;
+import org.apache.tsfile.write.record.Tablet;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.apache.iotdb.db.it.utils.TestUtils.prepareTableData;
+import static org.apache.iotdb.db.it.utils.TestUtils.tableResultSetEqualTest;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
+public class IoTDBTopKRuntimeFilterScanPathIT {
+
+ private static final String TREE_VIEW_DATABASE = "db_topk_rf_view";
+ private static final String TABLE_DATABASE = "db_topk_rf_table";
+ private static final String READ_TSFILE_DATABASE = "db_topk_rf_read_tsfile";
+
+ private static File tmpDir;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().initClusterEnvironment();
+ tmpDir = new File(Files.createTempDirectory("topk-rf-scan-path").toUri());
+ insertTreeModelData();
+ createTreeView();
+ createTableModelDatabase();
+ createReadTsFileDatabase();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ if (tmpDir != null) {
+ deleteDirectory(tmpDir);
+ }
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @Test
+ public void orderByTimeLimitOnTreeView() {
+ String[] expectedHeader = new String[] {"time", "device_id", "s1"};
+ String[] retArray =
+ new String[] {
+ "1970-01-01T00:00:00.032Z,d3,333,",
"1970-01-01T00:00:00.031Z,d2,222,",
+ };
+ tableResultSetEqualTest(
+ "SELECT * FROM table1 ORDER BY time DESC LIMIT 2",
+ expectedHeader,
+ retArray,
+ TREE_VIEW_DATABASE);
+ }
+
+ @Test
+ public void orderByTimeAscLimitOnTreeView() {
+ String[] expectedHeader = new String[] {"time", "device_id", "s1"};
+ String[] retArray =
+ new String[] {
+ "1970-01-01T00:00:00.010Z,d1,1,", "1970-01-01T00:00:00.011Z,d2,2,",
+ };
+ tableResultSetEqualTest(
+ "SELECT * FROM table1 ORDER BY time ASC LIMIT 2",
+ expectedHeader,
+ retArray,
+ TREE_VIEW_DATABASE);
+ }
+
+ @Test
+ public void orderByTimeDescLimitOnTable() {
+ String[] expectedHeader = new String[] {"time", "device_id", "s1"};
+ String[] retArray =
+ new String[] {
+ "1970-01-01T00:00:00.032Z,d3,333,",
"1970-01-01T00:00:00.031Z,d2,222,",
+ };
+ tableResultSetEqualTest(
+ "SELECT * FROM table1 ORDER BY time DESC LIMIT 2",
+ expectedHeader,
+ retArray,
+ TABLE_DATABASE);
+ }
+
+ @Test
+ public void orderByTimeAscLimitOnTable() {
+ String[] expectedHeader = new String[] {"time", "device_id", "s1"};
+ String[] retArray =
+ new String[] {
+ "1970-01-01T00:00:00.010Z,d1,1,", "1970-01-01T00:00:00.011Z,d2,2,",
+ };
+ tableResultSetEqualTest(
+ "SELECT * FROM table1 ORDER BY time ASC LIMIT 2", expectedHeader,
retArray, TABLE_DATABASE);
+ }
+
+ @Test
+ public void unionAllSiblingTopKOrderByTimeLimit() {
+ String[] expectedHeader = new String[] {"time", "device_id", "s1"};
+ String[] retArray =
+ new String[] {
+ "1970-01-01T00:00:00.030Z,d1,111,",
"1970-01-01T00:00:00.032Z,d3,333,",
+ };
+ tableResultSetEqualTest(
+ "(SELECT time, device_id, s1 FROM table1 WHERE device_id = 'd1' ORDER
BY time DESC LIMIT 1)"
+ + " UNION ALL"
+ + " (SELECT time, device_id, s1 FROM table1 WHERE device_id = 'd3'
ORDER BY time DESC LIMIT 1)",
+ expectedHeader,
+ retArray,
+ TABLE_DATABASE);
+ }
+
+ @Test
+ public void orderByTimeAscLimitOnReadTsFile() throws Exception {
+ File tsFile = new File(tmpDir, "topk-rf-asc.tsfile");
+ try (TsFileWriter writer = new TsFileWriter(tsFile)) {
+ generateTableWithDistinctDeviceTimes(
+ writer, "table1", Arrays.asList("tag1", "tag2"), Arrays.asList("s1",
"s2"), 1, 3);
+ }
+
+ String[] expectedHeader = new String[] {"time", "tag1", "tag2", "s1",
"s2"};
+ String[] retArray =
+ new String[] {
+ "1970-01-01T00:00:00.001Z,tag1_1,tag2_1,1,1,",
+ "1970-01-01T00:00:00.002Z,tag1_2,tag2_2,2,2,",
+ };
+ tableResultSetEqualTest(
+ "SELECT time, tag1, tag2, s1, s2 FROM read_tsfile(PATHS => '"
+ + toSqlPath(tsFile)
+ + "', TABLE_NAME => 'table1') ORDER BY time ASC LIMIT 2",
+ expectedHeader,
+ retArray,
+ READ_TSFILE_DATABASE);
+ }
+
+ @Test
+ public void orderByTimeLimitOnReadTsFile() throws Exception {
+ File tsFile = new File(tmpDir, "topk-rf.tsfile");
+ try (TsFileWriter writer = new TsFileWriter(tsFile)) {
+ generateTableWithDistinctDeviceTimes(
+ writer, "table1", Arrays.asList("tag1", "tag2"), Arrays.asList("s1",
"s2"), 1, 3);
+ }
+
+ String[] expectedHeader = new String[] {"time", "tag1", "tag2", "s1",
"s2"};
+ String[] retArray =
+ new String[] {
+ "1970-01-01T00:00:00.003Z,tag1_3,tag2_3,3,3,",
+ "1970-01-01T00:00:00.002Z,tag1_2,tag2_2,2,2,",
+ };
+ tableResultSetEqualTest(
+ "SELECT time, tag1, tag2, s1, s2 FROM read_tsfile(PATHS => '"
+ + toSqlPath(tsFile)
+ + "', TABLE_NAME => 'table1') ORDER BY time DESC LIMIT 2",
+ expectedHeader,
+ retArray,
+ READ_TSFILE_DATABASE);
+ }
+
+ private static void insertTreeModelData() throws Exception {
+ try (Connection connection =
EnvFactory.getEnv().getConnection(BaseEnv.TREE_SQL_DIALECT);
+ Statement statement = connection.createStatement()) {
+ statement.execute("CREATE DATABASE root.db_topk_rf");
+ statement.execute(
+ "CREATE TIMESERIES root.db_topk_rf.d1.s1 WITH DATATYPE=INT32,
ENCODING=RLE");
+ statement.execute(
+ "CREATE TIMESERIES root.db_topk_rf.d2.s1 WITH DATATYPE=INT32,
ENCODING=RLE");
+ statement.execute(
+ "CREATE TIMESERIES root.db_topk_rf.d3.s1 WITH DATATYPE=INT32,
ENCODING=RLE");
+ statement.execute(
+ "INSERT INTO root.db_topk_rf.d1(timestamp,s1) VALUES(10, 1), (20,
11), (30, 111)");
+ statement.execute(
+ "INSERT INTO root.db_topk_rf.d2(timestamp,s1) VALUES(11, 2), (21,
22), (31, 222)");
+ statement.execute(
+ "INSERT INTO root.db_topk_rf.d3(timestamp,s1) VALUES(12, 3), (22,
33), (32, 333)");
+ }
+ }
+
+ private static void createTreeView() throws Exception {
+ try (Connection connection = EnvFactory.getEnv().getTableConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("CREATE DATABASE " + TREE_VIEW_DATABASE);
+ statement.execute("USE " + TREE_VIEW_DATABASE);
+ statement.execute(
+ "CREATE VIEW table1(device_id STRING TAG, s1 INT32 FIELD) AS
root.db_topk_rf.**");
+ }
+ }
+
+ private static void createTableModelDatabase() throws Exception {
+ prepareTableData(
+ new String[] {
+ "CREATE DATABASE " + TABLE_DATABASE,
+ "USE " + TABLE_DATABASE,
+ "CREATE TABLE table1(device_id STRING TAG, s1 INT32 FIELD)",
+ "INSERT INTO table1(time, device_id, s1) VALUES (10, 'd1', 1), (20,
'd1', 11), (30, 'd1', 111)",
+ "INSERT INTO table1(time, device_id, s1) VALUES (11, 'd2', 2), (21,
'd2', 22), (31, 'd2', 222)",
+ "INSERT INTO table1(time, device_id, s1) VALUES (12, 'd3', 3), (22,
'd3', 33), (32, 'd3', 333)",
+ });
+ }
+
+ private static void createReadTsFileDatabase() throws Exception {
+ try (Connection connection = EnvFactory.getEnv().getTableConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("CREATE DATABASE " + READ_TSFILE_DATABASE);
+ }
+ }
+
+ private static void generateTableWithDistinctDeviceTimes(
+ TsFileWriter writer,
+ String tableName,
+ List<String> tagColumns,
+ List<String> fieldColumns,
+ int deviceStart,
+ int deviceEnd)
+ throws IOException, WriteProcessException {
+ List<String> columnNames = new ArrayList<>(tagColumns.size() +
fieldColumns.size());
+ List<TSDataType> columnTypes = new ArrayList<>(tagColumns.size() +
fieldColumns.size());
+ List<ColumnCategory> columnCategories =
+ new ArrayList<>(tagColumns.size() + fieldColumns.size());
+ for (String tagColumn : tagColumns) {
+ columnNames.add(tagColumn);
+ columnTypes.add(TSDataType.STRING);
+ columnCategories.add(ColumnCategory.TAG);
+ }
+ for (String fieldColumn : fieldColumns) {
+ columnNames.add(fieldColumn);
+ columnTypes.add(TSDataType.INT32);
+ columnCategories.add(ColumnCategory.FIELD);
+ }
+
+ writer.registerTableSchema(
+ new TableSchema(tableName, columnNames, columnTypes,
columnCategories));
+ Tablet tablet = new Tablet(tableName, columnNames, columnTypes,
columnCategories);
+ for (int deviceIndex = deviceStart; deviceIndex <= deviceEnd;
deviceIndex++) {
+ int time = deviceIndex;
+ int row = tablet.getRowSize();
+ tablet.addTimestamp(row, time);
+ for (int i = 0; i < tagColumns.size(); i++) {
+ tablet.addValue(row, i, tagColumns.get(i) + "_" + deviceIndex);
+ }
+ for (int i = 0; i < fieldColumns.size(); i++) {
+ tablet.addValue(row, tagColumns.size() + i, time);
+ }
+ if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
+ writer.writeTable(tablet);
+ tablet.reset();
+ }
+ }
+ if (tablet.getRowSize() != 0) {
+ writer.writeTable(tablet);
+ }
+ }
+
+ private static void generateTable(
+ TsFileWriter writer,
+ String tableName,
+ List<String> tagColumns,
+ List<String> fieldColumns,
+ int deviceStart,
+ int deviceEnd)
+ throws IOException, WriteProcessException {
+ List<String> columnNames = new ArrayList<>(tagColumns.size() +
fieldColumns.size());
+ List<TSDataType> columnTypes = new ArrayList<>(tagColumns.size() +
fieldColumns.size());
+ List<ColumnCategory> columnCategories =
+ new ArrayList<>(tagColumns.size() + fieldColumns.size());
+ for (String tagColumn : tagColumns) {
+ columnNames.add(tagColumn);
+ columnTypes.add(TSDataType.STRING);
+ columnCategories.add(ColumnCategory.TAG);
+ }
+ for (String fieldColumn : fieldColumns) {
+ columnNames.add(fieldColumn);
+ columnTypes.add(TSDataType.INT32);
+ columnCategories.add(ColumnCategory.FIELD);
+ }
+
+ writer.registerTableSchema(
+ new TableSchema(tableName, columnNames, columnTypes,
columnCategories));
+ Tablet tablet = new Tablet(tableName, columnNames, columnTypes,
columnCategories);
+ for (int deviceIndex = deviceStart; deviceIndex <= deviceEnd;
deviceIndex++) {
+ for (int time = 1; time <= 3; time++) {
+ int row = tablet.getRowSize();
+ tablet.addTimestamp(row, time);
+ for (int i = 0; i < tagColumns.size(); i++) {
+ tablet.addValue(row, i, tagColumns.get(i) + "_" + deviceIndex);
+ }
+ for (int i = 0; i < fieldColumns.size(); i++) {
+ tablet.addValue(row, tagColumns.size() + i, time);
+ }
+ if (tablet.getRowSize() == tablet.getMaxRowNumber()) {
+ writer.writeTable(tablet);
+ tablet.reset();
+ }
+ }
+ }
+ if (tablet.getRowSize() != 0) {
+ writer.writeTable(tablet);
+ }
+ }
+
+ private static String toSqlPath(File file) {
+ return file.getAbsolutePath().replace("\\", "\\\\").replace("'", "''");
+ }
+
+ private static void deleteDirectory(File directory) {
+ File[] files = directory.listFiles();
+ if (files != null) {
+ for (File file : files) {
+ if (file.isDirectory()) {
+ deleteDirectory(file);
+ } else {
+ file.delete();
+ }
+ }
+ }
+ directory.delete();
+ }
+}
diff --git
a/iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java
b/iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java
index 05b1686e548..9e22e34ceb9 100644
---
a/iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java
+++
b/iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java
@@ -405,6 +405,8 @@ public final class CalcMessages {
public static final String
EXCEPTION_THE_SIZE_OF_READYQUEUE_CANNOT_BE_NEGATIVE_DOT_01D8D0CB = "The size of
readyQueue cannot be negative.";
public static final String
EXCEPTION_THE_SIZE_BETWEEN_WHENTRANSFORMERS_AND_THENTRANSFORMERS_NEEDS_TO_BE_SAME_AC796883
= "the size between whenTransformers and thenTransformers needs to be same";
public static final String
EXCEPTION_EXCEED_MAX_CALL_TIMES_OF_GETCOLUMN_69C77C7E = "Exceed max call times
of getColumn";
+ public static final String
EXCEPTION_TOPK_RUNTIME_FILTER_REQUIRES_SINGLE_TIME_ORDER_BY_7EED6208 =
+ "TopK runtime filter requires ORDER BY time on a single INT64 or
TIMESTAMP column";
public static final String
EXCEPTION_FILTER_IS_NOT_SUPPORTED_IN_ARG_DOT_FILTER_IS_ARG_DOT_417C4F3C =
"Filter is not supported in %s. Filter is %s.";
public static final String EXCEPTION_NULL_9B41EF67 = "null";
public static final String EXCEPTION_ARG_MUST_HAVE_JOIN_KEYS_DOT_C24DAB2D =
"%s must have join keys.";
diff --git
a/iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java
b/iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java
index 20c689943a6..817d8376e60 100644
---
a/iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java
+++
b/iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java
@@ -382,6 +382,8 @@ public final class CalcMessages {
public static final String
EXCEPTION_THE_SIZE_OF_READYQUEUE_CANNOT_BE_NEGATIVE_DOT_01D8D0CB = "readyQueue
的大小不能为负数。";
public static final String
EXCEPTION_THE_SIZE_BETWEEN_WHENTRANSFORMERS_AND_THENTRANSFORMERS_NEEDS_TO_BE_SAME_AC796883
= "whenTransformers 和 thenTransformers 的大小应相同";
public static final String
EXCEPTION_EXCEED_MAX_CALL_TIMES_OF_GETCOLUMN_69C77C7E = "超过 getColumn 的最大调用次数";
+ public static final String
EXCEPTION_TOPK_RUNTIME_FILTER_REQUIRES_SINGLE_TIME_ORDER_BY_7EED6208 =
+ "TopK Runtime Filter 要求 ORDER BY 单个 INT64 或 TIMESTAMP 类型的 time 列";
public static final String
EXCEPTION_FILTER_IS_NOT_SUPPORTED_IN_ARG_DOT_FILTER_IS_ARG_DOT_417C4F3C = "%s
不支持过滤条件。过滤条件为 %s。";
public static final String EXCEPTION_NULL_9B41EF67 = "null";
public static final String EXCEPTION_ARG_MUST_HAVE_JOIN_KEYS_DOT_C24DAB2D =
"%s 必须包含连接键。";
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..872860d75d1 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,9 +19,11 @@
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;
+import org.apache.iotdb.calc.utils.datastructure.MergeSortKey;
import org.apache.iotdb.calc.utils.datastructure.SortKey;
import org.apache.tsfile.block.column.Column;
@@ -33,6 +35,29 @@ import java.util.Comparator;
import java.util.List;
public class TableTopKOperator extends TopKOperator {
+
+ private final int timeFilterColumnIndex;
+
+ public TableTopKOperator(
+ CommonOperatorContext operatorContext,
+ List<Operator> childrenOperators,
+ List<TSDataType> dataTypes,
+ Comparator<SortKey> comparator,
+ int topValue,
+ boolean childrenDataInOrder,
+ TopKRuntimeFilter topKRuntimeFilter,
+ int timeFilterColumnIndex) {
+ super(
+ operatorContext,
+ childrenOperators,
+ dataTypes,
+ comparator,
+ topValue,
+ childrenDataInOrder,
+ topKRuntimeFilter);
+ this.timeFilterColumnIndex = timeFilterColumnIndex;
+ }
+
public TableTopKOperator(
CommonOperatorContext operatorContext,
List<Operator> childrenOperators,
@@ -40,7 +65,20 @@ public class TableTopKOperator extends TopKOperator {
Comparator<SortKey> comparator,
int topValue,
boolean childrenDataInOrder) {
- super(operatorContext, childrenOperators, dataTypes, comparator, topValue,
childrenDataInOrder);
+ super(
+ operatorContext,
+ childrenOperators,
+ dataTypes,
+ comparator,
+ topValue,
+ childrenDataInOrder,
+ null);
+ this.timeFilterColumnIndex = -1;
+ }
+
+ @Override
+ protected long extractThresholdTime(MergeSortKey peek) {
+ return
peek.tsBlock.getColumn(timeFilterColumnIndex).getLong(peek.rowIndex);
}
@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..b2754cbbdd6 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) {
@@ -226,6 +249,19 @@ public abstract class TopKOperator implements
ProcessOperator {
return null;
}
+ private void updateTopKRuntimeFilter() {
+ if (topKRuntimeFilter == null || mergeSortHeap.getHeapSize() < topValue) {
+ return;
+ }
+ MergeSortKey peek = mergeSortHeap.peek();
+ topKRuntimeFilter.updateThreshold(extractThresholdTime(peek));
+ }
+
+ /** Table model overrides to read the time field. */
+ protected long extractThresholdTime(MergeSortKey peek) {
+ return peek.tsBlock.getTimeByIndex(peek.rowIndex);
+ }
+
@Override
public void close() throws Exception {
for (int i = childIndex; i < childrenOperators.size(); i++) {
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..a2782ac0798 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;
@@ -888,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));
@@ -903,6 +905,25 @@ public abstract class TableOperatorGenerator<
sortItemIndexList,
sortItemDataTypeList,
context.getTableTypeProvider());
+ if (topKRuntimeFilter != null) {
+ checkArgument(
+ sortItemsCount == 1,
+
CalcMessages.EXCEPTION_TOPK_RUNTIME_FILTER_REQUIRES_SINGLE_TIME_ORDER_BY_7EED6208);
+ TSDataType sortType = sortItemDataTypeList.get(0);
+ checkArgument(
+ sortType == TSDataType.INT64 || sortType == TSDataType.TIMESTAMP,
+
CalcMessages.EXCEPTION_TOPK_RUNTIME_FILTER_REQUIRES_SINGLE_TIME_ORDER_BY_7EED6208);
+ return new TableTopKOperator(
+ operatorContext,
+ children,
+ dataTypes,
+ getComparatorForTable(
+ node.getOrderingScheme().getOrderingList(), sortItemIndexList,
sortItemDataTypeList),
+ (int) node.getCount(),
+ node.isChildrenDataInOrder(),
+ topKRuntimeFilter,
+ sortItemIndexList.get(0));
+ }
return new TableTopKOperator(
operatorContext,
children,
@@ -913,6 +934,10 @@ public abstract class TableOperatorGenerator<
node.isChildrenDataInOrder());
}
+ protected TopKRuntimeFilter registerRuntimeFilter(C context, TopKNode node) {
+ return null;
+ }
+
protected List<TSDataType> getOutputColumnTypes(PlanNode node,
ITableTypeProvider typeProvider) {
return node.getOutputSymbols().stream()
.map(s -> getTSDataType(typeProvider.getTableModelType(s)))
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/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/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..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
@@ -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();
@@ -205,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;
@@ -276,7 +281,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 +344,27 @@ 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 boolean skipByTopKRuntimeFilter(
+ Statistics<? extends Serializable> statistics, Runnable skip) {
+ if (!mayQualifyRuntimeFilterRange(statistics)) {
+ skip.run();
+ return true;
+ }
+ return false;
+ }
+
@SuppressWarnings("squid:S3740")
public Statistics currentFileTimeStatistics() {
return firstTimeSeriesMetadata.getTimeStatistics();
@@ -369,7 +395,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 +445,10 @@ public class SeriesScanUtil implements Accountable {
return;
}
+ if (skipByTopKRuntimeFilter(firstChunkMetadata.getStatistics(),
this::skipCurrentChunk)) {
+ return;
+ }
+
// globalTimeFilter.canSkip() must be FALSE
Filter pushDownFilter = scanOptions.getPushDownFilter();
if (pushDownFilter != null && pushDownFilter.canSkip(firstChunkMetadata)) {
@@ -542,7 +572,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 +949,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 +971,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 +1465,10 @@ public class SeriesScanUtil implements Accountable {
return;
}
+ if (skipByTopKRuntimeFilter(firstPageReader.getStatistics(),
this::skipCurrentPage)) {
+ return;
+ }
+
IPageReader pageReader = firstPageReader.getPageReader();
// globalTimeFilter.canSkip() must be FALSE
@@ -1892,6 +1955,10 @@ public class SeriesScanUtil implements Accountable {
return;
}
+ if (skipByTopKRuntimeFilter(firstTimeSeriesMetadata.getStatistics(),
this::skipCurrentFile)) {
+ return;
+ }
+
// globalTimeFilter.canSkip() must be FALSE
Filter pushDownFilter = scanOptions.getPushDownFilter();
if (pushDownFilter != null &&
pushDownFilter.canSkip(firstTimeSeriesMetadata)) {
@@ -2393,10 +2460,23 @@ 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)) {
+ curSeqFileIndex--;
+ continue;
+ }
if (dataSource.isSeqSatisfied(
deviceID, curSeqFileIndex, scanOptions.getGlobalTimeFilter(),
false)) {
- break;
+ if (filter == null
+ || dataSource.isSeqSatisfiedByRuntimeFilter(curSeqFileIndex,
filter, false)) {
+ break;
+ }
+ dataSource.setSeqTsFileResourceInvalidated(curSeqFileIndex);
+ if (!dataSource.hasValidResource()) {
+ runtimeFilterExhausted = true;
+ return false;
+ }
}
curSeqFileIndex--;
}
@@ -2405,10 +2485,23 @@ 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)) {
+ curUnseqFileIndex++;
+ continue;
+ }
if (dataSource.isUnSeqSatisfied(
deviceID, curUnseqFileIndex, scanOptions.getGlobalTimeFilter(),
false)) {
- break;
+ if (filter == null
+ || dataSource.isUnSeqSatisfiedByRuntimeFilter(curUnseqFileIndex,
filter)) {
+ break;
+ }
+ dataSource.setUnseqTsFileResourceInvalidated(curUnseqFileIndex);
+ if (!dataSource.hasValidResource()) {
+ runtimeFilterExhausted = true;
+ return false;
+ }
}
curUnseqFileIndex++;
}
@@ -2522,10 +2615,23 @@ 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)) {
+ curSeqFileIndex++;
+ continue;
+ }
if (dataSource.isSeqSatisfied(
deviceID, curSeqFileIndex, scanOptions.getGlobalTimeFilter(),
false)) {
- break;
+ if (filter == null
+ || dataSource.isSeqSatisfiedByRuntimeFilter(curSeqFileIndex,
filter, false)) {
+ break;
+ }
+ dataSource.setSeqTsFileResourceInvalidated(curSeqFileIndex);
+ if (!dataSource.hasValidResource()) {
+ runtimeFilterExhausted = true;
+ return false;
+ }
}
curSeqFileIndex++;
}
@@ -2534,10 +2640,23 @@ 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)) {
+ curUnseqFileIndex++;
+ continue;
+ }
if (dataSource.isUnSeqSatisfied(
deviceID, curUnseqFileIndex, scanOptions.getGlobalTimeFilter(),
false)) {
- break;
+ if (filter == null
+ || dataSource.isUnSeqSatisfiedByRuntimeFilter(curUnseqFileIndex,
filter)) {
+ break;
+ }
+ dataSource.setUnseqTsFileResourceInvalidated(curUnseqFileIndex);
+ if (!dataSource.hasValidResource()) {
+ runtimeFilterExhausted = true;
+ return false;
+ }
}
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..6e9794dcf82 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
@@ -82,7 +82,7 @@ public abstract class AbstractTableScanOperator extends
AbstractSeriesScanOperat
private TsBlock measurementDataBlock;
- private QueryDataSource queryDataSource;
+ protected QueryDataSource queryDataSource;
protected int currentDeviceIndex;
@@ -217,8 +217,16 @@ 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;
+ }
+ return shouldStopScanByRuntimeFilter();
}
@Override
@@ -251,6 +259,10 @@ public abstract class AbstractTableScanOperator extends
AbstractSeriesScanOperat
}
protected void moveToNextDevice() {
+ if (shouldStopScanByRuntimeFilter()) {
+ currentDeviceIndex = deviceCount;
+ return;
+ }
currentDeviceIndex++;
if (currentDeviceIndex < deviceCount) {
// construct AlignedSeriesScanUtil for next device
@@ -264,6 +276,11 @@ public abstract class AbstractTableScanOperator extends
AbstractSeriesScanOperat
}
}
+ /** Returns true when file-level RF has pruned all seq/unseq files — scan
can stop globally. */
+ protected boolean shouldStopScanByRuntimeFilter() {
+ return seriesScanOptions.getTopKRuntimeFilter() != null &&
!queryDataSource.hasValidResource();
+ }
+
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/execution/operator/source/relational/ExternalTsFileTableScanOperator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileTableScanOperator.java
index b11e38f9008..a38fcacd046 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileTableScanOperator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileTableScanOperator.java
@@ -97,12 +97,23 @@ public class ExternalTsFileTableScanOperator extends
TableScanOperator {
super.initQueryDataSource(currentDataSource);
}
+ @Override
+ protected boolean shouldStopScanByRuntimeFilter() {
+ // Each device uses its own QueryDataSource; exhausting files for a
non-last device must not
+ // stop scanning remaining devices in the same external TsFile task.
+ if (currentDeviceIndex < deviceCount - 1) {
+ return false;
+ }
+ return super.shouldStopScanByRuntimeFilter();
+ }
+
@Override
protected void moveToNextDevice() {
currentDeviceIndex++;
if (currentDeviceIndex < deviceCount) {
constructAlignedSeriesScanUtil();
- seriesScanUtil.initQueryDataSource(updateCurrentDeviceQueryDataSource());
+ queryDataSource = updateCurrentDeviceQueryDataSource();
+ seriesScanUtil.initQueryDataSource(queryDataSource);
this.operatorContext.recordSpecifiedInfo(
CommonOperatorUtils.CURRENT_DEVICE_INDEX_STRING,
Integer.toString(currentDeviceIndex));
}
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..e09d05066a3 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,6 +19,7 @@
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;
@@ -46,6 +47,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;
@@ -222,6 +224,53 @@ 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.getTopKRuntimeFilterSourceId() == null) {
+ return null;
+ }
+ return context.dataNodeQueryContext.registerTopKRuntimeFilter(
+ node.getTopKRuntimeFilterSourceId(),
+ 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.getTopKRuntimeFilterSourceId() == null) {
+ return null;
+ }
+ return context.dataNodeQueryContext.getTopKRuntimeFilter(
+ scanNode.getTopKRuntimeFilterSourceId());
+ }
+
+ @Override
+ protected TopKRuntimeFilter registerRuntimeFilter(
+ 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
protected String getSortTmpDir(CommonOperatorContext operatorContext) {
OperatorContext dataNodeOperatorContext = (OperatorContext)
operatorContext;
@@ -626,6 +675,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);
@@ -973,13 +1023,15 @@ public class DataNodeTableOperatorGenerator
IDeviceID.TreeDeviceIdColumnValueExtractor idColumnValueExtractor =
createTreeDeviceIdColumnValueExtractor(DataNodeTreeViewSchemaUtils.getPrefixPath(tsTable));
+ TopKRuntimeFilter topKRuntimeFilter =
resolveTopKRuntimeFilterForDeviceScan(node, context);
AbstractTableScanOperator.AbstractTableScanOperatorParameter parameter =
constructAbstractTableScanOperatorParameter(
node,
context,
TreeAlignedDeviceViewScanOperator.class.getSimpleName(),
node.getMeasurementColumnNameMap(),
- tsTable.getCachedTableTTL());
+ tsTable.getCachedTableTTL(),
+ topKRuntimeFilter);
TreeAlignedDeviceViewScanOperator treeAlignedDeviceViewScanOperator =
new TreeAlignedDeviceViewScanOperator(parameter,
idColumnValueExtractor);
@@ -1091,16 +1143,78 @@ 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) {
+ 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 +1233,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);
@@ -1148,13 +1263,15 @@ public class DataNodeTableOperatorGenerator
return new EmptyDataOperator(operatorContext);
}
+ TopKRuntimeFilter topKRuntimeFilter =
resolveTopKRuntimeFilterForDeviceScan(node, context);
AbstractTableScanOperator.AbstractTableScanOperatorParameter parameter =
constructAbstractTableScanOperatorParameter(
node,
context,
ExternalTsFileTableScanOperator.class.getSimpleName(),
Collections.emptyMap(),
- Long.MAX_VALUE);
+ Long.MAX_VALUE,
+ topKRuntimeFilter);
AbstractTableScanOperator externalTsFileTableScanOperator =
new ExternalTsFileTableScanOperator(parameter,
node.getDeviceTaskPartitionIndex());
@@ -1932,6 +2049,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 +2095,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/plan/node/PlanGraphPrinter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
index 06aba49200b..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,6 +698,10 @@ public class PlanGraphPrinter implements
PlanVisitor<List<String>, PlanGraphPrin
boxValue.add(
String.format(
"PushDownLimitToEachDevice: %s",
deviceTableScanNode.isPushLimitToEachDevice()));
+ String topKRuntimeFilterSourceId =
deviceTableScanNode.getTopKRuntimeFilterSourceId();
+ if (topKRuntimeFilterSourceId != null) {
+ boxValue.add(String.format("TOPN OPT: %s", topKRuntimeFilterSourceId));
+ }
}
boxValue.add(
@@ -1049,6 +1053,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.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/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/TableDistributedPlanGenerator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
index f8c5f193e1a..c731bee31cc 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
@@ -435,12 +435,14 @@ public class TableDistributedPlanGenerator
}
TopKNode newTopKNode = (TopKNode) node.clone();
+ // root TopK is not a runtime filter producer.
+ newTopKNode.setTopKRuntimeFilterSourceId(null);
for (PlanNode child : childrenNodes) {
PlanNode newChild;
if (canTopKEliminated(node.getOrderingScheme(), node.getCount(), child))
{
newChild = child;
} else {
- newChild =
+ TopKNode regionTopK =
new TopKNode(
queryId.genPlanNodeId(),
Collections.singletonList(child),
@@ -448,6 +450,8 @@ public class TableDistributedPlanGenerator
node.getCount(),
node.getOutputSymbols(),
node.isChildrenDataInOrder());
+
regionTopK.setTopKRuntimeFilterSourceId(node.getTopKRuntimeFilterSourceId());
+ newChild = regionTopK;
}
newTopKNode.addChild(newChild);
}
@@ -779,6 +783,7 @@ public class TableDistributedPlanGenerator
partition.getPartitionIndex(),
node.getSchemaFilter());
splitNode.setRegionReplicaSet(localRegionReplicaSet);
+
splitNode.setTopKRuntimeFilterSourceId(node.getTopKRuntimeFilterSourceId());
result.add(splitNode);
}
sortPropertyContext.ifPresent(
@@ -914,6 +919,7 @@ public class TableDistributedPlanGenerator
node.isPushLimitToEachDevice(),
node.containsNonAlignedDevice());
scanNode.setRegionReplicaSet(regionReplicaSets.get(0));
+
scanNode.setTopKRuntimeFilterSourceId(node.getTopKRuntimeFilterSourceId());
return scanNode;
});
deviceTableScanNode.appendDeviceEntry(deviceEntry);
@@ -1000,6 +1006,7 @@ public class TableDistributedPlanGenerator
node.isPushLimitToEachDevice(),
node.containsNonAlignedDevice());
scanNode.setRegionReplicaSet(regionReplicaSet);
+
scanNode.setTopKRuntimeFilterSourceId(node.getTopKRuntimeFilterSourceId());
return scanNode;
});
deviceTableScanNode.appendDeviceEntry(deviceEntry);
@@ -1094,6 +1101,7 @@ public class TableDistributedPlanGenerator
node.getTreeDBName(),
node.getMeasurementColumnNameMap());
scanNode.setRegionReplicaSet(regionReplicaSet);
+
scanNode.setTopKRuntimeFilterSourceId(node.getTopKRuntimeFilterSourceId());
pair.left = scanNode;
}
@@ -1116,6 +1124,7 @@ public class TableDistributedPlanGenerator
node.getTreeDBName(),
node.getMeasurementColumnNameMap());
scanNode.setRegionReplicaSet(regionReplicaSet);
+
scanNode.setTopKRuntimeFilterSourceId(node.getTopKRuntimeFilterSourceId());
pair.right = scanNode;
}
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..867fe4ecf67 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
@@ -176,7 +176,10 @@ 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);
+
+ 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..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
@@ -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 String 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,8 @@ public class DeviceTableScanNode extends TableScanNode {
}
ReadWriteIOUtils.write(node.pushLimitToEachDevice, byteBuffer);
+
+ ReadWriteIOUtils.write(node.topKRuntimeFilterSourceId, byteBuffer);
}
protected static void serializeMemberVariables(
@@ -198,6 +206,8 @@ public class DeviceTableScanNode extends TableScanNode {
}
ReadWriteIOUtils.write(node.pushLimitToEachDevice, stream);
+
+ ReadWriteIOUtils.write(node.topKRuntimeFilterSourceId, stream);
}
protected static void deserializeMemberVariables(
@@ -227,6 +237,8 @@ public class DeviceTableScanNode extends TableScanNode {
}
node.pushLimitToEachDevice = ReadWriteIOUtils.readBool(byteBuffer);
+
+ node.topKRuntimeFilterSourceId = ReadWriteIOUtils.readString(byteBuffer);
}
@Override
@@ -307,6 +319,15 @@ public class DeviceTableScanNode extends TableScanNode {
this.containsNonAlignedDevice = true;
}
+ @Nullable
+ public String getTopKRuntimeFilterSourceId() {
+ return topKRuntimeFilterSourceId;
+ }
+
+ public void setTopKRuntimeFilterSourceId(@Nullable String
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/LogicalOptimizeFactory.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java
index afe68803948..78bb5f984a1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/LogicalOptimizeFactory.java
@@ -390,6 +390,7 @@ public class LogicalOptimizeFactory {
new MergeLimitWithSort(),
new MergeLimitOverProjectWithSort(),
new PushTopKThroughUnion())),
+ new TopKRuntimeFilterOptimizer(),
new ParallelizeGrouping());
this.planOptimizers = optimizerBuilder.build();
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..137af685e06
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterOptimizer.java
@@ -0,0 +1,83 @@
+/*
+ * 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.node.TopKNode;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+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> Logical plan planning (after {@code
PushTopKThroughUnion}).
+ *
+ * <p>Marks the {@code TopK + DeviceTableScan} pattern for TopK runtime
filter. A qualifying TopK
+ * uses its own plan node id as {@code topKRuntimeFilterSourceId}; the same id
is stamped on direct
+ * raw {@link DeviceTableScanNode} children. Distributed planning later copies
this id onto
+ * per-region TopK nodes and clears it on the root TopK.
+ */
+public class TopKRuntimeFilterOptimizer implements PlanOptimizer {
+
+ @Override
+ public PlanNode optimize(PlanNode plan, Context context) {
+ if
(!IoTDBDescriptor.getInstance().getConfig().isEnableTopKRuntimeFilter()) {
+ return plan;
+ }
+ return plan.accept(new Rewriter(), null);
+ }
+
+ private static class Rewriter implements PlanVisitor<PlanNode, Void> {
+
+ @Override
+ public PlanNode visitPlan(PlanNode node, Void unused) {
+ for (PlanNode child : node.getChildren()) {
+ child.accept(this, null);
+ }
+ return node;
+ }
+
+ @Override
+ public PlanNode visitTopK(TopKNode node, Void unused) {
+ String topKId = node.getPlanNodeId().getId();
+ for (PlanNode child : node.getChildren()) {
+ boolean isRawDeviceTableScan =
+ child instanceof DeviceTableScanNode && !(child instanceof
AggregationTableScanNode);
+ if (isRawDeviceTableScan
+ && isOrderByTimeOnly(node.getOrderingScheme(),
(DeviceTableScanNode) child)) {
+ node.setTopKRuntimeFilterSourceId(topKId);
+ ((DeviceTableScanNode) child).setTopKRuntimeFilterSourceId(topKId);
+ } else {
+ child.accept(this, null);
+ }
+ }
+ return node;
+ }
+ }
+
+ private static boolean isOrderByTimeOnly(
+ OrderingScheme orderingScheme, DeviceTableScanNode scanNode) {
+ if (orderingScheme.getOrderBy().size() != 1) {
+ return false;
+ }
+ return scanNode.isTimeColumn(orderingScheme.getOrderBy().get(0));
+ }
+}
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..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
@@ -19,11 +19,14 @@
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;
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.Comparator;
@@ -70,6 +73,22 @@ public class QueryDataSource implements IQueryDataSource {
private String databaseName = null;
+ /**
+ * 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 BitMap seqInvalidatedByRuntimeFilter;
+
+ private BitMap unseqInvalidatedByRuntimeFilter;
+
+ /**
+ * 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) {
@@ -91,6 +110,79 @@ public class QueryDataSource implements IQueryDataSource {
this.unseqResources = other.unseqResources;
this.unSeqFileOrderIndex = other.unSeqFileOrderIndex;
this.databaseName = other.databaseName;
+ if (other.runtimeFilterTrackingEnabled) {
+ this.runtimeFilterTrackingEnabled = true;
+ this.validSize = other.validSize;
+ this.seqInvalidatedByRuntimeFilter =
other.seqInvalidatedByRuntimeFilter.clone();
+ this.unseqInvalidatedByRuntimeFilter =
other.unseqInvalidatedByRuntimeFilter.clone();
+ }
+ }
+
+ /** 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;
+ }
+
+ /** 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.
+ *
+ * <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 (!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) {
+ 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.isMarked(index);
+ }
+ return
unseqInvalidatedByRuntimeFilter.isMarked(unSeqFileOrderIndex[index]);
}
public List<TsFileResource> getSeqResources() {
@@ -144,6 +236,11 @@ public class QueryDataSource implements IQueryDataSource {
return curSeqSatisfied;
}
+ public boolean isSeqSatisfiedByRuntimeFilter(
+ 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 +293,10 @@ public class QueryDataSource implements IQueryDataSource {
return curUnSeqSatisfied;
}
+ public boolean isUnSeqSatisfiedByRuntimeFilter(int curIndex,
TopKRuntimeFilter filter) {
+ return isResourceSatisfiedByRuntimeFilter(curIndex, filter, false, false);
+ }
+
public long getCurrentUnSeqOrderTime(int curIndex) {
if (curIndex != this.curUnSeqIndex) {
throw new IllegalArgumentException(
@@ -265,6 +366,19 @@ 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]);
+ // 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/execution/operator/source/relational/ExternalTsFileTableScanOperatorRuntimeFilterTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileTableScanOperatorRuntimeFilterTest.java
new file mode 100644
index 00000000000..b44e3acfab0
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/ExternalTsFileTableScanOperatorRuntimeFilterTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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.execution.operator.source.relational;
+
+import org.apache.iotdb.calc.execution.filter.TopKRuntimeFilter;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import
org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import
org.apache.iotdb.db.queryengine.execution.operator.source.relational.AbstractTableScanOperator.AbstractTableScanOperatorParameter;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
+import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.common.type.TypeFactory;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+public class ExternalTsFileTableScanOperatorRuntimeFilterTest {
+
+ @Test
+ public void shouldNotStopWhenNonLastDeviceExhausted() throws Exception {
+ ExternalTsFileTableScanOperator operator =
+ new TestExternalTsFileTableScanOperator(createParameter());
+ QueryDataSource firstDeviceDataSource =
Mockito.mock(QueryDataSource.class);
+ Mockito.when(firstDeviceDataSource.hasValidResource()).thenReturn(false);
+ operator.queryDataSource = firstDeviceDataSource;
+ operator.currentDeviceIndex = 0;
+
+ Assert.assertFalse(operator.shouldStopScanByRuntimeFilter());
+ }
+
+ @Test
+ public void shouldUseCurrentDeviceQueryDataSourceOnLastDevice() throws
Exception {
+ ExternalTsFileTableScanOperator operator =
+ new TestExternalTsFileTableScanOperator(createParameter());
+ QueryDataSource firstDeviceDataSource =
Mockito.mock(QueryDataSource.class);
+ Mockito.when(firstDeviceDataSource.hasValidResource()).thenReturn(false);
+ QueryDataSource lastDeviceDataSource = Mockito.mock(QueryDataSource.class);
+ Mockito.when(lastDeviceDataSource.hasValidResource()).thenReturn(true);
+
+ operator.currentDeviceIndex = 1;
+ operator.queryDataSource = lastDeviceDataSource;
+
+ Assert.assertFalse(operator.shouldStopScanByRuntimeFilter());
+
+ Mockito.when(lastDeviceDataSource.hasValidResource()).thenReturn(false);
+ Assert.assertTrue(operator.shouldStopScanByRuntimeFilter());
+ }
+
+ private static AbstractTableScanOperatorParameter createParameter() {
+ OperatorContext operatorContext = Mockito.mock(OperatorContext.class);
+ PlanNodeId sourceId = new PlanNodeId("external-scan");
+ List<ColumnSchema> columnSchemas =
+ Collections.singletonList(
+ new ColumnSchema(
+ "time", TypeFactory.getType(TSDataType.INT64), false,
TsTableColumnCategory.TIME));
+ int[] columnsIndexArray = new int[] {0};
+ List<DeviceEntry> deviceEntries = Arrays.asList(mockDeviceEntry(),
mockDeviceEntry());
+ SeriesScanOptions seriesScanOptions =
+ new SeriesScanOptions.Builder().withTopKRuntimeFilter(new
TopKRuntimeFilter(false)).build();
+ List<String> measurementColumnNames = Collections.emptyList();
+ List<IMeasurementSchema> measurementSchemas = Collections.emptyList();
+ Set<String> allSensors = new HashSet<>();
+ allSensors.add("");
+ return new AbstractTableScanOperatorParameter(
+ allSensors,
+ operatorContext,
+ sourceId,
+ columnSchemas,
+ columnsIndexArray,
+ deviceEntries,
+ Ordering.ASC,
+ seriesScanOptions,
+ measurementColumnNames,
+ measurementSchemas,
+ 1000);
+ }
+
+ private static DeviceEntry mockDeviceEntry() {
+ DeviceEntry deviceEntry = Mockito.mock(DeviceEntry.class);
+ IDeviceID deviceID = Mockito.mock(IDeviceID.class);
+ Mockito.when(deviceEntry.getDeviceID()).thenReturn(deviceID);
+ return deviceEntry;
+ }
+
+ private static final class TestExternalTsFileTableScanOperator
+ extends ExternalTsFileTableScanOperator {
+
+ private
TestExternalTsFileTableScanOperator(AbstractTableScanOperatorParameter
parameter) {
+ super(parameter, 0);
+ }
+
+ @Override
+ protected void constructAlignedSeriesScanUtil() {
+ // Skip TsFile path construction in unit test.
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java
index 01acba6b5d9..6d5102d408a 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java
@@ -664,6 +664,31 @@ public final class PlanMatchPattern {
return node(TopKNode.class, source).with(new TopKMatcher(orderBy, count,
childrenDataInOrder));
}
+ public static PlanMatchPattern topKWithRuntimeFilterSourceId(
+ String sourceId, PlanMatchPattern... sources) {
+ return topK(sources)
+ .with(
+ TopKNode.class,
+ node -> java.util.Objects.equals(sourceId,
node.getTopKRuntimeFilterSourceId()));
+ }
+
+ public static PlanMatchPattern topKWithoutRuntimeFilter(PlanMatchPattern...
sources) {
+ return topK(sources).with(TopKNode.class, node ->
node.getTopKRuntimeFilterSourceId() == null);
+ }
+
+ public static PlanMatchPattern tableScanWithRuntimeFilter(String
expectedTableName) {
+ return tableScan(expectedTableName)
+ .with(DeviceTableScanNode.class, node ->
node.getTopKRuntimeFilterSourceId() != null);
+ }
+
+ public static PlanMatchPattern tableScanWithRuntimeFilter(
+ String expectedTableName, String sourceId) {
+ return tableScan(expectedTableName)
+ .with(
+ DeviceTableScanNode.class,
+ node -> java.util.Objects.equals(sourceId,
node.getTopKRuntimeFilterSourceId()));
+ }
+
/*public static PlanMatchPattern topN(long count, List<Ordering> orderBy,
TopNNode.Step step, PlanMatchPattern source)
{
return node(TopNNode.class, source).with(new TopNMatcher(count, orderBy,
step));
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterDistributedPlanTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterDistributedPlanTest.java
new file mode 100644
index 00000000000..27a1262a539
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterDistributedPlanTest.java
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations;
+
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode;
+import
org.apache.iotdb.commons.queryengine.plan.relational.planner.node.TopKNode;
+import
org.apache.iotdb.commons.queryengine.plan.relational.planner.node.UnionNode;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
+import
org.apache.iotdb.db.queryengine.plan.relational.planner.PlanNodeSearcher;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanAssert.assertPlan;
+import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.anyTree;
+import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.exchange;
+import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.output;
+import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.tableScanWithRuntimeFilter;
+import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.topKWithRuntimeFilterSourceId;
+import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.topKWithoutRuntimeFilter;
+import static
org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.union;
+
+public class TopKRuntimeFilterDistributedPlanTest {
+
+ private static final String TABLE1 = "testdb.table1";
+ private static final String TABLE2 = "testdb.table2";
+
+ private boolean originalEnableTopKRuntimeFilter;
+
+ @Before
+ public void setUp() {
+ originalEnableTopKRuntimeFilter =
+ IoTDBDescriptor.getInstance().getConfig().isEnableTopKRuntimeFilter();
+ IoTDBDescriptor.getInstance().getConfig().setEnableTopKRuntimeFilter(true);
+ IoTDBDescriptor.getInstance().getConfig().setDataNodeId(1);
+ }
+
+ @After
+ public void tearDown() {
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setEnableTopKRuntimeFilter(originalEnableTopKRuntimeFilter);
+ }
+
+ @Test
+ public void singleRegionKeepsLogicalTopKSourceId() {
+ String sql = "SELECT time FROM table1 WHERE tag1 = 'shanghai' ORDER BY
time DESC LIMIT 10";
+ PlanTester planTester = new PlanTester();
+ LogicalQueryPlan logicalPlan = planTester.createPlan(sql);
+
+ TopKNode logicalTopK = findProducerTopK(logicalPlan.getRootNode());
+ String logicalSourceId = logicalTopK.getTopKRuntimeFilterSourceId();
+ Assert.assertEquals(logicalTopK.getPlanNodeId().getId(), logicalSourceId);
+
+ assertPlan(
+ logicalPlan,
+ output(
+ topKWithRuntimeFilterSourceId(
+ logicalSourceId, tableScanWithRuntimeFilter(TABLE1,
logicalSourceId))));
+
+ PlanNode distributedPlan = planTester.getFragmentPlan(0);
+
+ assertPlan(
+ distributedPlan,
+ output(
+ topKWithRuntimeFilterSourceId(
+ logicalSourceId, tableScanWithRuntimeFilter(TABLE1,
logicalSourceId))));
+ }
+
+ @Test
+ public void multiRegionCopiesSourceIdToRegionTopKAndClearsRoot() {
+ String sql = "SELECT time FROM table1 ORDER BY time DESC LIMIT 10";
+ PlanTester planTester = new PlanTester();
+ LogicalQueryPlan logicalPlan = planTester.createPlan(sql);
+
+ TopKNode logicalTopK = findProducerTopK(logicalPlan.getRootNode());
+ String logicalSourceId = logicalTopK.getTopKRuntimeFilterSourceId();
+ Assert.assertEquals(logicalTopK.getPlanNodeId().getId(), logicalSourceId);
+
+ assertPlan(
+ logicalPlan,
+ output(
+ topKWithRuntimeFilterSourceId(
+ logicalSourceId, tableScanWithRuntimeFilter(TABLE1,
logicalSourceId))));
+
+ PlanNode coordinatorFragment = planTester.getFragmentPlan(0);
+ assertPlan(
+ coordinatorFragment,
+ output(anyTree(topKWithoutRuntimeFilter(exchange(), exchange(),
exchange()))));
+ Assert.assertNull(findProducerTopKOrNull(coordinatorFragment));
+
+ for (int i = 1; i <= 3; i++) {
+ PlanNode regionFragment = planTester.getFragmentPlan(i);
+ assertPlan(
+ regionFragment,
+ topKWithRuntimeFilterSourceId(
+ logicalSourceId, tableScanWithRuntimeFilter(TABLE1,
logicalSourceId)));
+ }
+ }
+
+ @Test
+ public void unionBranchesKeepDistinctSourceIdsAfterDistribution() {
+ String sql =
+ "(SELECT time FROM table1) UNION ALL (SELECT time FROM table2) ORDER
BY time DESC LIMIT 10";
+ PlanTester planTester = new PlanTester();
+ LogicalQueryPlan logicalPlan = planTester.createPlan(sql);
+
+ UnionNode unionNode =
+ (UnionNode)
+ PlanNodeSearcher.searchFrom(logicalPlan.getRootNode())
+ .where(UnionNode.class::isInstance)
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ TopKNode leftBranchTopK = (TopKNode) unionNode.getChildren().get(0);
+ TopKNode rightBranchTopK = (TopKNode) unionNode.getChildren().get(1);
+ String leftSourceId = leftBranchTopK.getTopKRuntimeFilterSourceId();
+ String rightSourceId = rightBranchTopK.getTopKRuntimeFilterSourceId();
+ Assert.assertNotNull(leftSourceId);
+ Assert.assertNotNull(rightSourceId);
+ Assert.assertEquals(leftBranchTopK.getPlanNodeId().getId(), leftSourceId);
+ Assert.assertEquals(rightBranchTopK.getPlanNodeId().getId(),
rightSourceId);
+ Assert.assertNotEquals(leftSourceId, rightSourceId);
+
+ assertPlan(
+ logicalPlan,
+ output(
+ topKWithoutRuntimeFilter(
+ union(
+ topKWithRuntimeFilterSourceId(
+ leftSourceId, tableScanWithRuntimeFilter(TABLE1,
leftSourceId)),
+ topKWithRuntimeFilterSourceId(
+ rightSourceId, tableScanWithRuntimeFilter(TABLE2,
rightSourceId))))));
+
+ assertPlan(
+ planTester.getFragmentPlan(0),
+ output(topKWithoutRuntimeFilter(union(exchange(), exchange()))));
+
+ assertPlan(
+ planTester.getFragmentPlan(1),
+ topKWithoutRuntimeFilter(exchange(), exchange(), exchange()));
+ for (int i = 2; i <= 4; i++) {
+ assertPlan(
+ planTester.getFragmentPlan(i),
+ topKWithRuntimeFilterSourceId(
+ leftSourceId, tableScanWithRuntimeFilter(TABLE1, leftSourceId)));
+ }
+ assertPlan(
+ planTester.getFragmentPlan(5),
+ topKWithoutRuntimeFilter(exchange(), exchange(), exchange()));
+ for (int i = 6; i <= 8; i++) {
+ assertPlan(
+ planTester.getFragmentPlan(i),
+ topKWithRuntimeFilterSourceId(
+ rightSourceId, tableScanWithRuntimeFilter(TABLE2,
rightSourceId)));
+ }
+ }
+
+ private static TopKNode findProducerTopK(PlanNode root) {
+ return (TopKNode)
+ PlanNodeSearcher.searchFrom(root)
+ .where(
+ node ->
+ node instanceof TopKNode
+ && ((TopKNode) node).getTopKRuntimeFilterSourceId() !=
null)
+ .findFirst()
+ .orElseThrow(AssertionError::new);
+ }
+
+ private static TopKNode findProducerTopKOrNull(PlanNode root) {
+ return (TopKNode)
+ PlanNodeSearcher.searchFrom(root)
+ .where(
+ node ->
+ node instanceof TopKNode
+ && ((TopKNode) node).getTopKRuntimeFilterSourceId() !=
null)
+ .findFirst()
+ .orElse(null);
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterTreeViewPlanTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterTreeViewPlanTest.java
new file mode 100644
index 00000000000..91b16f17d5e
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/TopKRuntimeFilterTreeViewPlanTest.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.relational.planner.optimizations;
+
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode;
+import
org.apache.iotdb.commons.queryengine.plan.relational.planner.node.TopKNode;
+import org.apache.iotdb.commons.schema.table.TreeViewSchema;
+import org.apache.iotdb.commons.schema.table.TsTable;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester;
+import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
+import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeAlignedDeviceViewScanNode;
+import
org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeNonAlignedDeviceViewScanNode;
+import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import static
org.apache.iotdb.db.queryengine.plan.relational.analyzer.TestMetadata.DEVICE_VIEW_TEST_TABLE;
+import static
org.apache.iotdb.db.queryengine.plan.relational.analyzer.TestMetadata.TREE_VIEW_DB;
+
+public class TopKRuntimeFilterTreeViewPlanTest {
+
+ private static final String DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME =
+ String.format("%s.\"%s\"", TREE_VIEW_DB, DEVICE_VIEW_TEST_TABLE);
+
+ @Before
+ public void setup() {
+ TsTable tsTable = new TsTable(DEVICE_VIEW_TEST_TABLE);
+ tsTable.addProp(TsTable.TTL_PROPERTY, Long.MAX_VALUE + "");
+ tsTable.addProp(TreeViewSchema.TREE_PATH_PATTERN, "root.test.**");
+ DataNodeTableCache.getInstance().preUpdateTable(TREE_VIEW_DB, tsTable,
null);
+ DataNodeTableCache.getInstance().commitUpdateTable(TREE_VIEW_DB,
DEVICE_VIEW_TEST_TABLE, null);
+ }
+
+ @After
+ public void tearDown() {
+ DataNodeTableCache.getInstance().invalid(TREE_VIEW_DB);
+ }
+
+ @Test
+ public void marksRuntimeFilterOnTreeViewRegionTopKAndScan() {
+ PlanTester planTester = new PlanTester();
+ planTester.createPlan(
+ "SELECT time, s1 FROM "
+ + DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME
+ + " ORDER BY time DESC LIMIT 10");
+
+ boolean foundProducer = false;
+ boolean foundConsumer = false;
+ String rootTopKId = null;
+ for (int i = 1; i <= 4; i++) {
+ RuntimeFilterMark mark =
collectRuntimeFilterMark(planTester.getFragmentPlan(i));
+ if (mark.producerTopKId != null) {
+ foundProducer = true;
+ rootTopKId = mark.producerTopKId;
+ }
+ if (mark.scanSourceId != null) {
+ foundConsumer = true;
+ if (rootTopKId != null) {
+ Assert.assertEquals(rootTopKId, mark.scanSourceId);
+ }
+ }
+ }
+
+ Assert.assertTrue(foundProducer);
+ Assert.assertTrue(foundConsumer);
+ }
+
+ private static RuntimeFilterMark collectRuntimeFilterMark(PlanNode root) {
+ RuntimeFilterMark mark = new RuntimeFilterMark();
+ collectRuntimeFilterMark(root, mark);
+ return mark;
+ }
+
+ private static void collectRuntimeFilterMark(PlanNode node,
RuntimeFilterMark mark) {
+ if (node instanceof TopKNode) {
+ TopKNode topKNode = (TopKNode) node;
+ if (topKNode.getTopKRuntimeFilterSourceId() != null) {
+ mark.producerTopKId = topKNode.getTopKRuntimeFilterSourceId();
+ }
+ }
+ if (node instanceof DeviceTableScanNode) {
+ DeviceTableScanNode scanNode = (DeviceTableScanNode) node;
+ if (scanNode.getTopKRuntimeFilterSourceId() != null) {
+ mark.scanSourceId = scanNode.getTopKRuntimeFilterSourceId();
+ Assert.assertTrue(
+ scanNode instanceof TreeAlignedDeviceViewScanNode
+ || scanNode instanceof TreeNonAlignedDeviceViewScanNode);
+ }
+ }
+ for (PlanNode child : node.getChildren()) {
+ collectRuntimeFilterMark(child, mark);
+ }
+ }
+
+ private static final class RuntimeFilterMark {
+ private String producerTopKId;
+ private String scanSourceId;
+ }
+}
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 bfb98ebadc1..0150f9db6fa 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 34376dcdbd5..a28e39a74cc 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,6 +52,14 @@ public class TopKNode extends MultiChildProcessNode {
private final boolean childrenDataInOrder;
+ private final transient boolean topKRuntimeFilterAscending;
+
+ // 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 String topKRuntimeFilterSourceId;
+
public TopKNode(
PlanNodeId id,
OrderingScheme scheme,
@@ -61,6 +71,7 @@ public class TopKNode extends MultiChildProcessNode {
this.count = count;
this.outputSymbols = outputSymbols;
this.childrenDataInOrder = childrenDataInOrder;
+ this.topKRuntimeFilterAscending =
computeTopKRuntimeFilterAscending(scheme);
}
public TopKNode(
@@ -75,11 +86,20 @@ 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() {
- return new TopKNode(getPlanNodeId(), orderingScheme, count, outputSymbols,
childrenDataInOrder);
+ TopKNode cloned =
+ new TopKNode(getPlanNodeId(), orderingScheme, count, outputSymbols,
childrenDataInOrder);
+ cloned.topKRuntimeFilterSourceId = topKRuntimeFilterSourceId;
+ return cloned;
}
@Override
@@ -102,6 +122,7 @@ public class TopKNode extends MultiChildProcessNode {
Symbol.serialize(symbol, byteBuffer);
}
ReadWriteIOUtils.write(childrenDataInOrder, byteBuffer);
+ ReadWriteIOUtils.write(topKRuntimeFilterSourceId, byteBuffer);
}
@Override
@@ -114,6 +135,7 @@ public class TopKNode extends MultiChildProcessNode {
Symbol.serialize(symbol, stream);
}
ReadWriteIOUtils.write(childrenDataInOrder, stream);
+ ReadWriteIOUtils.write(topKRuntimeFilterSourceId, stream);
}
public static TopKNode deserialize(ByteBuffer byteBuffer) {
@@ -125,8 +147,12 @@ public class TopKNode extends MultiChildProcessNode {
outputSymbols.add(Symbol.deserialize(byteBuffer));
}
boolean childrenDataInOrder = ReadWriteIOUtils.readBool(byteBuffer);
+ String topKRuntimeFilterSourceId = ReadWriteIOUtils.readString(byteBuffer);
PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
- return new TopKNode(planNodeId, orderingScheme, count, outputSymbols,
childrenDataInOrder);
+ TopKNode topKNode =
+ new TopKNode(planNodeId, orderingScheme, count, outputSymbols,
childrenDataInOrder);
+ topKNode.topKRuntimeFilterSourceId = topKRuntimeFilterSourceId;
+ return topKNode;
}
@Override
@@ -139,7 +165,10 @@ 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.topKRuntimeFilterSourceId = topKRuntimeFilterSourceId;
+ return topKNode;
}
public OrderingScheme getOrderingScheme() {
@@ -154,6 +183,20 @@ public class TopKNode extends MultiChildProcessNode {
return childrenDataInOrder;
}
+ public boolean isTopKRuntimeFilterAscending() {
+ return topKRuntimeFilterAscending;
+ }
+
+ /** A non-null source id marks this TopK as a runtime filter producer. */
+ @Nullable
+ public String getTopKRuntimeFilterSourceId() {
+ return topKRuntimeFilterSourceId;
+ }
+
+ public void setTopKRuntimeFilterSourceId(@Nullable String
topKRuntimeFilterSourceId) {
+ this.topKRuntimeFilterSourceId = topKRuntimeFilterSourceId;
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
@@ -162,12 +205,14 @@ 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)
+ && Objects.equal(topKRuntimeFilterSourceId,
sortNode.topKRuntimeFilterSourceId);
}
@Override
public int hashCode() {
- return Objects.hashCode(super.hashCode(), orderingScheme, outputSymbols,
count);
+ return Objects.hashCode(
+ super.hashCode(), orderingScheme, outputSymbols, count,
topKRuntimeFilterSourceId);
}
@Override
