This is an automated email from the ASF dual-hosted git repository. shuwenwei pushed a commit to branch addParamsForReadTsFileTableFunction in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 8ce75e286f4d04a51e326e1d84aef4dc7c77521a Author: shuwenwei <[email protected]> AuthorDate: Wed Jul 8 10:59:14 2026 +0800 Add DEVICE_METADATA_INFO_SWAP_THRESHOLD parameter to read_tsfile table function Introduces a new optional INT64 parameter DEVICE_METADATA_INFO_SWAP_THRESHOLD that allows users to control the flush run file size upper limit for external TsFile query device metadata swapping. The default value is 8 MiB, matching the previous hard-coded DEVICE_TASK_BUCKET_TARGET_SIZE_IN_BYTES constant. The constant is moved from ExternalTsFileQueryResource.DeviceTaskPartition to ReadTsFileTableFunction and threaded through the parameter handle, MPPQueryContext, and ExternalTsFileQueryResource as an instance field. --- .../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 | 57 ++++++++++++++++++++-- .../plan/relational/planner/RelationPlanner.java | 5 +- .../ExternalTsFileQueryResourceTest.java | 7 ++- 7 files changed, 78 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..6249e9b8531 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,12 @@ 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); + ReadTsFileTableFunctionHandle handle = new ReadTsFileTableFunctionHandle( schemaCollector.getTableName(), @@ -101,7 +115,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 +162,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 +239,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 +247,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 +266,8 @@ public class ReadTsFileTableFunction implements TableFunction { outputSchema.getFields().stream() .map(DescribedSchema.Field::getType) .collect(Collectors.toList()), - outputColumnCategories); + outputColumnCategories, + deviceMetadataInfoSwapThreshold); } private ReadTsFileTableFunctionHandle( @@ -236,7 +275,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 +291,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 +314,10 @@ public class ReadTsFileTableFunction implements TableFunction { return outputColumnCategories; } + public long getDeviceMetadataInfoSwapThreshold() { + return deviceMetadataInfoSwapThreshold; + } + @Override public byte[] serialize() { throw new UnsupportedOperationException( @@ -299,6 +344,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) {
