You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by we...@apache.org on 2026/08/12 01:49:27 UTC
(iotdb) branch deviceEntrySpill-dev created (now d6914dea463)
This is an automated email from the ASF dual-hosted git repository.
Wei-hao-Li pushed a change to branch deviceEntrySpill-dev
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at d6914dea463 modify
This branch includes the following new commits:
new 701f7b07da2 extracted model-1
new d6914dea463 modify
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 02/02: modify
Posted by we...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Wei-hao-Li pushed a commit to branch deviceEntrySpill-dev
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d6914dea463be7ed6dc364b33ee6fa56192d4e73
Author: Weihao Li <18...@163.com>
AuthorDate: Wed Aug 12 09:48:45 2026 +0800
modify
Signed-off-by: Weihao Li <18...@163.com>
---
.../db/queryengine/common/MPPQueryContext.java | 32 ++++------
.../metadata/fetcher/TableDeviceSchemaFetcher.java | 74 ++++++++++------------
.../spill/AbstractDeviceEntryMaterializer.java | 4 +-
.../metadata/spill/DeviceEntryDiskSpiller.java | 1 -
.../metadata/spill/DeviceEntryIOContext.java | 19 +++---
.../metadata/spill/DeviceEntryMaterializer.java | 5 +-
.../FragmentInstanceStatisticsDrawer.java | 16 ++---
.../FragmentInstanceStatisticsJsonDrawer.java | 11 ++--
.../statistics/QueryPlanStatistics.java | 41 ++++--------
.../FragmentInstanceStatisticsJsonDrawerTest.java | 14 ++--
10 files changed, 94 insertions(+), 123 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
index f7ff41469ad..25edf3a4131 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
@@ -528,32 +528,28 @@ public class MPPQueryContext implements IAuditEntity {
return queryPlanStatistics.getDispatchCost();
}
- public void recordDeviceEntryDiskIO(long bytes, long timeCost) {
- getOrCreateQueryPlanStatistics().recordDeviceEntryDiskIO(bytes, timeCost);
+ public void recordDeviceEntryDiskIODuringFetchSchema(long bytes, long timeCost) {
+ getOrCreateQueryPlanStatistics().recordDeviceEntryDiskIODuringFetchSchema(bytes, timeCost);
}
- public void recordDeviceEntrySegment() {
- getOrCreateQueryPlanStatistics().recordDeviceEntrySegment();
+ public void recordDeviceEntryCount(long count) {
+ getOrCreateQueryPlanStatistics().recordDeviceEntryCount(count);
}
- public void recordDeviceEntrySortedRun() {
- getOrCreateQueryPlanStatistics().recordDeviceEntrySortedRun();
+ public long getDiskIOSizeForDeviceEntryDuringFetchSchema() {
+ return queryPlanStatistics == null
+ ? 0
+ : queryPlanStatistics.getDiskIOSizeForDeviceEntryDuringFetchSchema();
}
- public long getDiskIOSizeForDeviceEntry() {
- return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDiskIOSizeForDeviceEntry();
+ public long getDiskIOTimeCostForDeviceEntryDuringFetchSchema() {
+ return queryPlanStatistics == null
+ ? 0
+ : queryPlanStatistics.getDiskIOTimeCostForDeviceEntryDuringFetchSchema();
}
- public long getDiskIOTimeCostForDeviceEntry() {
- return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDiskIOTimeCostForDeviceEntry();
- }
-
- public long getDeviceEntrySegmentCount() {
- return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDeviceEntrySegmentCount();
- }
-
- public long getDeviceEntrySortedRunCount() {
- return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDeviceEntrySortedRunCount();
+ public long getDeviceEntryCount() {
+ return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDeviceEntryCount();
}
private QueryPlanStatistics getOrCreateQueryPlanStatistics() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
index 57f20ba6d45..1ac6e1789a4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
@@ -245,14 +245,7 @@ public class TableDeviceSchemaFetcher {
mayContainDuplicateDevice,
false)) {
fetchMissingDeviceSchemaForQuery(
- database,
- tableInstance,
- attributeColumns,
- statement,
- deviceEntryMap,
- null,
- queryContext,
- null);
+ database, tableInstance, attributeColumns, statement, deviceEntryMap, null, queryContext);
}
// TODO table metadata: implement deduplicate during schemaRegion execution
@@ -278,8 +271,7 @@ public class TableDeviceSchemaFetcher {
if (TreeViewSchema.isTreeViewTable(tableInstance)) {
final Map<String, List<DeviceEntry>> deviceEntryMap = new HashMap<>();
final AtomicBoolean mayContainDuplicateDevice = new AtomicBoolean(false);
- // pass by reference
- final AtomicBoolean containsNonAlignedDevice = new AtomicBoolean(false);
+ boolean containsNonAlignedDevice = false;
final ShowDevice statement = new ShowDevice(database, table);
try (DeviceEntryMaterializer materializer =
new DeviceEntryMaterializer(
@@ -302,21 +294,21 @@ public class TableDeviceSchemaFetcher {
for (DeviceEntry entry : entries) {
appendToMaterializer(materializer, entry, queryContext, true);
if (entry instanceof NonAlignedDeviceEntry) {
- containsNonAlignedDevice.set(true);
+ containsNonAlignedDevice = true;
}
}
entries.clear();
}
if (needRemoteFetch) {
- fetchMissingDeviceSchemaForQuery(
- database,
- tableInstance,
- attributeColumns,
- statement,
- deviceEntryMap,
- materializer,
- queryContext,
- containsNonAlignedDevice);
+ containsNonAlignedDevice |=
+ fetchMissingDeviceSchemaForQuery(
+ database,
+ tableInstance,
+ attributeColumns,
+ statement,
+ deviceEntryMap,
+ materializer,
+ queryContext);
}
if (deviceEntryMap.size() > 1) {
throw new SemanticException(
@@ -327,7 +319,7 @@ public class TableDeviceSchemaFetcher {
final String resultDatabase =
deviceEntryMap.isEmpty() ? null : deviceEntryMap.keySet().iterator().next();
return new DeviceEntryDataSetResult(
- resultDatabase, materializer.finish(), containsNonAlignedDevice.get());
+ resultDatabase, materializer.finish(), containsNonAlignedDevice);
} catch (IOException e) {
throw new UncheckedIOException(e);
}
@@ -357,8 +349,7 @@ public class TableDeviceSchemaFetcher {
statement,
cachedEntries,
null,
- queryContext,
- null);
+ queryContext);
}
cachedEntries.put(
database, new ArrayList<>(new LinkedHashSet<>(cachedEntries.get(database))));
@@ -383,8 +374,7 @@ public class TableDeviceSchemaFetcher {
statement,
cachedEntries,
materializer,
- queryContext,
- null);
+ queryContext);
}
final DeviceEntryDataSet dataSet = materializer.finish();
return new DeviceEntryDataSetResult(database, dataSet, false);
@@ -629,16 +619,16 @@ public class TableDeviceSchemaFetcher {
return IDeviceID.Factory.DEFAULT_FACTORY.create(deviceIdNodes);
}
- private void fetchMissingDeviceSchemaForQuery(
+ private boolean fetchMissingDeviceSchemaForQuery(
final String database,
final TsTable tableInstance,
final List<String> attributeColumns,
final ShowDevice statement,
final Map<String, List<DeviceEntry>> deviceEntryMap,
final DeviceEntryMaterializer materializer,
- final MPPQueryContext mppQueryContext,
- final AtomicBoolean containsNonAlignedDevice) {
+ final MPPQueryContext mppQueryContext) {
Throwable t = null;
+ boolean containsNonAlignedDevice = false;
final long queryId = SessionManager.getInstance().requestQueryId();
// For the correctness of attribute remote update
@@ -712,14 +702,14 @@ public class TableDeviceSchemaFetcher {
deviceEntryMap.get(database),
materializer);
} else {
- constructTreeResults(
- tsBlock.get(),
- columnHeaderList,
- tableInstance,
- mppQueryContext,
- deviceEntryMap,
- materializer,
- containsNonAlignedDevice);
+ containsNonAlignedDevice |=
+ constructTreeResults(
+ tsBlock.get(),
+ columnHeaderList,
+ tableInstance,
+ mppQueryContext,
+ deviceEntryMap,
+ materializer);
}
}
} else {
@@ -732,6 +722,7 @@ public class TableDeviceSchemaFetcher {
TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
}
}
+ return containsNonAlignedDevice;
} catch (final Throwable throwable) {
t = throwable;
throw throwable;
@@ -803,14 +794,14 @@ public class TableDeviceSchemaFetcher {
}
}
- private void constructTreeResults(
+ private boolean constructTreeResults(
final TsBlock tsBlock,
final List<ColumnHeader> columnHeaderList,
final TsTable tableInstance,
final MPPQueryContext mppQueryContext,
final Map<String, List<DeviceEntry>> deviceEntryMap,
- final DeviceEntryMaterializer materializer,
- final AtomicBoolean containsNonAlignedDevice) {
+ final DeviceEntryMaterializer materializer) {
+ boolean containsNonAlignedDevice = false;
final Column[] columns = tsBlock.getValueColumns();
for (int i = 0; i < tsBlock.getPositionCount(); i++) {
final String[] nodes = new String[tableInstance.getTagNum()];
@@ -822,6 +813,7 @@ public class TableDeviceSchemaFetcher {
columns[columns.length - 2].getBoolean(i)
? new AlignedDeviceEntry(deviceID, new Binary[0])
: new NonAlignedDeviceEntry(deviceID, new Binary[0]);
+ containsNonAlignedDevice |= deviceEntry instanceof NonAlignedDeviceEntry;
final List<DeviceEntry> deviceEntries =
deviceEntryMap.computeIfAbsent(
columns[columns.length - 1].getBinary(i).getStringValue(TSFileConfig.STRING_CHARSET),
@@ -831,11 +823,9 @@ public class TableDeviceSchemaFetcher {
deviceEntries.add(deviceEntry);
} else {
appendToMaterializer(materializer, deviceEntry, mppQueryContext, false);
- if (deviceEntry instanceof NonAlignedDeviceEntry) {
- containsNonAlignedDevice.set(true);
- }
}
}
+ return containsNonAlignedDevice;
}
private void constructNodesArrayAndAttributeMap(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/AbstractDeviceEntryMaterializer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/AbstractDeviceEntryMaterializer.java
index 6959387a4aa..a481423b3b7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/AbstractDeviceEntryMaterializer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/AbstractDeviceEntryMaterializer.java
@@ -97,8 +97,8 @@ public abstract class AbstractDeviceEntryMaterializer implements AutoCloseable {
public abstract void forceSpill() throws IOException;
- public final void setQueryContext(MPPQueryContext queryContext) {
- ioContext = new DeviceEntryIOContext(queryContext);
+ protected final void setQueryContext(MPPQueryContext queryContext, boolean duringFetchSchema) {
+ ioContext = new DeviceEntryIOContext(queryContext, duringFetchSchema);
}
protected final DeviceEntryIOContext ioContext() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDiskSpiller.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDiskSpiller.java
index 50c3965dfee..53b8dda0632 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDiskSpiller.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDiskSpiller.java
@@ -101,7 +101,6 @@ public final class DeviceEntryDiskSpiller implements AutoCloseable {
sealedSegments.add(sealedFile);
if (ioContext != null) {
ioContext.recordDiskIO(0, startNanos);
- ioContext.recordSegment();
}
temporaryFile = null;
currentBytes = 0;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryIOContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryIOContext.java
index c434e0fece7..653cb91f0f3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryIOContext.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryIOContext.java
@@ -27,11 +27,13 @@ import java.util.concurrent.TimeUnit;
public final class DeviceEntryIOContext {
private final MPPQueryContext queryContext;
+ private final boolean duringFetchSchema;
private final long timeoutStartNanos;
private final long remainingTimeoutNanos;
- public DeviceEntryIOContext(MPPQueryContext queryContext) {
+ public DeviceEntryIOContext(MPPQueryContext queryContext, boolean duringFetchSchema) {
this.queryContext = queryContext;
+ this.duringFetchSchema = duringFetchSchema;
this.timeoutStartNanos = System.nanoTime();
long elapsedMillis = Math.max(0, System.currentTimeMillis() - queryContext.getStartTime());
long remainingTimeoutMillis = Math.max(0, queryContext.getTimeOut() - elapsedMillis);
@@ -46,15 +48,16 @@ public final class DeviceEntryIOContext {
}
public void recordDiskIO(long bytes, long startNanos) {
- queryContext.recordDeviceEntryDiskIO(bytes, System.nanoTime() - startNanos);
+ long timeCost = System.nanoTime() - startNanos;
+ if (duringFetchSchema) {
+ queryContext.recordDeviceEntryDiskIODuringFetchSchema(bytes, timeCost);
+ }
checkTimeout();
}
- public void recordSegment() {
- queryContext.recordDeviceEntrySegment();
- }
-
- public void recordSortedRun() {
- queryContext.recordDeviceEntrySortedRun();
+ public void recordDeviceEntryCount(long count) {
+ if (duringFetchSchema) {
+ queryContext.recordDeviceEntryCount(count);
+ }
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializer.java
index 92d7cf64d48..41308337953 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializer.java
@@ -47,7 +47,7 @@ public final class DeviceEntryMaterializer extends AbstractDeviceEntryMaterializ
boolean rawSegment,
MPPQueryContext queryContext) {
this(queryId, planNodeId, thresholdInBytes, rawSegment);
- setQueryContext(queryContext);
+ setQueryContext(queryContext, rawSegment);
}
@Override
@@ -97,6 +97,9 @@ public final class DeviceEntryMaterializer extends AbstractDeviceEntryMaterializ
new SpilledDeviceEntryDataSet(
queryId(), ownerDirectory(), spiller.finish(), entryCount(), !rawSegment);
}
+ if (ioContext() != null) {
+ ioContext().recordDeviceEntryCount(entryCount());
+ }
markFinished();
return dataSet;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java
index da6f415af8d..7c4a45cfcac 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java
@@ -57,21 +57,15 @@ public class FragmentInstanceStatisticsDrawer {
planHeader,
1,
String.format(
- "Disk IO Size for DeviceEntry: %d bytes", context.getDiskIOSizeForDeviceEntry()));
+ "Disk IO Size for DeviceEntry During FetchSchema: %d bytes",
+ context.getDiskIOSizeForDeviceEntryDuringFetchSchema()));
addLine(
planHeader,
1,
String.format(
- "Disk IO Time Cost for DeviceEntry: %.3f ms",
- context.getDiskIOTimeCostForDeviceEntry() * NS_TO_MS_FACTOR));
- addLine(
- planHeader,
- 1,
- String.format("DeviceEntry Segment Count: %d", context.getDeviceEntrySegmentCount()));
- addLine(
- planHeader,
- 1,
- String.format("DeviceEntry Sorted Run Count: %d", context.getDeviceEntrySortedRunCount()));
+ "Disk IO Time Cost for DeviceEntry During FetchSchema: %.3f ms",
+ context.getDiskIOTimeCostForDeviceEntryDuringFetchSchema() * NS_TO_MS_FACTOR));
+ addLine(planHeader, 1, String.format("DeviceEntry Count: %d", context.getDeviceEntryCount()));
addLine(
planHeader,
0,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java
index 9e727cd7517..ef887034dac 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java
@@ -55,12 +55,13 @@ public class FragmentInstanceStatisticsJsonDrawer {
"fetchPartitionCostMs", formatMs(context.getFetchPartitionCost() * NS_TO_MS_FACTOR));
planStatistics.addProperty(
"fetchSchemaCostMs", formatMs(context.getFetchSchemaCost() * NS_TO_MS_FACTOR));
- planStatistics.addProperty("diskIOSizeForDeviceEntry", context.getDiskIOSizeForDeviceEntry());
planStatistics.addProperty(
- "diskIOTimeCostForDeviceEntryMs",
- formatMs(context.getDiskIOTimeCostForDeviceEntry() * NS_TO_MS_FACTOR));
- planStatistics.addProperty("deviceEntrySegmentCount", context.getDeviceEntrySegmentCount());
- planStatistics.addProperty("deviceEntrySortedRunCount", context.getDeviceEntrySortedRunCount());
+ "diskIOSizeForDeviceEntryDuringFetchSchema",
+ context.getDiskIOSizeForDeviceEntryDuringFetchSchema());
+ planStatistics.addProperty(
+ "diskIOTimeCostForDeviceEntryDuringFetchSchemaMs",
+ formatMs(context.getDiskIOTimeCostForDeviceEntryDuringFetchSchema() * NS_TO_MS_FACTOR));
+ planStatistics.addProperty("deviceEntryCount", context.getDeviceEntryCount());
planStatistics.addProperty(
"logicalPlanCostMs", formatMs(context.getLogicalPlanCost() * NS_TO_MS_FACTOR));
planStatistics.addProperty(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java
index 592d45d2e63..aa46addce37 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java
@@ -19,8 +19,6 @@
package org.apache.iotdb.db.queryengine.statistics;
-import java.util.concurrent.atomic.AtomicLong;
-
public class QueryPlanStatistics {
private long analyzeCost;
private long fetchPartitionCost;
@@ -29,12 +27,9 @@ public class QueryPlanStatistics {
private long logicalOptimizationCost;
private long distributionPlanCost;
private long dispatchCost = 0;
- // DeviceEntry materialization may involve multiple Region materializers. Use the same
- // lock-free accumulation style as execution.fragment.QueryStatistics.
- private final AtomicLong diskIOSizeForDeviceEntry = new AtomicLong();
- private final AtomicLong diskIOTimeCostForDeviceEntry = new AtomicLong();
- private final AtomicLong deviceEntrySegmentCount = new AtomicLong();
- private final AtomicLong deviceEntrySortedRunCount = new AtomicLong();
+ private long diskIOSizeForDeviceEntryDuringFetchSchema;
+ private long diskIOTimeCostForDeviceEntryDuringFetchSchema;
+ private long deviceEntryCount;
public void setAnalyzeCost(long analyzeCost) {
this.analyzeCost = analyzeCost;
@@ -92,32 +87,24 @@ public class QueryPlanStatistics {
return dispatchCost;
}
- public void recordDeviceEntryDiskIO(long bytes, long timeCost) {
- diskIOSizeForDeviceEntry.addAndGet(bytes);
- diskIOTimeCostForDeviceEntry.addAndGet(timeCost);
- }
-
- public void recordDeviceEntrySegment() {
- deviceEntrySegmentCount.incrementAndGet();
- }
-
- public void recordDeviceEntrySortedRun() {
- deviceEntrySortedRunCount.incrementAndGet();
+ public void recordDeviceEntryDiskIODuringFetchSchema(long bytes, long timeCost) {
+ diskIOSizeForDeviceEntryDuringFetchSchema += bytes;
+ diskIOTimeCostForDeviceEntryDuringFetchSchema += timeCost;
}
- public long getDiskIOSizeForDeviceEntry() {
- return diskIOSizeForDeviceEntry.get();
+ public void recordDeviceEntryCount(long count) {
+ deviceEntryCount += count;
}
- public long getDiskIOTimeCostForDeviceEntry() {
- return diskIOTimeCostForDeviceEntry.get();
+ public long getDiskIOSizeForDeviceEntryDuringFetchSchema() {
+ return diskIOSizeForDeviceEntryDuringFetchSchema;
}
- public long getDeviceEntrySegmentCount() {
- return deviceEntrySegmentCount.get();
+ public long getDiskIOTimeCostForDeviceEntryDuringFetchSchema() {
+ return diskIOTimeCostForDeviceEntryDuringFetchSchema;
}
- public long getDeviceEntrySortedRunCount() {
- return deviceEntrySortedRunCount.get();
+ public long getDeviceEntryCount() {
+ return deviceEntryCount;
}
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java
index ae76d47b3fe..4dd52b45d71 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java
@@ -60,10 +60,8 @@ public class FragmentInstanceStatisticsJsonDrawerTest {
context.setLogicalPlanCost(4000000L); // 4ms
context.setLogicalOptimizationCost(5000000L); // 5ms
context.setDistributionPlanCost(6000000L); // 6ms
- context.recordDeviceEntryDiskIO(8192L, 7000000L); // 8 KiB, 7ms
- context.recordDeviceEntrySegment();
- context.recordDeviceEntrySegment();
- context.recordDeviceEntrySortedRun();
+ context.recordDeviceEntryDiskIODuringFetchSchema(8192L, 7000000L); // 8 KiB, 7ms
+ context.recordDeviceEntryCount(3);
drawer.renderPlanStatistics(context);
@@ -82,10 +80,10 @@ public class FragmentInstanceStatisticsJsonDrawerTest {
assertEquals(4.0, planStats.get("logicalPlanCostMs").getAsDouble(), 0.01);
assertEquals(5.0, planStats.get("logicalOptimizationCostMs").getAsDouble(), 0.01);
assertEquals(6.0, planStats.get("distributionPlanCostMs").getAsDouble(), 0.01);
- assertEquals(8192L, planStats.get("diskIOSizeForDeviceEntry").getAsLong());
- assertEquals(7.0, planStats.get("diskIOTimeCostForDeviceEntryMs").getAsDouble(), 0.01);
- assertEquals(2L, planStats.get("deviceEntrySegmentCount").getAsLong());
- assertEquals(1L, planStats.get("deviceEntrySortedRunCount").getAsLong());
+ assertEquals(8192L, planStats.get("diskIOSizeForDeviceEntryDuringFetchSchema").getAsLong());
+ assertEquals(
+ 7.0, planStats.get("diskIOTimeCostForDeviceEntryDuringFetchSchemaMs").getAsDouble(), 0.01);
+ assertEquals(3L, planStats.get("deviceEntryCount").getAsLong());
}
@Test
(iotdb) 01/02: extracted model-1
Posted by we...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Wei-hao-Li pushed a commit to branch deviceEntrySpill-dev
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 701f7b07da27c793c9cdca030a318ba11c8a98bf
Author: Weihao Li <18...@163.com>
AuthorDate: Wed Aug 12 09:26:19 2026 +0800
extracted model-1
Signed-off-by: Weihao Li <18...@163.com>
---
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 ++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 12 ++
.../db/queryengine/common/MPPQueryContext.java | 35 ++++
.../plan/relational/metadata/DeviceEntry.java | 13 ++
.../plan/relational/metadata/Metadata.java | 8 +-
.../relational/metadata/TableMetadataImpl.java | 62 +-----
.../metadata/fetcher/TableDeviceSchemaFetcher.java | 211 +++++++++++++++++++--
.../spill/AbstractDeviceEntryMaterializer.java | 169 +++++++++++++++++
.../metadata/spill/DeviceEntryDataSet.java | 46 +++++
.../metadata/spill/DeviceEntryDataSetResult.java | 46 +++++
.../metadata/spill/DeviceEntryDiskSpiller.java | 132 +++++++++++++
.../spill/DeviceEntryFileSpillerReader.java | 143 ++++++++++++++
.../metadata/spill/DeviceEntryIOContext.java | 60 ++++++
.../metadata/spill/DeviceEntryMaterializer.java | 132 +++++++++++++
.../metadata/spill/DeviceEntryReader.java | 34 ++++
.../metadata/spill/DeviceEntrySpillManager.java | 189 ++++++++++++++++++
.../metadata/spill/InMemoryDeviceEntryDataSet.java | 76 ++++++++
.../metadata/spill/SpilledDeviceEntryDataSet.java | 85 +++++++++
.../optimizations/PushPredicateIntoTableScan.java | 26 ++-
.../FragmentInstanceStatisticsDrawer.java | 19 ++
.../FragmentInstanceStatisticsJsonDrawer.java | 6 +
.../statistics/QueryPlanStatistics.java | 37 ++++
.../spill/DeviceEntryMaterializerTest.java | 153 +++++++++++++++
.../FragmentInstanceStatisticsJsonDrawerTest.java | 8 +
.../conf/iotdb-system.properties.template | 4 +
25 files changed, 1644 insertions(+), 73 deletions(-)
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 1d726366ddd..19bfa1bf9e6 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
@@ -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 class IoTDBConfig {
this.queryDir = queryDir;
}
+ public long getTableQueryDeviceEntryBatchSizeInBytes() {
+ return tableQueryDeviceEntryBatchSizeInBytes;
+ }
+
+ public void setTableQueryDeviceEntryBatchSizeInBytes(long tableQueryDeviceEntryBatchSizeInBytes) {
+ this.tableQueryDeviceEntryBatchSizeInBytes = tableQueryDeviceEntryBatchSizeInBytes;
+ }
+
public String getRatisDataRegionSnapshotDir() {
return ratisDataRegionSnapshotDir;
}
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 15d2b7d003d..ed46c49e411 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
@@ -352,6 +352,18 @@ public class IoTDBDescriptor {
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]);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
index 495ebc42b65..f7ff41469ad 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
@@ -528,6 +528,41 @@ public class MPPQueryContext implements IAuditEntity {
return queryPlanStatistics.getDispatchCost();
}
+ public void recordDeviceEntryDiskIO(long bytes, long timeCost) {
+ getOrCreateQueryPlanStatistics().recordDeviceEntryDiskIO(bytes, timeCost);
+ }
+
+ public void recordDeviceEntrySegment() {
+ getOrCreateQueryPlanStatistics().recordDeviceEntrySegment();
+ }
+
+ public void recordDeviceEntrySortedRun() {
+ getOrCreateQueryPlanStatistics().recordDeviceEntrySortedRun();
+ }
+
+ public long getDiskIOSizeForDeviceEntry() {
+ return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDiskIOSizeForDeviceEntry();
+ }
+
+ public long getDiskIOTimeCostForDeviceEntry() {
+ return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDiskIOTimeCostForDeviceEntry();
+ }
+
+ public long getDeviceEntrySegmentCount() {
+ return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDeviceEntrySegmentCount();
+ }
+
+ public long getDeviceEntrySortedRunCount() {
+ return queryPlanStatistics == null ? 0 : queryPlanStatistics.getDeviceEntrySortedRunCount();
+ }
+
+ private QueryPlanStatistics getOrCreateQueryPlanStatistics() {
+ if (queryPlanStatistics == null) {
+ queryPlanStatistics = new QueryPlanStatistics();
+ }
+ return queryPlanStatistics;
+ }
+
public void setAnalyzeCost(long analyzeCost) {
if (queryPlanStatistics == null) {
queryPlanStatistics = new QueryPlanStatistics();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java
index 4701e0e9c62..1b5e6ab693b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java
@@ -29,6 +29,7 @@ import org.apache.tsfile.utils.Accountable;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.RamUsageEstimator;
+import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -98,6 +99,14 @@ public abstract class DeviceEntry implements Accountable {
stream);
}
+ public byte[] serializeToBytes() throws IOException {
+ final ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
+ try (DataOutputStream output = new DataOutputStream(byteStream)) {
+ serialize(output);
+ }
+ return byteStream.toByteArray();
+ }
+
public static DeviceEntry deserialize(final ByteBuffer byteBuffer) {
final IDeviceID iDeviceID = StringArrayDeviceID.deserialize(byteBuffer);
int size = readInt(byteBuffer);
@@ -109,6 +118,10 @@ public abstract class DeviceEntry implements Accountable {
return constructDeviceEntry(iDeviceID, attributeColumnValues, readInt(byteBuffer));
}
+ public static DeviceEntry deserialize(final byte[] bytes) {
+ return deserialize(ByteBuffer.wrap(bytes));
+ }
+
public static void serializeBinary(final ByteBuffer byteBuffer, final Binary binary) {
if (binary == null) {
write(-1, byteBuffer);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java
index 8b7f0f32656..5c219440c07 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/Metadata.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.commons.partition.DataPartition;
import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
import org.apache.iotdb.commons.partition.SchemaPartition;
import org.apache.iotdb.commons.queryengine.common.SessionInfo;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.commons.queryengine.plan.relational.function.ITableFunctionFactory;
import org.apache.iotdb.commons.queryengine.plan.relational.function.OperatorType;
import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName;
@@ -37,13 +38,13 @@ import org.apache.iotdb.commons.udf.builtin.relational.TableBuiltinWindowFunctio
import org.apache.iotdb.db.exception.load.LoadAnalyzeTableColumnDisorderException;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableHeaderSchemaValidator;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSetResult;
import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.read.common.type.Type;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -86,11 +87,12 @@ public interface Metadata extends ITypeMetadata, ITableFunctionFactory {
* index scanning
* @param attributeColumns attribute column names
*/
- Map<String, List<DeviceEntry>> indexScan(
+ DeviceEntryDataSetResult indexScan(
final QualifiedObjectName tableName,
final List<Expression> expressionList,
final List<String> attributeColumns,
- final MPPQueryContext context);
+ final MPPQueryContext context,
+ final PlanNodeId planNodeId);
/**
* This method is used for table column validation and should be invoked before device validation.
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
index 9e736a35c58..30ea4a63645 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TableMetadataImpl.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.commons.partition.DataPartition;
import org.apache.iotdb.commons.partition.DataPartitionQueryParam;
import org.apache.iotdb.commons.partition.SchemaPartition;
import org.apache.iotdb.commons.queryengine.common.SessionInfo;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.commons.queryengine.plan.relational.function.OperatorType;
import org.apache.iotdb.commons.queryengine.plan.relational.function.TableFunctionFactory;
import org.apache.iotdb.commons.queryengine.plan.relational.function.arithmetic.AdditionResolver;
@@ -57,6 +58,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.function.DataNodeTableBui
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableDeviceSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableDeviceSchemaValidator;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableHeaderSchemaValidator;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSetResult;
import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl;
import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
import org.apache.iotdb.db.schemaengine.table.ITableCache;
@@ -75,7 +77,6 @@ import org.apache.tsfile.read.common.type.TypeFactory;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
-import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
@@ -1439,24 +1440,6 @@ public class TableMetadataImpl implements Metadata {
functionName));
}
break;
- case SqlConstant.IRATE:
- validateRateFunctionArguments(
- functionName,
- argumentTypes,
- 2,
- DataNodeQueryMessages
- .EXCEPTION_AGGREGATE_FUNCTION_ARG_REQUIRES_2_ARGUMENTS_VALUE_TIME_E2F55C08);
- break;
- case SqlConstant.RATE:
- case SqlConstant.INCREASE:
- case SqlConstant.DELTA:
- validateRateFunctionArguments(
- functionName,
- argumentTypes,
- 4,
- DataNodeQueryMessages
- .EXCEPTION_AGGREGATE_FUNCTION_ARG_REQUIRES_4_ARGUMENTS_VALUE_TIME_WINDOW_START_WINDOW_END_FBEC794B);
- break;
case SqlConstant.COUNT:
break;
default:
@@ -1498,10 +1481,6 @@ public class TableMetadataImpl implements Metadata {
case SqlConstant.REGR_INTERCEPT:
case SqlConstant.SKEWNESS:
case SqlConstant.KURTOSIS:
- case SqlConstant.RATE:
- case SqlConstant.INCREASE:
- case SqlConstant.IRATE:
- case SqlConstant.DELTA:
return DOUBLE;
case SqlConstant.APPROX_MOST_FREQUENT:
return STRING;
@@ -1631,33 +1610,6 @@ public class TableMetadataImpl implements Metadata {
throw new SemanticException(DataNodeQueryMessages.UNKNOWN_FUNCTION + functionName);
}
- private static void validateRateFunctionArguments(
- String functionName,
- List<? extends Type> argumentTypes,
- int expectedArgumentCount,
- String argumentCountError) {
- if (argumentTypes.size() != expectedArgumentCount) {
- throw new SemanticException(String.format(argumentCountError, functionName));
- }
- if (!CommonMetadataUtils.isSupportedMathNumericType(argumentTypes.get(0))) {
- throw new SemanticException(
- String.format(
- DataNodeQueryMessages
- .EXCEPTION_AGGREGATE_FUNCTION_ARG_ONLY_SUPPORTS_INT32_INT64_FLOAT_AND_DOUBLE_AS_THE_FIRST_ARGUMENT_8D201434,
- functionName));
- }
- for (int i = 1; i < argumentTypes.size(); i++) {
- Type argumentType = argumentTypes.get(i);
- if (!INT64.equals(argumentType) && !TIMESTAMP.equals(argumentType)) {
- throw new SemanticException(
- String.format(
- DataNodeQueryMessages
- .EXCEPTION_THE_TIME_ARGUMENTS_OF_AGGREGATE_FUNCTION_ARG_SHOULD_BE_TIMESTAMP_OR_INT64_TYPE_9C736DE3,
- functionName));
- }
- }
- }
-
@Override
public boolean isAggregationFunction(
final SessionInfo session, final String functionName, final AccessControl accessControl) {
@@ -1677,18 +1629,20 @@ public class TableMetadataImpl implements Metadata {
}
@Override
- public Map<String, List<DeviceEntry>> indexScan(
+ public DeviceEntryDataSetResult indexScan(
final QualifiedObjectName tableName,
final List<Expression> expressionList,
final List<String> attributeColumns,
- final MPPQueryContext context) {
+ final MPPQueryContext context,
+ final PlanNodeId planNodeId) {
return TableDeviceSchemaFetcher.getInstance()
- .fetchDeviceSchemaForDataQuery(
+ .fetchDeviceSchemaForDataQueryAsDataSet(
tableName.getDatabaseName(),
tableName.getObjectName(),
expressionList,
attributeColumns,
- context);
+ context,
+ planNodeId);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
index 024b9ef44ac..57f20ba6d45 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
@@ -22,6 +22,8 @@ package org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher;
import org.apache.iotdb.commons.exception.IoTDBException;
import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.commons.exception.QueryTimeoutException;
+import org.apache.iotdb.commons.exception.SemanticException;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression;
import org.apache.iotdb.commons.schema.column.ColumnHeader;
import org.apache.iotdb.commons.schema.filter.SchemaFilter;
@@ -48,6 +50,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.De
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.IDeviceSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceSchemaCache;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TreeDeviceNormalSchema;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSet;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSetResult;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryMaterializer;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AbstractTraverseDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.FetchDevice;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ShowDevice;
@@ -64,6 +69,8 @@ import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.Pair;
+import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -238,7 +245,14 @@ public class TableDeviceSchemaFetcher {
mayContainDuplicateDevice,
false)) {
fetchMissingDeviceSchemaForQuery(
- database, tableInstance, attributeColumns, statement, deviceEntryMap, queryContext);
+ database,
+ tableInstance,
+ attributeColumns,
+ statement,
+ deviceEntryMap,
+ null,
+ queryContext,
+ null);
}
// TODO table metadata: implement deduplicate during schemaRegion execution
@@ -253,6 +267,132 @@ public class TableDeviceSchemaFetcher {
: deviceEntryMap;
}
+ public DeviceEntryDataSetResult fetchDeviceSchemaForDataQueryAsDataSet(
+ final String database,
+ final String table,
+ final List<Expression> expressionList,
+ final List<String> attributeColumns,
+ final MPPQueryContext queryContext,
+ final PlanNodeId planNodeId) {
+ final TsTable tableInstance = DataNodeTableCache.getInstance().getTable(database, table);
+ if (TreeViewSchema.isTreeViewTable(tableInstance)) {
+ final Map<String, List<DeviceEntry>> deviceEntryMap = new HashMap<>();
+ final AtomicBoolean mayContainDuplicateDevice = new AtomicBoolean(false);
+ // pass by reference
+ final AtomicBoolean containsNonAlignedDevice = new AtomicBoolean(false);
+ final ShowDevice statement = new ShowDevice(database, table);
+ try (DeviceEntryMaterializer materializer =
+ new DeviceEntryMaterializer(
+ queryContext.getQueryId().getId(),
+ planNodeId,
+ CONFIG.getTableQueryDeviceEntryBatchSizeInBytes(),
+ true,
+ queryContext)) {
+ final boolean needRemoteFetch =
+ parseFilter4TraverseDevice(
+ tableInstance,
+ expressionList,
+ statement,
+ deviceEntryMap,
+ attributeColumns,
+ queryContext,
+ mayContainDuplicateDevice,
+ false);
+ for (List<DeviceEntry> entries : deviceEntryMap.values()) {
+ for (DeviceEntry entry : entries) {
+ appendToMaterializer(materializer, entry, queryContext, true);
+ if (entry instanceof NonAlignedDeviceEntry) {
+ containsNonAlignedDevice.set(true);
+ }
+ }
+ entries.clear();
+ }
+ if (needRemoteFetch) {
+ fetchMissingDeviceSchemaForQuery(
+ database,
+ tableInstance,
+ attributeColumns,
+ statement,
+ deviceEntryMap,
+ materializer,
+ queryContext,
+ containsNonAlignedDevice);
+ }
+ if (deviceEntryMap.size() > 1) {
+ throw new SemanticException(
+ DataNodeQueryMessages.TREE_DEVICE_VIEW_WITH_MULTIPLE_DATABASES
+ + deviceEntryMap.keySet()
+ + DataNodeQueryMessages.IS_UNSUPPORTED_YET);
+ }
+ final String resultDatabase =
+ deviceEntryMap.isEmpty() ? null : deviceEntryMap.keySet().iterator().next();
+ return new DeviceEntryDataSetResult(
+ resultDatabase, materializer.finish(), containsNonAlignedDevice.get());
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ final Map<String, List<DeviceEntry>> cachedEntries = new HashMap<>();
+ cachedEntries.put(database, new ArrayList<>());
+ final AtomicBoolean mayContainDuplicateDevice = new AtomicBoolean(false);
+ final ShowDevice statement = new ShowDevice(database, table);
+ final boolean needRemoteFetch =
+ parseFilter4TraverseDevice(
+ tableInstance,
+ expressionList,
+ statement,
+ cachedEntries,
+ attributeColumns,
+ queryContext,
+ mayContainDuplicateDevice,
+ false);
+
+ if (mayContainDuplicateDevice.get()) {
+ if (needRemoteFetch) {
+ fetchMissingDeviceSchemaForQuery(
+ database,
+ tableInstance,
+ attributeColumns,
+ statement,
+ cachedEntries,
+ null,
+ queryContext,
+ null);
+ }
+ cachedEntries.put(
+ database, new ArrayList<>(new LinkedHashSet<>(cachedEntries.get(database))));
+ }
+
+ try (DeviceEntryMaterializer materializer =
+ new DeviceEntryMaterializer(
+ queryContext.getQueryId().getId(),
+ planNodeId,
+ CONFIG.getTableQueryDeviceEntryBatchSizeInBytes(),
+ true,
+ queryContext)) {
+ for (DeviceEntry entry : cachedEntries.get(database)) {
+ appendToMaterializer(materializer, entry, queryContext, true);
+ }
+ cachedEntries.get(database).clear();
+ if (needRemoteFetch && !mayContainDuplicateDevice.get()) {
+ fetchMissingDeviceSchemaForQuery(
+ database,
+ tableInstance,
+ attributeColumns,
+ statement,
+ cachedEntries,
+ materializer,
+ queryContext,
+ null);
+ }
+ final DeviceEntryDataSet dataSet = materializer.finish();
+ return new DeviceEntryDataSetResult(database, dataSet, false);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
// Used by show/count device and update device.
// Update / Delete device will not access cache
public boolean parseFilter4TraverseDevice(
@@ -495,7 +635,9 @@ public class TableDeviceSchemaFetcher {
final List<String> attributeColumns,
final ShowDevice statement,
final Map<String, List<DeviceEntry>> deviceEntryMap,
- final MPPQueryContext mppQueryContext) {
+ final DeviceEntryMaterializer materializer,
+ final MPPQueryContext mppQueryContext,
+ final AtomicBoolean containsNonAlignedDevice) {
Throwable t = null;
final long queryId = SessionManager.getInstance().requestQueryId();
@@ -567,10 +709,17 @@ public class TableDeviceSchemaFetcher {
statement,
mppQueryContext,
attributeColumns,
- deviceEntryMap.get(database));
+ deviceEntryMap.get(database),
+ materializer);
} else {
constructTreeResults(
- tsBlock.get(), columnHeaderList, tableInstance, mppQueryContext, deviceEntryMap);
+ tsBlock.get(),
+ columnHeaderList,
+ tableInstance,
+ mppQueryContext,
+ deviceEntryMap,
+ materializer,
+ containsNonAlignedDevice);
}
}
} else {
@@ -602,7 +751,8 @@ public class TableDeviceSchemaFetcher {
final ShowDevice statement,
final MPPQueryContext mppQueryContext,
final List<String> attributeColumns,
- final List<DeviceEntry> deviceEntryList) {
+ final List<DeviceEntry> deviceEntryList,
+ final DeviceEntryMaterializer materializer) {
final Column[] columns = tsBlock.getValueColumns();
for (int i = 0; i < tsBlock.getPositionCount(); i++) {
final String[] nodes = new String[tableInstance.getTagNum() + 1];
@@ -619,22 +769,48 @@ public class TableDeviceSchemaFetcher {
final AlignedDeviceEntry deviceEntry =
new AlignedDeviceEntry(
deviceID, attributeColumns.stream().map(attributeMap::get).toArray(Binary[]::new));
- mppQueryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed());
- deviceEntryList.add(deviceEntry);
+ if (materializer == null) {
+ mppQueryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed());
+ deviceEntryList.add(deviceEntry);
+ } else {
+ appendToMaterializer(materializer, deviceEntry, mppQueryContext, false);
+ }
// Only cache those exact device query
- // Fetch paths is null iff there are fuzzy queries related to tag columns
+ // Fetch paths is null iff there are fuzzy queries related to id columns
if (Objects.nonNull(statement.getPartitionKeyList())) {
cache.putAttributes(statement.getDatabase(), deviceID, attributeMap);
}
}
}
+ private static void appendToMaterializer(
+ DeviceEntryMaterializer materializer,
+ DeviceEntry deviceEntry,
+ MPPQueryContext queryContext,
+ boolean memoryAlreadyReserved) {
+ try {
+ long releasedRamBytes = materializer.appendWithMemoryControl(deviceEntry);
+ if (releasedRamBytes > 0) {
+ queryContext.releaseMemoryReservedForFrontEnd(releasedRamBytes);
+ }
+ if (memoryAlreadyReserved && materializer.isSpilled()) {
+ queryContext.releaseMemoryReservedForFrontEnd(deviceEntry.ramBytesUsed());
+ } else if (!memoryAlreadyReserved && !materializer.isSpilled()) {
+ queryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed());
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
private void constructTreeResults(
final TsBlock tsBlock,
final List<ColumnHeader> columnHeaderList,
final TsTable tableInstance,
final MPPQueryContext mppQueryContext,
- final Map<String, List<DeviceEntry>> deviceEntryMap) {
+ final Map<String, List<DeviceEntry>> deviceEntryMap,
+ final DeviceEntryMaterializer materializer,
+ final AtomicBoolean containsNonAlignedDevice) {
final Column[] columns = tsBlock.getValueColumns();
for (int i = 0; i < tsBlock.getPositionCount(); i++) {
final String[] nodes = new String[tableInstance.getTagNum()];
@@ -646,12 +822,19 @@ public class TableDeviceSchemaFetcher {
columns[columns.length - 2].getBoolean(i)
? new AlignedDeviceEntry(deviceID, new Binary[0])
: new NonAlignedDeviceEntry(deviceID, new Binary[0]);
- mppQueryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed());
- deviceEntryMap
- .computeIfAbsent(
+ final List<DeviceEntry> deviceEntries =
+ deviceEntryMap.computeIfAbsent(
columns[columns.length - 1].getBinary(i).getStringValue(TSFileConfig.STRING_CHARSET),
- k -> new ArrayList<>())
- .add(deviceEntry);
+ k -> new ArrayList<>());
+ if (materializer == null) {
+ mppQueryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed());
+ deviceEntries.add(deviceEntry);
+ } else {
+ appendToMaterializer(materializer, deviceEntry, mppQueryContext, false);
+ if (deviceEntry instanceof NonAlignedDeviceEntry) {
+ containsNonAlignedDevice.set(true);
+ }
+ }
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/AbstractDeviceEntryMaterializer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/AbstractDeviceEntryMaterializer.java
new file mode 100644
index 00000000000..6959387a4aa
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/AbstractDeviceEntryMaterializer.java
@@ -0,0 +1,169 @@
+/*
+ * 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.metadata.spill;
+
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+
+import org.apache.tsfile.external.commons.io.FileUtils;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+
+public abstract class AbstractDeviceEntryMaterializer implements AutoCloseable {
+
+ private final String queryId;
+ private final PlanNodeId planNodeId;
+ private final long thresholdInBytes;
+ private final List<DeviceEntry> bufferedEntries = new ArrayList<>();
+
+ private long entryCount;
+ private Path ownerDirectory;
+ private boolean ownerRegistered;
+ private boolean finished;
+ private DeviceEntryIOContext ioContext;
+
+ protected AbstractDeviceEntryMaterializer(
+ String queryId, PlanNodeId planNodeId, long thresholdInBytes) {
+ if (thresholdInBytes <= 0) {
+ throw new IllegalArgumentException();
+ }
+ this.queryId = queryId;
+ this.planNodeId = planNodeId;
+ this.thresholdInBytes = thresholdInBytes;
+ }
+
+ /**
+ * Appends a DeviceEntry to this materializer's in-memory buffer. Spill decisions are external.
+ */
+ public abstract void append(DeviceEntry entry) throws IOException;
+
+ public abstract DeviceEntryDataSet finish() throws IOException;
+
+ protected final String queryId() {
+ return queryId;
+ }
+
+ protected final long thresholdInBytes() {
+ return thresholdInBytes;
+ }
+
+ protected final void appendToBuffer(DeviceEntry entry) {
+ bufferedEntries.add(entry);
+ entryCount++;
+ }
+
+ protected final void incrementEntryCount() {
+ entryCount++;
+ }
+
+ protected final Iterable<DeviceEntry> bufferedEntries() {
+ return bufferedEntries;
+ }
+
+ protected final boolean isBufferEmpty() {
+ return bufferedEntries.isEmpty();
+ }
+
+ protected final List<DeviceEntry> copyBufferedEntries() {
+ return new ArrayList<>(bufferedEntries);
+ }
+
+ protected final void sortBufferedEntries(Comparator<DeviceEntry> comparator) {
+ bufferedEntries.sort(comparator);
+ }
+
+ public abstract void forceSpill() throws IOException;
+
+ public final void setQueryContext(MPPQueryContext queryContext) {
+ ioContext = new DeviceEntryIOContext(queryContext);
+ }
+
+ protected final DeviceEntryIOContext ioContext() {
+ return ioContext;
+ }
+
+ protected final long entryCount() {
+ return entryCount;
+ }
+
+ protected final void clearBuffer() {
+ bufferedEntries.clear();
+ }
+
+ protected final Path ownerDirectory() {
+ return ownerDirectory;
+ }
+
+ protected final Path ensureOwnerDirectory() throws IOException {
+ if (ownerDirectory == null) {
+ ownerDirectory = DeviceEntrySpillManager.getInstance().register(queryId, planNodeId);
+ ownerRegistered = true;
+ }
+ return ownerDirectory;
+ }
+
+ protected final Path ensureUnregisteredOwnerDirectory(Path rootDirectory) throws IOException {
+ if (ownerDirectory == null) {
+ Path normalizedRoot = rootDirectory.normalize();
+ Path directory = normalizedRoot.resolve(queryId).resolve(planNodeId.getId()).normalize();
+ if (!directory.startsWith(normalizedRoot)) {
+ throw new IllegalArgumentException();
+ }
+ Files.createDirectories(directory);
+ ownerDirectory = directory;
+ }
+ return ownerDirectory;
+ }
+
+ protected final void checkNotFinished() {
+ if (finished) {
+ throw new IllegalStateException();
+ }
+ }
+
+ protected final void markFinished() {
+ finished = true;
+ }
+
+ protected final void cleanupOwnerDirectory() throws IOException {
+ if (ownerDirectory != null) {
+ if (ownerRegistered) {
+ DeviceEntrySpillManager.getInstance().deregisterOwner(queryId, ownerDirectory);
+ } else {
+ FileUtils.deleteDirectory(ownerDirectory.toFile());
+ }
+ ownerDirectory = null;
+ ownerRegistered = false;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (!finished) {
+ cleanupOwnerDirectory();
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSet.java
new file mode 100644
index 00000000000..dabff4daff1
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSet.java
@@ -0,0 +1,46 @@
+/*
+ * 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.metadata.spill;
+
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+
+import java.io.IOException;
+import java.util.List;
+
+public interface DeviceEntryDataSet extends AutoCloseable {
+
+ public long getEntryCount();
+
+ public boolean isSpilled();
+
+ public DeviceEntryReader openReader() throws IOException;
+
+ public default DeviceEntryReader openConsumingReader() throws IOException {
+ throw new UnsupportedOperationException("Open consuming reader is not supported");
+ }
+
+ public default List<DeviceEntry> getInlineEntries() {
+ throw new UnsupportedOperationException(
+ "Only InMemoryDeviceEntryDataSet supports get inline device entries");
+ }
+
+ @Override
+ public void close() throws IOException;
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSetResult.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSetResult.java
new file mode 100644
index 00000000000..7faed3c5a93
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDataSetResult.java
@@ -0,0 +1,46 @@
+/*
+ * 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.metadata.spill;
+
+public final class DeviceEntryDataSetResult {
+
+ private final String database;
+ private final DeviceEntryDataSet dataSet;
+ private final boolean containsNonAlignedDevice;
+
+ public DeviceEntryDataSetResult(
+ String database, DeviceEntryDataSet dataSet, boolean containsNonAlignedDevice) {
+ this.database = database;
+ this.dataSet = dataSet;
+ this.containsNonAlignedDevice = containsNonAlignedDevice;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ public DeviceEntryDataSet getDataSet() {
+ return dataSet;
+ }
+
+ public boolean containsNonAlignedDevice() {
+ return containsNonAlignedDevice;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDiskSpiller.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDiskSpiller.java
new file mode 100644
index 00000000000..50c3965dfee
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryDiskSpiller.java
@@ -0,0 +1,132 @@
+/*
+ * 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.metadata.spill;
+
+import java.io.BufferedOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.List;
+
+public final class DeviceEntryDiskSpiller implements AutoCloseable {
+
+ private final Path directory;
+ private final long targetSegmentBytes;
+ private final DeviceEntryIOContext ioContext;
+ private final List<Path> sealedSegments = new ArrayList<>();
+
+ private DataOutputStream output;
+ private Path temporaryFile;
+ private long currentBytes;
+ private int nextSegmentId;
+
+ public DeviceEntryDiskSpiller(Path directory, long targetSegmentBytes) throws IOException {
+ this(directory, targetSegmentBytes, null);
+ }
+
+ public DeviceEntryDiskSpiller(
+ Path directory, long targetSegmentBytes, DeviceEntryIOContext ioContext) throws IOException {
+ this.directory = directory;
+ this.targetSegmentBytes = targetSegmentBytes;
+ this.ioContext = ioContext;
+ Files.createDirectories(directory);
+ }
+
+ public void append(byte[] serializedEntry) throws IOException {
+ checkTimeout();
+ long startNanos = System.nanoTime();
+ int recordBytes = Integer.BYTES + serializedEntry.length;
+ if (currentBytes > 0 && currentBytes + recordBytes > targetSegmentBytes) {
+ sealCurrentSegment();
+ }
+ ensureOutput();
+ output.writeInt(serializedEntry.length);
+ output.write(serializedEntry);
+ currentBytes += recordBytes;
+ recordDiskIO(recordBytes, startNanos);
+ }
+
+ public List<Path> finish() throws IOException {
+ sealCurrentSegment();
+ return List.copyOf(sealedSegments);
+ }
+
+ private void ensureOutput() throws IOException {
+ if (output != null) {
+ return;
+ }
+ temporaryFile = directory.resolve(String.format("segment-%06d.tmp", nextSegmentId));
+ output = new DataOutputStream(new BufferedOutputStream(Files.newOutputStream(temporaryFile)));
+ currentBytes = 0;
+ }
+
+ private void sealCurrentSegment() throws IOException {
+ if (output == null) {
+ return;
+ }
+ checkTimeout();
+ long startNanos = System.nanoTime();
+ output.close();
+ output = null;
+ Path sealedFile = directory.resolve(String.format("segment-%06d.bin", nextSegmentId++));
+ try {
+ Files.move(
+ temporaryFile,
+ sealedFile,
+ StandardCopyOption.ATOMIC_MOVE,
+ StandardCopyOption.REPLACE_EXISTING);
+ } catch (IOException e) {
+ Files.move(temporaryFile, sealedFile, StandardCopyOption.REPLACE_EXISTING);
+ }
+ sealedSegments.add(sealedFile);
+ if (ioContext != null) {
+ ioContext.recordDiskIO(0, startNanos);
+ ioContext.recordSegment();
+ }
+ temporaryFile = null;
+ currentBytes = 0;
+ }
+
+ private void checkTimeout() {
+ if (ioContext != null) {
+ ioContext.checkTimeout();
+ }
+ }
+
+ private void recordDiskIO(long bytes, long startNanos) {
+ if (ioContext != null) {
+ ioContext.recordDiskIO(bytes, startNanos);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (output != null) {
+ output.close();
+ output = null;
+ }
+ if (temporaryFile != null) {
+ Files.deleteIfExists(temporaryFile);
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryFileSpillerReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryFileSpillerReader.java
new file mode 100644
index 00000000000..935c9b51b59
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryFileSpillerReader.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.db.queryengine.plan.relational.metadata.spill;
+
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+
+import java.io.BufferedInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+public final class DeviceEntryFileSpillerReader implements DeviceEntryReader {
+
+ private final List<Path> segments;
+ private final boolean deleteSegmentAfterRead;
+ private final DeviceEntryIOContext ioContext;
+ private int segmentIndex;
+ private DataInputStream input;
+ private Path currentSegment;
+ private DeviceEntry next;
+
+ public DeviceEntryFileSpillerReader(List<Path> segments) {
+ this(segments, false, null);
+ }
+
+ public DeviceEntryFileSpillerReader(List<Path> segments, boolean deleteSegmentAfterRead) {
+ this(segments, deleteSegmentAfterRead, null);
+ }
+
+ public DeviceEntryFileSpillerReader(
+ List<Path> segments, boolean deleteSegmentAfterRead, DeviceEntryIOContext ioContext) {
+ this.segments = segments;
+ this.deleteSegmentAfterRead = deleteSegmentAfterRead;
+ this.ioContext = ioContext;
+ }
+
+ @Override
+ public boolean hasNext() throws IOException {
+ if (next != null) {
+ return true;
+ }
+ while (true) {
+ if (input == null && !openNextSegment()) {
+ return false;
+ }
+ checkTimeout();
+ long startNanos = System.nanoTime();
+ Integer length = readRecordLength();
+ if (length == null) {
+ closeCurrentSegment(true);
+ continue;
+ }
+ byte[] bytes = new byte[length];
+ input.readFully(bytes);
+ if (ioContext != null) {
+ ioContext.recordDiskIO(Integer.BYTES + length, startNanos);
+ }
+ next = DeviceEntry.deserialize(bytes);
+ return true;
+ }
+ }
+
+ private void checkTimeout() {
+ if (ioContext != null) {
+ ioContext.checkTimeout();
+ }
+ }
+
+ @Override
+ public DeviceEntry next() throws IOException {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ DeviceEntry result = next;
+ next = null;
+ return result;
+ }
+
+ private boolean openNextSegment() throws IOException {
+ if (segmentIndex >= segments.size()) {
+ return false;
+ }
+ currentSegment = segments.get(segmentIndex++);
+ input = new DataInputStream(new BufferedInputStream(Files.newInputStream(currentSegment)));
+ return true;
+ }
+
+ private void closeCurrentSegment(boolean fullyConsumed) throws IOException {
+ input.close();
+ input = null;
+ if (fullyConsumed && deleteSegmentAfterRead) {
+ try {
+ Files.deleteIfExists(currentSegment);
+ } catch (IOException ignored) {
+ // Query cleanup will retry deleting a segment that could not be deleted eagerly.
+ }
+ }
+ currentSegment = null;
+ }
+
+ private Integer readRecordLength() throws IOException {
+ int firstByte = input.read();
+ if (firstByte < 0) {
+ return null;
+ }
+ int length =
+ (firstByte << 24)
+ | (input.readUnsignedByte() << 16)
+ | (input.readUnsignedByte() << 8)
+ | input.readUnsignedByte();
+ if (length < 0) {
+ throw new IOException();
+ }
+ return length;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (input != null) {
+ closeCurrentSegment(false);
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryIOContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryIOContext.java
new file mode 100644
index 00000000000..c434e0fece7
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryIOContext.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.relational.metadata.spill;
+
+import org.apache.iotdb.db.exception.query.QueryTimeoutRuntimeException;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+
+import java.util.concurrent.TimeUnit;
+
+public final class DeviceEntryIOContext {
+
+ private final MPPQueryContext queryContext;
+ private final long timeoutStartNanos;
+ private final long remainingTimeoutNanos;
+
+ public DeviceEntryIOContext(MPPQueryContext queryContext) {
+ this.queryContext = queryContext;
+ this.timeoutStartNanos = System.nanoTime();
+ long elapsedMillis = Math.max(0, System.currentTimeMillis() - queryContext.getStartTime());
+ long remainingTimeoutMillis = Math.max(0, queryContext.getTimeOut() - elapsedMillis);
+ this.remainingTimeoutNanos = TimeUnit.MILLISECONDS.toNanos(remainingTimeoutMillis);
+ }
+
+ public void checkTimeout() {
+ if (System.nanoTime() - timeoutStartNanos >= remainingTimeoutNanos) {
+ throw new QueryTimeoutRuntimeException(
+ queryContext.getStartTime(), System.currentTimeMillis(), queryContext.getTimeOut());
+ }
+ }
+
+ public void recordDiskIO(long bytes, long startNanos) {
+ queryContext.recordDeviceEntryDiskIO(bytes, System.nanoTime() - startNanos);
+ checkTimeout();
+ }
+
+ public void recordSegment() {
+ queryContext.recordDeviceEntrySegment();
+ }
+
+ public void recordSortedRun() {
+ queryContext.recordDeviceEntrySortedRun();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializer.java
new file mode 100644
index 00000000000..92d7cf64d48
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializer.java
@@ -0,0 +1,132 @@
+/*
+ * 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.metadata.spill;
+
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+
+import java.io.IOException;
+import java.nio.file.Path;
+
+public final class DeviceEntryMaterializer extends AbstractDeviceEntryMaterializer {
+
+ private final boolean rawSegment;
+ private DeviceEntryDiskSpiller spiller;
+ // Only be used in fetchDeviceSchema, manages memory itself
+ private long rawBufferedRamBytes;
+
+ public DeviceEntryMaterializer(
+ String queryId, PlanNodeId planNodeId, long thresholdInBytes, boolean rawSegment) {
+ super(queryId, planNodeId, thresholdInBytes);
+ this.rawSegment = rawSegment;
+ }
+
+ public DeviceEntryMaterializer(
+ String queryId,
+ PlanNodeId planNodeId,
+ long thresholdInBytes,
+ boolean rawSegment,
+ MPPQueryContext queryContext) {
+ this(queryId, planNodeId, thresholdInBytes, rawSegment);
+ setQueryContext(queryContext);
+ }
+
+ @Override
+ public void append(DeviceEntry entry) throws IOException {
+ checkNotFinished();
+ appendToBuffer(entry);
+ }
+
+ /** Returns the RAM bytes released when Coordinator Raw Fetch switches to spill mode. */
+ public long appendWithMemoryControl(DeviceEntry entry) throws IOException {
+ checkNotFinished();
+ long ramBytesUsed = entry.ramBytesUsed();
+ if (spiller == null && rawBufferedRamBytes + ramBytesUsed <= thresholdInBytes()) {
+ appendToBuffer(entry);
+ rawBufferedRamBytes += ramBytesUsed;
+ return 0;
+ }
+ long releasedRamBytes = rawBufferedRamBytes;
+ ensureSpiller();
+ rawBufferedRamBytes = 0;
+ spiller.append(entry.serializeToBytes());
+ incrementEntryCount();
+ return releasedRamBytes;
+ }
+
+ @Override
+ public void forceSpill() throws IOException {
+ checkNotFinished();
+ if (spiller == null && !isBufferEmpty()) {
+ ensureSpiller();
+ }
+ rawBufferedRamBytes = 0;
+ }
+
+ public boolean isSpilled() {
+ return spiller != null;
+ }
+
+ @Override
+ public DeviceEntryDataSet finish() throws IOException {
+ checkNotFinished();
+ DeviceEntryDataSet dataSet;
+ if (spiller == null) {
+ dataSet = new InMemoryDeviceEntryDataSet(copyBufferedEntries());
+ } else {
+ dataSet =
+ new SpilledDeviceEntryDataSet(
+ queryId(), ownerDirectory(), spiller.finish(), entryCount(), !rawSegment);
+ }
+ markFinished();
+ return dataSet;
+ }
+
+ private void ensureSpiller() throws IOException {
+ if (spiller != null) {
+ return;
+ }
+ Path ownerDirectory =
+ rawSegment
+ ? ensureUnregisteredOwnerDirectory(
+ Path.of(IoTDBDescriptor.getInstance().getConfig().getQueryDir(), "device-entry"))
+ : ensureOwnerDirectory();
+ spiller =
+ new DeviceEntryDiskSpiller(
+ ownerDirectory.resolve(rawSegment ? "raw" : "fi"), thresholdInBytes(), ioContext());
+ for (DeviceEntry entry : bufferedEntries()) {
+ spiller.append(entry.serializeToBytes());
+ }
+ clearBuffer();
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ if (spiller != null) {
+ spiller.close();
+ }
+ } finally {
+ super.close();
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryReader.java
new file mode 100644
index 00000000000..5fa2e044974
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryReader.java
@@ -0,0 +1,34 @@
+/*
+ * 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.metadata.spill;
+
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+
+import java.io.IOException;
+
+public interface DeviceEntryReader extends AutoCloseable {
+
+ public boolean hasNext() throws IOException;
+
+ public DeviceEntry next() throws IOException;
+
+ @Override
+ public void close() throws IOException;
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntrySpillManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntrySpillManager.java
new file mode 100644
index 00000000000..162cae055dc
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntrySpillManager.java
@@ -0,0 +1,189 @@
+/*
+ * 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.metadata.spill;
+
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+
+import org.apache.tsfile.external.commons.io.FileUtils;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+public final class DeviceEntrySpillManager {
+
+ private final ConcurrentHashMap<String, Set<Path>> queryDirectories = new ConcurrentHashMap<>();
+
+ private DeviceEntrySpillManager() {}
+
+ public static DeviceEntrySpillManager getInstance() {
+ return DeviceEntrySpillManagerHolder.INSTANCE;
+ }
+
+ public Path register(String queryId, PlanNodeId planNodeId) throws IOException {
+ Path ownerDirectory = rootDirectory().resolve(queryId).resolve(planNodeId.getId());
+ Files.createDirectories(ownerDirectory);
+ queryDirectories
+ .computeIfAbsent(queryId, ignored -> ConcurrentHashMap.newKeySet())
+ .add(ownerDirectory);
+ return ownerDirectory;
+ }
+
+ public void deregisterOwner(String queryId, Path ownerDirectory) throws IOException {
+ Set<Path> owners = queryDirectories.get(queryId);
+ if (owners != null) {
+ owners.remove(ownerDirectory);
+ if (owners.isEmpty()) {
+ queryDirectories.remove(queryId, owners);
+ }
+ }
+ FileUtils.deleteDirectory(ownerDirectory.toFile());
+ }
+
+ public void deregisterQuery(String queryId) throws IOException {
+ queryDirectories.remove(queryId);
+ FileUtils.deleteDirectory(rootDirectory().resolve(queryId).toFile());
+ }
+
+ @TestOnly
+ public List<Path> listSegments(String queryId, String planNodeId) throws IOException {
+ Path dataSetDirectory = resolveRegisteredDataSetDirectory(queryId, planNodeId);
+ try (java.util.stream.Stream<Path> stream = Files.list(dataSetDirectory)) {
+ return stream
+ .filter(path -> path.getFileName().toString().matches("segment-[0-9]{6,}\\.bin"))
+ .sorted(
+ Comparator.comparingInt((Path path) -> path.getFileName().toString().length())
+ .thenComparing(path -> path.getFileName().toString()))
+ .toList();
+ }
+ }
+
+ public byte[] readSegment(String queryId, String dataSetId, int segmentId) throws IOException {
+ return Files.readAllBytes(resolveSegment(queryId, dataSetId, segmentId));
+ }
+
+ public Path resolveSegment(String queryId, String dataSetId, int segmentId) throws IOException {
+ Path segment = getRegisteredSegmentPath(queryId, dataSetId, segmentId);
+ if (!Files.isRegularFile(segment)) {
+ throw new java.nio.file.NoSuchFileException(segment.toString());
+ }
+ return segment;
+ }
+
+ public Path resolveSegment(String queryId, PlanNodeId planNodeId, int segmentId)
+ throws IOException {
+ return resolveSegment(queryId, planNodeId.getId(), segmentId);
+ }
+
+ public void deleteSegment(String queryId, String dataSetId, int segmentId) throws IOException {
+ Files.deleteIfExists(getRegisteredSegmentPath(queryId, dataSetId, segmentId));
+ }
+
+ public void deleteSegment(String queryId, PlanNodeId planNodeId, int segmentId)
+ throws IOException {
+ deleteSegment(queryId, planNodeId.getId(), segmentId);
+ }
+
+ public void finishSegmentDataSet(String queryId, String planNodeId) throws IOException {
+ deregisterOwner(queryId, rootDirectory().resolve(queryId).resolve(planNodeId));
+ }
+
+ public void deregisterFragment(String queryId, String fragmentInstanceId) throws IOException {
+ FileUtils.deleteDirectory(
+ resolveUnderRoot(fragmentRootDirectory(), queryId, fragmentInstanceId).toFile());
+ }
+
+ public void clearStaleFragmentData() throws IOException {
+ FileUtils.deleteDirectory(fragmentRootDirectory().toFile());
+ Files.createDirectories(fragmentRootDirectory());
+ }
+
+ private Path resolveRegisteredDataSetDirectory(String queryId, String dataSetId)
+ throws IOException {
+ Path relativeDataSetPath = Path.of(dataSetId);
+ if (relativeDataSetPath.isAbsolute()
+ || java.util.stream.StreamSupport.stream(relativeDataSetPath.spliterator(), false)
+ .anyMatch(path -> path.toString().equals("..") || path.toString().equals("."))) {
+ throw new IllegalArgumentException();
+ }
+ Path queryDirectory = rootDirectory().resolve(queryId).normalize();
+ Path dataSetDirectory = queryDirectory.resolve(relativeDataSetPath).resolve("fi").normalize();
+ if (!dataSetDirectory.startsWith(queryDirectory)) {
+ throw new IllegalArgumentException();
+ }
+ Set<Path> owners = queryDirectories.get(queryId);
+ boolean registered =
+ owners != null
+ && owners.stream()
+ .map(Path::normalize)
+ .anyMatch(owner -> dataSetDirectory.startsWith(owner) && Files.isDirectory(owner));
+ if (!registered || !Files.isDirectory(dataSetDirectory)) {
+ throw new java.nio.file.NoSuchFileException(dataSetDirectory.toString());
+ }
+ return dataSetDirectory;
+ }
+
+ private Path getRegisteredSegmentPath(String queryId, String dataSetId, int segmentId)
+ throws IOException {
+ if (segmentId < 0) {
+ throw new IllegalArgumentException();
+ }
+ return resolveRegisteredDataSetDirectory(queryId, dataSetId)
+ .resolve(String.format("segment-%06d.bin", segmentId));
+ }
+
+ public void clearStaleData() throws IOException {
+ FileUtils.deleteDirectory(rootDirectory().toFile());
+ Files.createDirectories(rootDirectory());
+ queryDirectories.clear();
+ }
+
+ private Path rootDirectory() {
+ return Path.of(IoTDBDescriptor.getInstance().getConfig().getQueryDir(), "device-entry");
+ }
+
+ private Path fragmentRootDirectory() {
+ return rootDirectory().resolve("fragment");
+ }
+
+ private Path resolveUnderRoot(Path root, String... children) {
+ Path result = root;
+ for (String child : children) {
+ result = result.resolve(child);
+ }
+ result = result.normalize();
+ if (!result.startsWith(root.normalize())) {
+ throw new IllegalArgumentException();
+ }
+ return result;
+ }
+
+ private static class DeviceEntrySpillManagerHolder {
+ private static final DeviceEntrySpillManager INSTANCE = new DeviceEntrySpillManager();
+
+ private DeviceEntrySpillManagerHolder() {}
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/InMemoryDeviceEntryDataSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/InMemoryDeviceEntryDataSet.java
new file mode 100644
index 00000000000..831cbe351f3
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/InMemoryDeviceEntryDataSet.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.queryengine.plan.relational.metadata.spill;
+
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+public final class InMemoryDeviceEntryDataSet implements DeviceEntryDataSet {
+
+ private final List<DeviceEntry> entries;
+
+ public InMemoryDeviceEntryDataSet(List<DeviceEntry> entries) {
+ this.entries = Collections.unmodifiableList(entries);
+ }
+
+ @Override
+ public long getEntryCount() {
+ return entries.size();
+ }
+
+ @Override
+ public boolean isSpilled() {
+ return false;
+ }
+
+ @Override
+ public DeviceEntryReader openReader() {
+ Iterator<DeviceEntry> iterator = entries.iterator();
+ return new DeviceEntryReader() {
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public DeviceEntry next() {
+ return iterator.next();
+ }
+
+ @Override
+ public void close() {
+ // No resource to release.
+ }
+ };
+ }
+
+ @Override
+ public List<DeviceEntry> getInlineEntries() {
+ return entries;
+ }
+
+ @Override
+ public void close() {
+ // The query context owns memory accounting for inline entries.
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/SpilledDeviceEntryDataSet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/SpilledDeviceEntryDataSet.java
new file mode 100644
index 00000000000..0380f210f34
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/SpilledDeviceEntryDataSet.java
@@ -0,0 +1,85 @@
+/*
+ * 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.metadata.spill;
+
+import org.apache.tsfile.external.commons.io.FileUtils;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+
+public final class SpilledDeviceEntryDataSet implements DeviceEntryDataSet {
+
+ private final String queryId;
+ private final Path ownerDirectory;
+ private final List<Path> segments;
+ private final long entryCount;
+ private final boolean managedBySpillManager;
+
+ public SpilledDeviceEntryDataSet(
+ String queryId,
+ Path ownerDirectory,
+ List<Path> segments,
+ long entryCount,
+ boolean managedBySpillManager) {
+ this.queryId = queryId;
+ this.ownerDirectory = ownerDirectory;
+ this.segments = segments;
+ this.entryCount = entryCount;
+ this.managedBySpillManager = managedBySpillManager;
+ }
+
+ @Override
+ public long getEntryCount() {
+ return entryCount;
+ }
+
+ @Override
+ public boolean isSpilled() {
+ return true;
+ }
+
+ public Path getOwnerDirectory() {
+ return ownerDirectory;
+ }
+
+ public List<Path> getSegments() {
+ return segments;
+ }
+
+ @Override
+ public DeviceEntryReader openReader() {
+ return new DeviceEntryFileSpillerReader(segments);
+ }
+
+ @Override
+ public DeviceEntryReader openConsumingReader() {
+ return new DeviceEntryFileSpillerReader(segments, true);
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (managedBySpillManager) {
+ DeviceEntrySpillManager.getInstance().deregisterOwner(queryId, ownerDirectory);
+ } else {
+ FileUtils.deleteDirectory(ownerDirectory.toFile());
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
index 1ceb4c86f6c..e74b6bb3a0f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
@@ -70,6 +70,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.analyzer.predicate.schema
import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.NonAlignedDeviceEntry;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSet;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryDataSetResult;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryReader;
import org.apache.iotdb.db.queryengine.plan.relational.planner.EqualityInference;
import org.apache.iotdb.db.queryengine.plan.relational.planner.IrExpressionInterpreter;
import org.apache.iotdb.db.queryengine.plan.relational.planner.IrTypeAnalyzer;
@@ -91,6 +94,8 @@ import org.apache.tsfile.utils.Pair;
import javax.annotation.Nullable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -746,7 +751,7 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
}
long startTime = System.nanoTime();
- final Map<String, List<DeviceEntry>> deviceEntriesMap =
+ final DeviceEntryDataSetResult deviceEntryDataSetResult =
metadata.indexScan(
tableScanNode.getQualifiedObjectName(),
metadataExpressions.stream()
@@ -756,7 +761,10 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
expression, tableScanNode.getAssignments()))
.collect(Collectors.toList()),
attributeColumns,
- queryContext);
+ queryContext,
+ tableScanNode.getPlanNodeId());
+ final Map<String, List<DeviceEntry>> deviceEntriesMap =
+ readDeviceEntries(deviceEntryDataSetResult);
if (deviceEntriesMap.size() > 1) {
throw new SemanticException(
DataNodeQueryMessages.TREE_DEVICE_VIEW_WITH_MULTIPLE_DATABASES
@@ -837,6 +845,20 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
}
}
+ private Map<String, List<DeviceEntry>> readDeviceEntries(
+ final DeviceEntryDataSetResult result) {
+ final List<DeviceEntry> entries = new ArrayList<>();
+ try (DeviceEntryDataSet dataSet = result.getDataSet();
+ DeviceEntryReader reader = dataSet.openReader()) {
+ while (reader.hasNext()) {
+ entries.add(reader.next());
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ return Collections.singletonMap(result.getDatabase(), entries);
+ }
+
@Override
public PlanNode visitJoin(JoinNode node, RewriteContext context) {
Expression inheritedPredicate =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java
index 02999d57b01..da6f415af8d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsDrawer.java
@@ -53,6 +53,25 @@ public class FragmentInstanceStatisticsDrawer {
0,
String.format(
"Fetch Schema Cost: %.3f ms", context.getFetchSchemaCost() * NS_TO_MS_FACTOR));
+ addLine(
+ planHeader,
+ 1,
+ String.format(
+ "Disk IO Size for DeviceEntry: %d bytes", context.getDiskIOSizeForDeviceEntry()));
+ addLine(
+ planHeader,
+ 1,
+ String.format(
+ "Disk IO Time Cost for DeviceEntry: %.3f ms",
+ context.getDiskIOTimeCostForDeviceEntry() * NS_TO_MS_FACTOR));
+ addLine(
+ planHeader,
+ 1,
+ String.format("DeviceEntry Segment Count: %d", context.getDeviceEntrySegmentCount()));
+ addLine(
+ planHeader,
+ 1,
+ String.format("DeviceEntry Sorted Run Count: %d", context.getDeviceEntrySortedRunCount()));
addLine(
planHeader,
0,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java
index 74571ad152f..9e727cd7517 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawer.java
@@ -55,6 +55,12 @@ public class FragmentInstanceStatisticsJsonDrawer {
"fetchPartitionCostMs", formatMs(context.getFetchPartitionCost() * NS_TO_MS_FACTOR));
planStatistics.addProperty(
"fetchSchemaCostMs", formatMs(context.getFetchSchemaCost() * NS_TO_MS_FACTOR));
+ planStatistics.addProperty("diskIOSizeForDeviceEntry", context.getDiskIOSizeForDeviceEntry());
+ planStatistics.addProperty(
+ "diskIOTimeCostForDeviceEntryMs",
+ formatMs(context.getDiskIOTimeCostForDeviceEntry() * NS_TO_MS_FACTOR));
+ planStatistics.addProperty("deviceEntrySegmentCount", context.getDeviceEntrySegmentCount());
+ planStatistics.addProperty("deviceEntrySortedRunCount", context.getDeviceEntrySortedRunCount());
planStatistics.addProperty(
"logicalPlanCostMs", formatMs(context.getLogicalPlanCost() * NS_TO_MS_FACTOR));
planStatistics.addProperty(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java
index edb13217db2..592d45d2e63 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/QueryPlanStatistics.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.queryengine.statistics;
+import java.util.concurrent.atomic.AtomicLong;
+
public class QueryPlanStatistics {
private long analyzeCost;
private long fetchPartitionCost;
@@ -27,6 +29,12 @@ public class QueryPlanStatistics {
private long logicalOptimizationCost;
private long distributionPlanCost;
private long dispatchCost = 0;
+ // DeviceEntry materialization may involve multiple Region materializers. Use the same
+ // lock-free accumulation style as execution.fragment.QueryStatistics.
+ private final AtomicLong diskIOSizeForDeviceEntry = new AtomicLong();
+ private final AtomicLong diskIOTimeCostForDeviceEntry = new AtomicLong();
+ private final AtomicLong deviceEntrySegmentCount = new AtomicLong();
+ private final AtomicLong deviceEntrySortedRunCount = new AtomicLong();
public void setAnalyzeCost(long analyzeCost) {
this.analyzeCost = analyzeCost;
@@ -83,4 +91,33 @@ public class QueryPlanStatistics {
public long getDispatchCost() {
return dispatchCost;
}
+
+ public void recordDeviceEntryDiskIO(long bytes, long timeCost) {
+ diskIOSizeForDeviceEntry.addAndGet(bytes);
+ diskIOTimeCostForDeviceEntry.addAndGet(timeCost);
+ }
+
+ public void recordDeviceEntrySegment() {
+ deviceEntrySegmentCount.incrementAndGet();
+ }
+
+ public void recordDeviceEntrySortedRun() {
+ deviceEntrySortedRunCount.incrementAndGet();
+ }
+
+ public long getDiskIOSizeForDeviceEntry() {
+ return diskIOSizeForDeviceEntry.get();
+ }
+
+ public long getDiskIOTimeCostForDeviceEntry() {
+ return diskIOTimeCostForDeviceEntry.get();
+ }
+
+ public long getDeviceEntrySegmentCount() {
+ return deviceEntrySegmentCount.get();
+ }
+
+ public long getDeviceEntrySortedRunCount() {
+ return deviceEntrySortedRunCount.get();
+ }
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializerTest.java
new file mode 100644
index 00000000000..9f13e8bd3c7
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/spill/DeviceEntryMaterializerTest.java
@@ -0,0 +1,153 @@
+/*
+ * 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.metadata.spill;
+
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.AlignedDeviceEntry;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+
+import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.utils.Binary;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class DeviceEntryMaterializerTest {
+
+ private Path queryDirectory;
+
+ @Before
+ public void setUp() throws Exception {
+ queryDirectory = Files.createTempDirectory("device-entry-spill-test");
+ IoTDBDescriptor.getInstance().getConfig().setQueryDir(queryDirectory.toString());
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ DeviceEntrySpillManager.getInstance().clearStaleData();
+ Files.deleteIfExists(queryDirectory.resolve("device-entry"));
+ Files.deleteIfExists(queryDirectory);
+ }
+
+ @Test
+ public void testKeepSmallDataSetInline() throws Exception {
+ List<DeviceEntry> expected = createEntries(3);
+ try (DeviceEntryMaterializer materializer =
+ new DeviceEntryMaterializer("q-inline", new PlanNodeId("scan-0"), Long.MAX_VALUE, true)) {
+ for (DeviceEntry entry : expected) {
+ materializer.append(entry);
+ }
+ try (DeviceEntryDataSet dataSet = materializer.finish()) {
+ assertFalse(dataSet.isSpilled());
+ assertEquals(expected, dataSet.getInlineEntries());
+ }
+ }
+ }
+
+ @Test
+ public void testSpillAndReadMultipleSegments() throws Exception {
+ List<DeviceEntry> expected = createEntries(20);
+ DeviceEntryDataSet dataSet;
+ try (DeviceEntryMaterializer materializer =
+ new DeviceEntryMaterializer("q-spill", new PlanNodeId("scan-0"), 128, true)) {
+ for (DeviceEntry entry : expected) {
+ materializer.append(entry);
+ }
+ dataSet = materializer.finish();
+ }
+
+ assertTrue(dataSet.isSpilled());
+ assertEquals(expected.size(), dataSet.getEntryCount());
+ List<DeviceEntry> actual = new ArrayList<>();
+ try (DeviceEntryReader reader = dataSet.openReader()) {
+ while (reader.hasNext()) {
+ actual.add(reader.next());
+ }
+ }
+ assertEquals(expected, actual);
+
+ dataSet.close();
+ assertFalse(Files.exists(queryDirectory.resolve("device-entry/q-spill/scan-0")));
+ }
+
+ @Test
+ public void testControlledSegmentAccess() throws Exception {
+ DeviceEntryDataSet dataSet;
+ try (DeviceEntryMaterializer materializer =
+ new DeviceEntryMaterializer("q-segment", new PlanNodeId("scan-0"), 128, true)) {
+ for (DeviceEntry entry : createEntries(20)) {
+ materializer.append(entry);
+ }
+ dataSet = materializer.finish();
+ }
+
+ DeviceEntrySpillManager manager = DeviceEntrySpillManager.getInstance();
+ Path rawDirectory = queryDirectory.resolve("device-entry/q-segment/scan-0/raw");
+ List<Path> segments;
+ try (java.util.stream.Stream<Path> stream = Files.list(rawDirectory)) {
+ segments =
+ stream.filter(path -> path.getFileName().toString().endsWith(".bin")).sorted().toList();
+ }
+ assertTrue(segments.size() > 1);
+ assertTrue(Files.size(segments.get(0)) > 0);
+ dataSet.close();
+ }
+
+ @Test
+ public void testSpillFileUsesLengthPrefixedRecordsWithoutHeaderOrCrc() throws Exception {
+ DeviceEntry entry = createEntries(1).get(0);
+ DeviceEntryDataSet dataSet;
+ try (DeviceEntryMaterializer materializer =
+ new DeviceEntryMaterializer("q-format", new PlanNodeId("scan-0"), 1, true)) {
+ materializer.append(entry);
+ dataSet = materializer.finish();
+ }
+
+ Path segment =
+ DeviceEntrySpillManager.getInstance().listSegments("q-format", "scan-0/raw").get(0);
+ byte[] fileBytes = Files.readAllBytes(segment);
+ byte[] payload = entry.serializeToBytes();
+ assertEquals(Integer.BYTES + payload.length, fileBytes.length);
+ assertEquals(payload.length, ByteBuffer.wrap(fileBytes).getInt());
+ dataSet.close();
+ }
+
+ private static List<DeviceEntry> createEntries(int count) {
+ List<DeviceEntry> entries = new ArrayList<>(count);
+ for (int i = 0; i < count; i++) {
+ entries.add(
+ new AlignedDeviceEntry(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table", "device" + i}),
+ new Binary[] {new Binary(("attribute" + i).getBytes(TSFileConfig.STRING_CHARSET))}));
+ }
+ return entries;
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java
index b83501a277c..ae76d47b3fe 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/statistics/FragmentInstanceStatisticsJsonDrawerTest.java
@@ -60,6 +60,10 @@ public class FragmentInstanceStatisticsJsonDrawerTest {
context.setLogicalPlanCost(4000000L); // 4ms
context.setLogicalOptimizationCost(5000000L); // 5ms
context.setDistributionPlanCost(6000000L); // 6ms
+ context.recordDeviceEntryDiskIO(8192L, 7000000L); // 8 KiB, 7ms
+ context.recordDeviceEntrySegment();
+ context.recordDeviceEntrySegment();
+ context.recordDeviceEntrySortedRun();
drawer.renderPlanStatistics(context);
@@ -78,6 +82,10 @@ public class FragmentInstanceStatisticsJsonDrawerTest {
assertEquals(4.0, planStats.get("logicalPlanCostMs").getAsDouble(), 0.01);
assertEquals(5.0, planStats.get("logicalOptimizationCostMs").getAsDouble(), 0.01);
assertEquals(6.0, planStats.get("distributionPlanCostMs").getAsDouble(), 0.01);
+ assertEquals(8192L, planStats.get("diskIOSizeForDeviceEntry").getAsLong());
+ assertEquals(7.0, planStats.get("diskIOTimeCostForDeviceEntryMs").getAsDouble(), 0.01);
+ assertEquals(2L, planStats.get("deviceEntrySegmentCount").getAsLong());
+ assertEquals(1L, planStats.get("deviceEntrySortedRunCount").getAsLong());
}
@Test
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 c4a06b68b14..81573e47f54 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
@@ -2445,3 +2445,7 @@ enable_retry_for_unknown_error=false
# effectiveMode: hot_reload
# Datatype: Boolean
include_null_value_in_write_throughput_metric=false
+# Maximum DeviceEntry bytes retained in memory before spilling for a table query.
+# <= 0 uses query execution memory / query_thread_count / 4.
+# Datatype: long; effectiveMode: hot_reload; unit: byte.
+# table_query_device_entry_batch_size_in_bytes=0