Support spill of devce entry in query process

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java

@@ -180,4 +180,10 @@ public DataNodeConfig setDnMultiDirStrategy(String multiDirStrategy) {
     setProperty("dn_multi_dir_strategy", multiDirStrategy);
     return this;
   }
+
+  @Override
+  public DataNodeConfig setTableQueryDeviceEntryBatchSizeInBytes(long batchSizeInBytes) {
+    setProperty("table_query_device_entry_batch_size_in_bytes", String.valueOf(batchSizeInBytes));
+    return this;
+  }
 }

integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java

@@ -125,4 +125,9 @@ public DataNodeConfig setDnDataDirs(String dnDataDirs) {
   public DataNodeConfig setDnMultiDirStrategy(String multiDirStrategy) {
     return this;
   }
+
+  @Override
+  public DataNodeConfig setTableQueryDeviceEntryBatchSizeInBytes(long batchSizeInBytes) {
+    return this;
+  }
 }

integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java

@@ -65,4 +65,6 @@ DataNodeConfig setLoadActiveListeningCheckIntervalSeconds(
   DataNodeConfig setDnDataDirs(String dnDataDirs);
 
   DataNodeConfig setDnMultiDirStrategy(String multiDirStrategy);
+
+  DataNodeConfig setTableQueryDeviceEntryBatchSizeInBytes(long batchSizeInBytes);
 }

integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBDeviceEntrySpillIT.java

@@ -0,0 +1,143 @@
+/*
+ * 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.itbase.category.TableClusterIT;
+import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
+import org.apache.iotdb.itbase.env.BaseEnv;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+
+import static org.junit.Assert.assertEquals;
+
+@Category({TableLocalStandaloneIT.class, TableClusterIT.class})
+public class IoTDBDeviceEntrySpillIT {
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    EnvFactory.getEnv().getConfig().getDataNodeConfig().setTableQueryDeviceEntryBatchSizeInBytes(1);
+    EnvFactory.getEnv().initClusterEnvironment();
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("CREATE DATABASE spill_test");
+      statement.execute(
+          "CREATE TABLE spill_test.device_data (tag1 STRING TAG, tag2 STRING TAG, "
+              + "value INT32 FIELD)");
+      statement.execute(
+          "INSERT INTO spill_test.device_data(tag1, tag2, time, value) "
+              + "VALUES ('a', 'x', 1, 10), ('a', 'x', 2, 20), "
+              + "('b', 'y', 1, 30), ('c', 'z', 1, 40)");
+    }
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+  }
+
+  @Test
+  public void testRawFullTableQueryWithSpill() throws Exception {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement();
+        ResultSet resultSet =
+            statement.executeQuery("SELECT tag1, tag2, value FROM spill_test.device_data")) {
+      int rowCount = 0;
+      while (resultSet.next()) {
+        rowCount++;
+      }
+      assertEquals(4, rowCount);
+    }
+  }
+
+  @Test
+  public void testRawQueriesWithTimeFilterProjectionFilterLimitAndOrdering() throws Exception {
+    String[] queries = {
+      "SELECT * FROM spill_test.device_data WHERE time >= 1 AND time < 3",
+      "SELECT time, value FROM spill_test.device_data WHERE time >= 1 AND time < 3",
+      "SELECT tag1, tag2, value FROM spill_test.device_data "
+          + "WHERE time >= 1 AND time < 3 AND value > 10",
+      "SELECT * FROM spill_test.device_data WHERE time >= 1 AND time < 3 LIMIT 2",
+      "SELECT * FROM spill_test.device_data WHERE time >= 1 AND time < 3 ORDER BY time ASC",
+      "SELECT * FROM spill_test.device_data WHERE time >= 1 AND time < 3 " + "ORDER BY tag1, time"
+    };
+    for (String query : queries) {
+      assertRowCount(query, query.contains("LIMIT 2") ? 2 : query.contains("value > 10") ? 2 : 4);
+    }
+  }
+
+  @Test
+  public void testAggregationQueryWithSpill() throws Exception {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement();
+        ResultSet resultSet =
+            statement.executeQuery("SELECT count(value) FROM spill_test.device_data")) {
+      assertEquals(true, resultSet.next());
+      assertEquals(4, resultSet.getLong(1));
+      assertEquals(false, resultSet.next());
+    }
+  }
+
+  @Test
+  public void testGroupedAggregationAcrossSpillSegments() throws Exception {
+    assertRowCount("SELECT tag1, count(*) FROM spill_test.device_data GROUP BY tag1", 3);
+    assertRowCount(
+        "SELECT tag1, tag2, count(*), sum(value) FROM spill_test.device_data "
+            + "GROUP BY tag1, tag2",
+        3);
+    assertRowCount(
+        "SELECT date_bin(1s, time), count(*) FROM spill_test.device_data "
+            + "GROUP BY date_bin(1s, time)",
+        2);
+  }
+
+  @Test
+  public void testOrPredicateDoesNotDuplicateDeviceRows() throws Exception {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement();
+        ResultSet resultSet =
+            statement.executeQuery(
+                "SELECT count(*) FROM spill_test.device_data "
+                    + "WHERE tag1 = 'a' OR tag2 = 'x'")) {
+      assertEquals(true, resultSet.next());
+      assertEquals(2, resultSet.getLong(1));
+      assertEquals(false, resultSet.next());
+    }
+  }
+
+  private void assertRowCount(String sql, int expectedRowCount) throws Exception {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement();
+        ResultSet resultSet = statement.executeQuery(sql)) {
+      int rowCount = 0;
+      while (resultSet.next()) {
+        rowCount++;
+      }
+      assertEquals(expectedRowCount, rowCount);
+    }
+  }
+}

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java

@@ -3818,5 +3818,10 @@ private DataNodeQueryMessages() {}
   public static final String EXCEPTION_VISIBLEALIASES_IS_NULL_630B27F1 = "visibleAliases is null";
   public static final String EXCEPTION_HAS_NO_PERMISSION_TO_EXECUTE_ARG_BECAUSE_ONLY_THE_SUPERUSER_CAN_ALTER_HIM_HERSELF_C5902893 =
       "Has no permission to execute %s, because only the superuser can alter him/herself.";
+  public static final String
+      LOG_FAILED_TO_CLEAN_DEVICEENTRY_DATA_SET_ASYNCHRONOUSLY_QUERYID_ARG_PLANNODEID_ARG_9106C4C5 =
+          "Failed to clean DeviceEntry data set asynchronously: queryId=%s, planNodeId=%s";
+  public static final String LOG_FAILED_TO_CLEAN_DEVICEENTRY_SPILL_DIRECTORY_QUERYID_ARG_ADF95D63 =
+      "Failed to clean DeviceEntry spill directory for query %s";
 
 }

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java

@@ -4575,5 +4575,10 @@ private DataNodeQueryMessages() {}
   public static final String EXCEPTION_VISIBLEALIASES_IS_NULL_630B27F1 = "visibleAliases 不能为空";
   public static final String EXCEPTION_HAS_NO_PERMISSION_TO_EXECUTE_ARG_BECAUSE_ONLY_THE_SUPERUSER_CAN_ALTER_HIM_HERSELF_C5902893 =
       "无权执行 %s，因为只有超级用户可以修改其自身。";
+  public static final String
+      LOG_FAILED_TO_CLEAN_DEVICEENTRY_DATA_SET_ASYNCHRONOUSLY_QUERYID_ARG_PLANNODEID_ARG_9106C4C5 =
+          "异步清理 DeviceEntry 数据集失败：queryId=%s，planNodeId=%s";
+  public static final String LOG_FAILED_TO_CLEAN_DEVICEENTRY_SPILL_DIRECTORY_QUERYID_ARG_ADF95D63 =
+      "清理 query %s 的 DeviceEntry spill 目录失败";
 
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -256,6 +256,9 @@ public class IoTDBConfig {
   private String queryDir =
       IoTDBConstant.DN_DEFAULT_DATA_DIR + File.separator + IoTDBConstant.QUERY_FOLDER_NAME;
 
+  /** Maximum DeviceEntry bytes kept in memory before a table-query spill. */
+  private long tableQueryDeviceEntryBatchSizeInBytes;
+
   /** External lib directory, stores user-uploaded JAR files */
   private String extDir = IoTDBConstant.EXT_FOLDER_NAME;
 
