This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 012026f0358 Add DEVICE_METADATA_INFO_SWAP_THRESHOLD parameter to
read_tsfile table function (#18146)
012026f0358 is described below
commit 012026f0358cc05f983188efc9fb6552008d4269
Author: shuwenwei <[email protected]>
AuthorDate: Wed Jul 8 16:13:56 2026 +0800
Add DEVICE_METADATA_INFO_SWAP_THRESHOLD parameter to read_tsfile table
function (#18146)
---
.../iotdb/db/i18n/DataNodeQueryMessages.java | 2 +
.../iotdb/db/i18n/DataNodeQueryMessages.java | 2 +
.../db/queryengine/common/MPPQueryContext.java | 8 ++-
.../read_tsfile/ExternalTsFileQueryResource.java | 9 ++--
.../tvf/read_tsfile/ReadTsFileTableFunction.java | 60 ++++++++++++++++++++--
.../plan/relational/planner/RelationPlanner.java | 5 +-
.../ExternalTsFileQueryResourceTest.java | 7 ++-
7 files changed, 81 insertions(+), 12 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index 7978fce5113..f32079208fd 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -868,6 +868,8 @@ public final class DataNodeQueryMessages {
"Invalid scalar argument: ";
public static final String ARGUMENT_SHOULD_BE_A_STRING =
"Argument %s should be a string";
+ public static final String ARGUMENT_SHOULD_BE_A_NUMBER =
+ "Argument %s should be a number";
public static final String ARGUMENT_SHOULD_CONTAIN_AT_LEAST_ONE_PATH =
"Argument %s should contain at least one path";
public static final String READ_TSFILE_PATH_IS_NOT_ALLOWED =
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index 34a7fa084bf..9e4a0188a11 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -852,6 +852,8 @@ public final class DataNodeQueryMessages {
"无效的标量参数：";
public static final String ARGUMENT_SHOULD_BE_A_STRING =
"参数 %s 应为字符串";
+ public static final String ARGUMENT_SHOULD_BE_A_NUMBER =
+ "参数 %s 应为数字";
public static final String ARGUMENT_SHOULD_CONTAIN_AT_LEAST_ONE_PATH =
"参数 %s 应至少包含一个路径";
public static final String READ_TSFILE_PATH_IS_NOT_ALLOWED =
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 6412d334903..d3926bcac79 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
@@ -291,7 +291,10 @@ public class MPPQueryContext implements IAuditEntity {
}
public ExternalTsFileQueryResource createExternalTsFileQueryResource(
- String tableName, List<String> tsFilePaths, Map<Symbol, ColumnSchema>
tableColumnSchema) {
+ String tableName,
+ List<String> tsFilePaths,
+ Map<Symbol, ColumnSchema> tableColumnSchema,
+ long deviceMetadataInfoSwapThreshold) {
int resourceIndex = externalTsFileQueryResourceIndex.getAndIncrement();
ExternalTsFileQueryResource externalTsFileQueryResource =
new ExternalTsFileQueryResource(
@@ -302,7 +305,8 @@ public class MPPQueryContext implements IAuditEntity {
.resolve(String.valueOf(resourceIndex)),
tableName,
tsFilePaths,
- tableColumnSchema);
+ tableColumnSchema,
+ deviceMetadataInfoSwapThreshold);
externalTsFileQueryResources.add(externalTsFileQueryResource);
return externalTsFileQueryResource;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java
index 75249c1c1fa..d16e602e3d2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResource.java
@@ -88,6 +88,7 @@ public class ExternalTsFileQueryResource {
private final String tableName;
private final List<String> tsFilePaths;
private final Map<Symbol, ColumnSchema> tableColumnSchema;
+ private final long deviceMetadataInfoSwapThreshold;
private final List<TsFileResource> sharedTsFileResources;
private final List<DeviceEntry> sharedDeviceEntries = new ArrayList<>();
private final List<DeviceTaskPartition> deviceTaskPartitions = new
ArrayList<>();
@@ -105,7 +106,8 @@ public class ExternalTsFileQueryResource {
Path tempRoot,
String tableName,
List<String> tsFilePaths,
- Map<Symbol, ColumnSchema> tableColumnSchema) {
+ Map<Symbol, ColumnSchema> tableColumnSchema,
+ long deviceMetadataInfoSwapThreshold) {
this.queryContext = requireNonNull(queryContext, "queryContext is null");
this.queryId = queryContext.getQueryId();
this.externalTsFileResourceMemoryReservationManager =
@@ -115,6 +117,7 @@ public class ExternalTsFileQueryResource {
this.tableName = tableName;
this.tsFilePaths = requireNonNull(tsFilePaths, "tsFilePaths");
this.tableColumnSchema = tableColumnSchema;
+ this.deviceMetadataInfoSwapThreshold = deviceMetadataInfoSwapThreshold;
this.sharedTsFileResources = createTsFileResources(this.tsFilePaths);
for (String tsFilePath : tsFilePaths) {
FileReaderManager.getInstance().increaseExternalFileReaderReference(tsFilePath);
@@ -280,7 +283,6 @@ public class ExternalTsFileQueryResource {
public class DeviceTaskPartition {
- private static final long DEVICE_TASK_BUCKET_TARGET_SIZE_IN_BYTES = 8L *
1024 * 1024;
private static final long MEMORY_RESERVE_BATCH_SIZE_IN_BYTES = 1024 * 1024;
private final int partitionIndex;
@@ -341,7 +343,8 @@ public class ExternalTsFileQueryResource {
}
private boolean shouldFlush() {
- if (getPendingMemoryBytes() >= DEVICE_TASK_BUCKET_TARGET_SIZE_IN_BYTES) {
+ if (getPendingMemoryBytes()
+ >= ExternalTsFileQueryResource.this.deviceMetadataInfoSwapThreshold)
{
return true;
}
if (unreservedBytes < MEMORY_RESERVE_BATCH_SIZE_IN_BYTES) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ReadTsFileTableFunction.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ReadTsFileTableFunction.java
index d141d0112c8..71ee6aa267f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ReadTsFileTableFunction.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ReadTsFileTableFunction.java
@@ -53,8 +53,11 @@ import java.util.stream.Collectors;
/** Reads one or more TsFiles as a table function source. */
public class ReadTsFileTableFunction implements TableFunction {
+ static final long DEVICE_TASK_BUCKET_TARGET_SIZE_IN_BYTES = 8L * 1024 * 1024;
private static final String TABLE_NAME_PARAMETER_NAME = "TABLE_NAME";
private static final String PATHS_PARAMETER_NAME = "PATHS";
+ private static final String
DEVICE_METADATA_INFO_SWAP_THRESHOLD_PARAMETER_NAME =
+ "DEVICE_METADATA_INFO_SWAP_THRESHOLD";
private MPPQueryContext mppQueryContext;
@@ -70,6 +73,11 @@ public class ReadTsFileTableFunction implements
TableFunction {
.name(TABLE_NAME_PARAMETER_NAME)
.type(Type.STRING)
.defaultValue("")
+ .build(),
+ ScalarParameterSpecification.builder()
+ .name(DEVICE_METADATA_INFO_SWAP_THRESHOLD_PARAMETER_NAME)
+ .type(Type.INT64)
+ .defaultValue(DEVICE_TASK_BUCKET_TARGET_SIZE_IN_BYTES)
.build());
}
@@ -92,6 +100,15 @@ public class ReadTsFileTableFunction implements
TableFunction {
}
DescribedSchema outputSchema = convertToDescribedSchema(mergedTableSchema);
+ long deviceMetadataInfoSwapThreshold =
+ getOptionalLongArgument(
+ arguments,
+ DEVICE_METADATA_INFO_SWAP_THRESHOLD_PARAMETER_NAME,
+ DEVICE_TASK_BUCKET_TARGET_SIZE_IN_BYTES);
+ if (deviceMetadataInfoSwapThreshold <= 0) {
+ deviceMetadataInfoSwapThreshold =
DEVICE_TASK_BUCKET_TARGET_SIZE_IN_BYTES;
+ }
+
ReadTsFileTableFunctionHandle handle =
new ReadTsFileTableFunctionHandle(
schemaCollector.getTableName(),
@@ -101,7 +118,8 @@ public class ReadTsFileTableFunction implements
TableFunction {
mergedTableSchema.getColumnTypes().stream()
.map(TsTableColumnCategory::fromTsFileColumnCategory)
.collect(Collectors.toList()),
- outputSchema);
+ outputSchema,
+ deviceMetadataInfoSwapThreshold);
return
TableFunctionAnalysis.builder().properColumnSchema(outputSchema).handle(handle).build();
}
@@ -147,6 +165,26 @@ public class ReadTsFileTableFunction implements
TableFunction {
return ((String) value).trim();
}
+ private static long getOptionalLongArgument(
+ Map<String, Argument> arguments, String name, long defaultValue) {
+ Argument argument = arguments.get(name);
+ if (argument == null) {
+ return defaultValue;
+ }
+ if (!(argument instanceof ScalarArgument)) {
+ throw new
UDFArgumentNotValidException(DataNodeQueryMessages.INVALID_SCALAR_ARGUMENT +
name);
+ }
+ Object value = ((ScalarArgument) argument).getValue();
+ if (value instanceof Long) {
+ return (Long) value;
+ }
+ if (value instanceof Integer) {
+ return ((Integer) value).longValue();
+ }
+ throw new UDFArgumentNotValidException(
+ String.format(DataNodeQueryMessages.ARGUMENT_SHOULD_BE_A_NUMBER,
name));
+ }
+
private static List<String> parseTsFilePaths(String tsFilePaths) {
List<String> paths =
Arrays.stream(tsFilePaths.split(","))
@@ -204,6 +242,7 @@ public class ReadTsFileTableFunction implements
TableFunction {
private List<String> outputColumnNames;
private List<Type> outputColumnTypes;
private List<TsTableColumnCategory> outputColumnCategories;
+ private long deviceMetadataInfoSwapThreshold;
public ReadTsFileTableFunctionHandle() {
this(
@@ -211,14 +250,16 @@ public class ReadTsFileTableFunction implements
TableFunction {
Collections.emptyList(),
Collections.emptyList(),
Collections.emptyList(),
- Collections.emptyList());
+ Collections.emptyList(),
+ DEVICE_TASK_BUCKET_TARGET_SIZE_IN_BYTES);
}
public ReadTsFileTableFunctionHandle(
String tableName,
List<String> tsFilePaths,
List<TsTableColumnCategory> outputColumnCategories,
- DescribedSchema outputSchema) {
+ DescribedSchema outputSchema,
+ long deviceMetadataInfoSwapThreshold) {
this(
tableName,
tsFilePaths,
@@ -228,7 +269,8 @@ public class ReadTsFileTableFunction implements
TableFunction {
outputSchema.getFields().stream()
.map(DescribedSchema.Field::getType)
.collect(Collectors.toList()),
- outputColumnCategories);
+ outputColumnCategories,
+ deviceMetadataInfoSwapThreshold);
}
private ReadTsFileTableFunctionHandle(
@@ -236,7 +278,8 @@ public class ReadTsFileTableFunction implements
TableFunction {
List<String> tsFilePaths,
List<String> outputColumnNames,
List<Type> outputColumnTypes,
- List<TsTableColumnCategory> outputColumnCategories) {
+ List<TsTableColumnCategory> outputColumnCategories,
+ long deviceMetadataInfoSwapThreshold) {
if (outputColumnNames.size() != outputColumnTypes.size()) {
throw new IllegalArgumentException(
DataNodeQueryMessages.OUTPUT_COLUMN_NAMES_AND_TYPES_SIZE_MISMATCH);
@@ -251,6 +294,7 @@ public class ReadTsFileTableFunction implements
TableFunction {
this.outputColumnTypes = Collections.unmodifiableList(new
ArrayList<>(outputColumnTypes));
this.outputColumnCategories =
Collections.unmodifiableList(new
ArrayList<>(outputColumnCategories));
+ this.deviceMetadataInfoSwapThreshold = deviceMetadataInfoSwapThreshold;
}
public String getTableName() {
@@ -273,6 +317,10 @@ public class ReadTsFileTableFunction implements
TableFunction {
return outputColumnCategories;
}
+ public long getDeviceMetadataInfoSwapThreshold() {
+ return deviceMetadataInfoSwapThreshold;
+ }
+
@Override
public byte[] serialize() {
throw new UnsupportedOperationException(
@@ -299,6 +347,8 @@ public class ReadTsFileTableFunction implements
TableFunction {
+ outputColumnTypes
+ ", outputColumnCategories="
+ outputColumnCategories
+ + ", deviceMetadataInfoSwapThreshold="
+ + deviceMetadataInfoSwapThreshold
+ '}';
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
index f64f3d20b8f..117d5db4e5c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
@@ -1655,7 +1655,10 @@ public class RelationPlanner implements
AstVisitor<RelationPlan, Void> {
assignments,
tagAndAttributeIndexMap,
queryContext.createExternalTsFileQueryResource(
- handle.getTableName(), handle.getTsFilePaths(), assignments));
+ handle.getTableName(),
+ handle.getTsFilePaths(),
+ assignments,
+ handle.getDeviceMetadataInfoSwapThreshold()));
return new RelationPlan(scanNode, scope, outputSymbols, outerContext);
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResourceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResourceTest.java
index fb0031546c9..6c7919eb6fb 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResourceTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/function/tvf/read_tsfile/ExternalTsFileQueryResourceTest.java
@@ -152,7 +152,12 @@ public class ExternalTsFileQueryResourceTest {
queryContext.setStartTime(System.currentTimeMillis());
queryContext.setTimeOut(Long.MAX_VALUE);
return new ExternalTsFileQueryResource(
- queryContext, root.toPath().resolve("tmp"), "table1", tsFilePaths,
Collections.emptyMap());
+ queryContext,
+ root.toPath().resolve("tmp"),
+ "table1",
+ tsFilePaths,
+ Collections.emptyMap(),
+ ReadTsFileTableFunction.DEVICE_TASK_BUCKET_TARGET_SIZE_IN_BYTES);
}
private DeviceTaskRunReader newReader(DeviceTaskPartition partition) {