@@ -1789,6 +1792,14 @@ public void setQueryDir(String queryDir) {
     this.queryDir = queryDir;
   }
 
+  public long getTableQueryDeviceEntryBatchSizeInBytes() {
+    return tableQueryDeviceEntryBatchSizeInBytes;
+  }
+
+  public void setTableQueryDeviceEntryBatchSizeInBytes(long tableQueryDeviceEntryBatchSizeInBytes) {
+    this.tableQueryDeviceEntryBatchSizeInBytes = tableQueryDeviceEntryBatchSizeInBytes;
+  }
+
   public String getRatisDataRegionSnapshotDir() {
     return ratisDataRegionSnapshotDir;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -352,6 +352,18 @@ public void loadProperties(TrimProperties properties) throws BadNodeUrlException
 
     conf.setQueryDir(
         FilePathUtils.regularizePath(conf.getSystemDir() + IoTDBConstant.QUERY_FOLDER_NAME));
+    long deviceEntryBatchSize =
+        Long.parseLong(
+            properties.getProperty(
+                "table_query_device_entry_batch_size_in_bytes",
+                Long.toString(conf.getTableQueryDeviceEntryBatchSizeInBytes())));
+    if (deviceEntryBatchSize <= 0) {
+      deviceEntryBatchSize =
+          memoryConfig.getOperatorsMemoryManager().getTotalMemorySizeInBytes()
+              / memoryConfig.getQueryThreadCount()
+              / 4;
+    }
+    conf.setTableQueryDeviceEntryBatchSizeInBytes(deviceEntryBatchSize);
     String[] defaultTierDirs = new String[conf.getTierDataDirs().length];
     for (int i = 0; i < defaultTierDirs.length; ++i) {
       defaultTierDirs[i] = String.join(",", conf.getTierDataDirs()[i]);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java

@@ -48,6 +48,7 @@
 import org.apache.iotdb.db.queryengine.plan.planner.LocalExecutionPlanner;
 import org.apache.iotdb.db.queryengine.plan.planner.memory.NotThreadSafeMemoryReservationManager;
 import org.apache.iotdb.db.queryengine.plan.relational.function.tvf.read_tsfile.ExternalTsFileQueryResource;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryIOContext;
 import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExplainOutputFormat;
 import org.apache.iotdb.db.queryengine.statistics.QueryPlanStatistics;
 
@@ -132,6 +133,8 @@
 
   private QueryPlanStatistics queryPlanStatistics = null;
 
+  private DeviceEntryIOContext deviceEntryIOContext;
+
   // To avoid query front-end from consuming too much memory, it needs to reserve memory when
   // constructing some Expression and PlanNode.
   private final MemoryReservationManager memoryReservationManager;
@@ -403,6 +406,13 @@
     this.startTime = startTime;
   }
 
+  public DeviceEntryIOContext getOrCreateDeviceEntryIOContext(boolean duringFetchSchema) {
+    if (deviceEntryIOContext == null) {
+      deviceEntryIOContext = new DeviceEntryIOContext(this, duringFetchSchema);
+    }
+    return deviceEntryIOContext;
+  }
+
   public void addFailedEndPoint(TEndPoint endPoint) {
     this.endPointBlackList.add(endPoint);
   }
@@ -528,6 +538,37 @@
     return queryPlanStatistics.getDispatchCost();
   }
 
+  public void recordDeviceEntryDiskIODuringFetchSchema(long bytes, long timeCost) {
+    getOrCreateQueryPlanStatistics().recordDeviceEntryDiskIODuringFetchSchema(bytes, timeCost);
+  }
+
+  public void recordDeviceEntryCount(long count) {
+    getOrCreateQueryPlanStatistics().recordDeviceEntryCount(count);
+  }
+
+  public long getDiskIOSizeForDeviceEntryDuringFetchSchema() {
+    return queryPlanStatistics == null
+        ? 0
+        : queryPlanStatistics.getDiskIOSizeForDeviceEntryDuringFetchSchema();
+  }
+
+  public long getDiskIOTimeCostForDeviceEntryDuringFetchSchema() {
+    return queryPlanStatistics == null
+        ? 0
+        : queryPlanStatistics.getDiskIOTimeCostForDeviceEntryDuringFetchSchema();
+  }
+
+  public long getDeviceEntryCount() {
+    return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDeviceEntryCount();
+  }
+
+  private QueryPlanStatistics getOrCreateQueryPlanStatistics() {
+    if (queryPlanStatistics == null) {
+      queryPlanStatistics = new QueryPlanStatistics();
+    }
+    return queryPlanStatistics;
+  }
+
   public void setAnalyzeCost(long analyzeCost) {
     if (queryPlanStatistics == null) {
       queryPlanStatistics = new QueryPlanStatistics();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/driver/DataDriver.java

@@ -27,6 +27,7 @@
 import org.apache.iotdb.db.queryengine.execution.operator.source.DataSourceOperator;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance;
 import org.apache.iotdb.db.storageengine.dataregion.read.IQueryDataSource;
+import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSourceType;
 
 import com.google.common.util.concurrent.SettableFuture;
 
@@ -36,6 +37,7 @@
 
 import static com.google.common.base.Throwables.throwIfUnchecked;
 import static org.apache.iotdb.calc.metric.QueryExecutionMetricSet.QUERY_RESOURCE_INIT;
+import static org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion.EMPTY_QUERY_DATA_SOURCE;
 import static org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion.UNFINISHED_QUERY_DATA_SOURCE;
 
 /**
@@ -111,6 +113,16 @@ private boolean initialize() throws QueryProcessException {
       List<DataSourceOperator> sourceOperators =
           ((DataDriverContext) driverContext).getSourceOperators();
       if (sourceOperators != null && !sourceOperators.isEmpty()) {
+        if (((DataDriverContext) driverContext)
+            .getQueryDataSourceType()
+            .filter(type -> type == QueryDataSourceType.BATCH_SERIES_SCAN)
+            .isPresent()) {
+          sourceOperators.forEach(
+              sourceOperator ->
+                  sourceOperator.initQueryDataSource(EMPTY_QUERY_DATA_SOURCE.clone()));
+          this.init = true;
+          return true;
+        }
         IQueryDataSource dataSource = initQueryDataSource();
         if (dataSource == null) {
           // If this driver is being initialized, meanwhile the whole FI was aborted or cancelled