You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2025/05/30 10:42:22 UTC
(iotdb) branch master updated: Load: Introduce thread pool for tablet conversion and insertion (#15611)
This is an automated email from the ASF dual-hosted git repository.
rong 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 87a37c2e37a Load: Introduce thread pool for tablet conversion and insertion (#15611)
87a37c2e37a is described below
commit 87a37c2e37ae1f3130df9521a7ba0d898af6813d
Author: Zikun Ma <55...@users.noreply.github.com>
AuthorDate: Fri May 30 18:42:17 2025 +0800
Load: Introduce thread pool for tablet conversion and insertion (#15611)
---
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 10 ++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 5 +
.../pipe/resource/memory/PipeMemoryWeightUtil.java | 12 +-
...leStatementDataTypeConvertExecutionVisitor.java | 98 +++++++-----
...eeStatementDataTypeConvertExecutionVisitor.java | 169 +++++++++++++++------
.../converter/LoadTsFileDataTypeConverter.java | 81 +++++++---
.../iotdb/commons/concurrent/ThreadName.java | 1 +
7 files changed, 262 insertions(+), 114 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 b91cb14f322..8137959a391 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
@@ -1083,6 +1083,8 @@ public class IoTDBConfig {
private long loadTsFileTabletConversionBatchMemorySizeInBytes = 4096 * 1024;
+ private int loadTsFileTabletConversionThreadCount = 5;
+
private int loadTsFileMaxDeviceCountToUseDeviceTimeIndex = 10000;
private long loadChunkMetadataMemorySizeInBytes = 33554432; // 32MB
@@ -3792,6 +3794,14 @@ public class IoTDBConfig {
loadTsFileTabletConversionBatchMemorySizeInBytes;
}
+ public int getLoadTsFileTabletConversionThreadCount() {
+ return loadTsFileTabletConversionThreadCount;
+ }
+
+ public void setLoadTsFileTabletConversionThreadCount(int loadTsFileTabletConversionThreadCount) {
+ this.loadTsFileTabletConversionThreadCount = loadTsFileTabletConversionThreadCount;
+ }
+
public int getLoadTsFileMaxDeviceCountToUseDeviceTimeIndex() {
return loadTsFileMaxDeviceCountToUseDeviceTimeIndex;
}
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 3e32fad522b..dfd854f724c 100755
--- 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
@@ -2176,6 +2176,11 @@ public class IoTDBDescriptor {
properties.getProperty(
"load_tsfile_tablet_conversion_batch_memory_size_in_bytes",
String.valueOf(conf.getLoadTsFileTabletConversionBatchMemorySizeInBytes()))));
+ conf.setLoadTsFileTabletConversionThreadCount(
+ Integer.parseInt(
+ properties.getProperty(
+ "load_tsfile_tablet_conversion_thread_count",
+ String.valueOf(conf.getLoadTsFileTabletConversionThreadCount()))));
conf.setLoadTsFileMaxDeviceCountToUseDeviceTimeIndex(
Integer.parseInt(
properties.getProperty(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryWeightUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryWeightUtil.java
index 7b3705cf2a5..2e2c29268ec 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryWeightUtil.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryWeightUtil.java
@@ -291,8 +291,8 @@ public class PipeMemoryWeightUtil {
return totalSizeInBytes;
}
- public static int calculateBatchDataRamBytesUsed(BatchData batchData) {
- int totalSizeInBytes = 0;
+ public static long calculateBatchDataRamBytesUsed(BatchData batchData) {
+ long totalSizeInBytes = 0;
// timestamp
totalSizeInBytes += 8;
@@ -307,16 +307,16 @@ public class PipeMemoryWeightUtil {
continue;
}
// consider variable references (plus 8) and memory alignment (round up to 8)
- totalSizeInBytes += roundUpToMultiple(primitiveType.getSize() + 8, 8);
+ totalSizeInBytes += roundUpToMultiple(primitiveType.getSize() + 8L, 8);
}
} else {
if (type.isBinary()) {
final Binary binary = batchData.getBinary();
// refer to org.apache.tsfile.utils.TsPrimitiveType.TsBinary.getSize
totalSizeInBytes +=
- roundUpToMultiple((binary == null ? 8 : binary.getLength() + 8) + 8, 8);
+ roundUpToMultiple((binary == null ? 8 : binary.ramBytesUsed()) + 8L, 8);
} else {
- totalSizeInBytes += roundUpToMultiple(TsPrimitiveType.getByType(type).getSize() + 8, 8);
+ totalSizeInBytes += roundUpToMultiple(TsPrimitiveType.getByType(type).getSize() + 8L, 8);
}
}
}
@@ -358,7 +358,7 @@ public class PipeMemoryWeightUtil {
* @param n The specified multiple.
* @return The nearest multiple of n greater than or equal to num.
*/
- private static int roundUpToMultiple(int num, int n) {
+ private static long roundUpToMultiple(long num, int n) {
if (n == 0) {
throw new IllegalArgumentException("The multiple n must be greater than 0");
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTableStatementDataTypeConvertExecutionVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTableStatementDataTypeConvertExecutionVisitor.java
index 23e8375fceb..af3e371c9ed 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTableStatementDataTypeConvertExecutionVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTableStatementDataTypeConvertExecutionVisitor.java
@@ -38,8 +38,14 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Objects;
import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+import static org.apache.iotdb.db.storageengine.load.converter.LoadTreeStatementDataTypeConvertExecutionVisitor.getExecutorPool;
public class LoadTableStatementDataTypeConvertExecutionVisitor
extends AstVisitor<Optional<TSStatus>, String> {
@@ -71,6 +77,7 @@ public class LoadTableStatementDataTypeConvertExecutionVisitor
LOGGER.info("Start data type conversion for LoadTsFileStatement: {}.", loadTsFileStatement);
+ final List<Future<TSStatus>> executionFutures = new ArrayList<>();
// TODO: Use batch insert after Table model supports insertMultiTablets
for (final File file : loadTsFileStatement.getTsFiles()) {
try (final TsFileInsertionEventTableParser parser =
@@ -98,42 +105,7 @@ public class LoadTableStatementDataTypeConvertExecutionVisitor
.constructStatement(),
loadTsFileStatement.isConvertOnTypeMismatch());
- TSStatus result;
- try {
- result =
- statement.accept(
- LoadTsFileDataTypeConverter.STATEMENT_STATUS_VISITOR,
- statementExecutor.execute(statement, databaseName));
-
- // Retry max 5 times if the write process is rejected
- for (int i = 0;
- i < 5
- && result.getCode()
- == TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode();
- i++) {
- Thread.sleep(100L * (i + 1));
- result =
- statement.accept(
- LoadTsFileDataTypeConverter.STATEMENT_STATUS_VISITOR,
- statementExecutor.execute(statement, databaseName));
- }
- } catch (final Exception e) {
- if (e instanceof InterruptedException) {
- Thread.currentThread().interrupt();
- }
- result = statement.accept(LoadTsFileDataTypeConverter.STATEMENT_EXCEPTION_VISITOR, e);
- }
-
- if (!(result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
- || result.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()
- || result.getCode()
- == TSStatusCode.LOAD_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())) {
- LOGGER.warn(
- "Failed to convert data type for LoadTsFileStatement: {}, status code is {}.",
- loadTsFileStatement,
- result.getCode());
- return Optional.empty();
- }
+ executionFutures.add(executeInsertTabletWithRetry(statement, databaseName));
}
} catch (final Exception e) {
LOGGER.warn(
@@ -142,6 +114,25 @@ public class LoadTableStatementDataTypeConvertExecutionVisitor
}
}
+ for (final Future<TSStatus> future : executionFutures) {
+ try {
+ final TSStatus result = future.get();
+ if (!(result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ || result.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()
+ || result.getCode()
+ == TSStatusCode.LOAD_IDEMPOTENT_CONFLICT_EXCEPTION.getStatusCode())) {
+ LOGGER.warn(
+ "Failed to convert data type for LoadTsFileStatement: {}, status code is {}.",
+ loadTsFileStatement,
+ result.getCode());
+ return Optional.empty();
+ }
+ } catch (ExecutionException | InterruptedException e) {
+ LOGGER.warn("Exception occurs when executing insertion during tablet conversion: ", e);
+ return Optional.empty();
+ }
+ }
+
if (loadTsFileStatement.isDeleteAfterLoad()) {
loadTsFileStatement
.getTsFiles()
@@ -160,4 +151,39 @@ public class LoadTableStatementDataTypeConvertExecutionVisitor
return Optional.of(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
}
+
+ private Future<TSStatus> executeInsertTabletWithRetry(
+ final LoadConvertedInsertTabletStatement statement, final String databaseName) {
+ return getExecutorPool()
+ .submit(
+ () -> {
+ TSStatus result;
+ try {
+ result =
+ statement.accept(
+ LoadTsFileDataTypeConverter.STATEMENT_STATUS_VISITOR,
+ statementExecutor.execute(statement, databaseName));
+
+ // Retry max 5 times if the write process is rejected
+ for (int i = 0;
+ i < 5
+ && result.getCode()
+ == TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode();
+ i++) {
+ Thread.sleep(100L * (i + 1));
+ result =
+ statement.accept(
+ LoadTsFileDataTypeConverter.STATEMENT_STATUS_VISITOR,
+ statementExecutor.execute(statement, databaseName));
+ }
+ } catch (final Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ result =
+ statement.accept(LoadTsFileDataTypeConverter.STATEMENT_EXCEPTION_VISITOR, e);
+ }
+ return result;
+ });
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitor.java
index 9420c9da361..838443a8b2e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTreeStatementDataTypeConvertExecutionVisitor.java
@@ -20,6 +20,9 @@
package org.apache.iotdb.db.storageengine.load.converter;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.concurrent.IoTThreadFactory;
+import org.apache.iotdb.commons.concurrent.ThreadName;
+import org.apache.iotdb.commons.concurrent.threadpool.WrappedThreadPoolExecutor;
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.connector.payload.evolvable.request.PipeTransferTabletRawReq;
@@ -46,6 +49,14 @@ import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import static org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil.calculateTabletSizeInBytes;
@@ -60,6 +71,9 @@ public class LoadTreeStatementDataTypeConvertExecutionVisitor
.getConfig()
.getLoadTsFileTabletConversionBatchMemorySizeInBytes();
+ private static final AtomicReference<WrappedThreadPoolExecutor> executorPool =
+ new AtomicReference<>();
+
private final StatementExecutor statementExecutor;
@FunctionalInterface
@@ -67,6 +81,21 @@ public class LoadTreeStatementDataTypeConvertExecutionVisitor
TSStatus execute(final Statement statement);
}
+ public static class CallerBlocksPolicy implements RejectedExecutionHandler {
+ public CallerBlocksPolicy() {}
+
+ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
+ if (!e.isShutdown()) {
+ try {
+ e.getQueue().put(r);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new RejectedExecutionException("task " + r + " rejected from " + e, ie);
+ }
+ }
+ }
+ }
+
public LoadTreeStatementDataTypeConvertExecutionVisitor(
final StatementExecutor statementExecutor) {
this.statementExecutor = statementExecutor;
@@ -90,6 +119,7 @@ public class LoadTreeStatementDataTypeConvertExecutionVisitor
final List<Long> tabletRawReqSizes = new ArrayList<>();
try {
+ final List<Future<TSStatus>> executionFutures = new ArrayList<>();
for (final File file : loadTsFileStatement.getTsFiles()) {
try (final TsFileInsertionEventScanParser parser =
new TsFileInsertionEventScanParser(
@@ -106,9 +136,16 @@ public class LoadTreeStatementDataTypeConvertExecutionVisitor
continue;
}
- final TSStatus result =
- executeInsertMultiTabletsWithRetry(
- tabletRawReqs, loadTsFileStatement.isConvertOnTypeMismatch());
+ final InsertMultiTabletsStatement batchStatement = new InsertMultiTabletsStatement();
+ batchStatement.setInsertTabletStatementList(
+ tabletRawReqs.stream()
+ .map(
+ req ->
+ new LoadConvertedInsertTabletStatement(
+ req.constructStatement(),
+ loadTsFileStatement.isConvertOnTypeMismatch()))
+ .collect(Collectors.toList()));
+ executionFutures.add(executeInsertMultiTabletsWithRetry(batchStatement));
for (final long memoryCost : tabletRawReqSizes) {
block.reduceMemoryUsage(memoryCost);
@@ -116,10 +153,6 @@ public class LoadTreeStatementDataTypeConvertExecutionVisitor
tabletRawReqs.clear();
tabletRawReqSizes.clear();
- if (!handleTSStatus(result, loadTsFileStatement)) {
- return Optional.empty();
- }
-
tabletRawReqs.add(tabletRawReq);
tabletRawReqSizes.add(curMemory);
block.addMemoryUsage(curMemory);
@@ -133,25 +166,39 @@ public class LoadTreeStatementDataTypeConvertExecutionVisitor
if (!tabletRawReqs.isEmpty()) {
try {
- final TSStatus result =
- executeInsertMultiTabletsWithRetry(
- tabletRawReqs, loadTsFileStatement.isConvertOnTypeMismatch());
+ final InsertMultiTabletsStatement batchStatement = new InsertMultiTabletsStatement();
+ batchStatement.setInsertTabletStatementList(
+ tabletRawReqs.stream()
+ .map(
+ req ->
+ new LoadConvertedInsertTabletStatement(
+ req.constructStatement(),
+ loadTsFileStatement.isConvertOnTypeMismatch()))
+ .collect(Collectors.toList()));
+ executionFutures.add(executeInsertMultiTabletsWithRetry(batchStatement));
for (final long memoryCost : tabletRawReqSizes) {
block.reduceMemoryUsage(memoryCost);
}
tabletRawReqs.clear();
tabletRawReqSizes.clear();
-
- if (!handleTSStatus(result, loadTsFileStatement)) {
- return Optional.empty();
- }
} catch (final Exception e) {
LOGGER.warn(
"Failed to convert data type for LoadTsFileStatement: {}.", loadTsFileStatement, e);
return Optional.empty();
}
}
+
+ for (final Future<TSStatus> future : executionFutures) {
+ try {
+ if (!handleTSStatus(future.get(), loadTsFileStatement)) {
+ return Optional.empty();
+ }
+ } catch (ExecutionException | InterruptedException e) {
+ LOGGER.warn("Exception occurs when executing insertion during tablet conversion: ", e);
+ return Optional.empty();
+ }
+ }
} finally {
for (final long memoryCost : tabletRawReqSizes) {
block.reduceMemoryUsage(memoryCost);
@@ -180,43 +227,67 @@ public class LoadTreeStatementDataTypeConvertExecutionVisitor
return Optional.of(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
}
- private TSStatus executeInsertMultiTabletsWithRetry(
- final List<PipeTransferTabletRawReq> tabletRawReqs, boolean isConvertOnTypeMismatch) {
- final InsertMultiTabletsStatement batchStatement = new InsertMultiTabletsStatement();
- batchStatement.setInsertTabletStatementList(
- tabletRawReqs.stream()
- .map(
- req ->
- new LoadConvertedInsertTabletStatement(
- req.constructStatement(), isConvertOnTypeMismatch))
- .collect(Collectors.toList()));
-
- TSStatus result;
- try {
- result =
- batchStatement.accept(
- LoadTsFileDataTypeConverter.STATEMENT_STATUS_VISITOR,
- statementExecutor.execute(batchStatement));
-
- // Retry max 5 times if the write process is rejected
- for (int i = 0;
- i < 5
- && result.getCode()
- == TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode();
- i++) {
- Thread.sleep(100L * (i + 1));
- result =
- batchStatement.accept(
- LoadTsFileDataTypeConverter.STATEMENT_STATUS_VISITOR,
- statementExecutor.execute(batchStatement));
- }
- } catch (final Exception e) {
- if (e instanceof InterruptedException) {
- Thread.currentThread().interrupt();
+ private Future<TSStatus> executeInsertMultiTabletsWithRetry(
+ final InsertMultiTabletsStatement batchStatement) {
+ return getExecutorPool()
+ .submit(
+ () -> {
+ TSStatus result;
+ try {
+ result =
+ batchStatement.accept(
+ LoadTsFileDataTypeConverter.STATEMENT_STATUS_VISITOR,
+ statementExecutor.execute(batchStatement));
+
+ // Retry max 5 times if the write process is rejected
+ for (int i = 0;
+ i < 5
+ && result.getCode()
+ == TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode();
+ i++) {
+ Thread.sleep(100L * (i + 1));
+ result =
+ batchStatement.accept(
+ LoadTsFileDataTypeConverter.STATEMENT_STATUS_VISITOR,
+ statementExecutor.execute(batchStatement));
+ }
+ } catch (final Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ result =
+ batchStatement.accept(
+ LoadTsFileDataTypeConverter.STATEMENT_EXCEPTION_VISITOR, e);
+ }
+ return result;
+ });
+ }
+
+ public static WrappedThreadPoolExecutor getExecutorPool() {
+ if (executorPool.get() == null) {
+ synchronized (executorPool) {
+ if (executorPool.get() == null) {
+ executorPool.set(
+ new WrappedThreadPoolExecutor(
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getLoadTsFileTabletConversionThreadCount(),
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getLoadTsFileTabletConversionThreadCount(),
+ 0L,
+ TimeUnit.SECONDS,
+ new ArrayBlockingQueue<>(
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .getLoadTsFileTabletConversionThreadCount()),
+ new IoTThreadFactory(ThreadName.LOAD_DATATYPE_CONVERT_POOL.getName()),
+ ThreadName.LOAD_DATATYPE_CONVERT_POOL.getName(),
+ new CallerBlocksPolicy()));
+ }
}
- result = batchStatement.accept(LoadTsFileDataTypeConverter.STATEMENT_EXCEPTION_VISITOR, e);
}
- return result;
+ return executorPool.get();
}
private static boolean handleTSStatus(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java
index 57e3df75ac0..57457708e40 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTsFileDataTypeConverter.java
@@ -20,7 +20,11 @@
package org.apache.iotdb.db.storageengine.load.converter;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.protocol.session.IClientSession;
+import org.apache.iotdb.db.protocol.session.InternalClientSession;
import org.apache.iotdb.db.protocol.session.SessionManager;
import org.apache.iotdb.db.queryengine.plan.Coordinator;
import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
@@ -36,6 +40,7 @@ import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.ZoneId;
import java.util.Optional;
public class LoadTsFileDataTypeConverter {
@@ -78,18 +83,33 @@ public class LoadTsFileDataTypeConverter {
}
private TSStatus executeForTableModel(final Statement statement, final String databaseName) {
- return Coordinator.getInstance()
- .executeForTableModel(
- isGeneratedByPipe ? new PipeEnrichedStatement(statement) : statement,
- relationalSqlParser,
- SESSION_MANAGER.getCurrSession(),
- SESSION_MANAGER.requestQueryId(),
- SESSION_MANAGER.getSessionInfoOfPipeReceiver(
- SESSION_MANAGER.getCurrSession(), databaseName),
- "",
- LocalExecutionPlanner.getInstance().metadata,
- IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold())
- .status;
+ final IClientSession session =
+ new InternalClientSession(
+ String.format(
+ "%s_%s",
+ LoadTsFileDataTypeConverter.class.getSimpleName(),
+ Thread.currentThread().getName()));
+ session.setUsername(AuthorityChecker.SUPER_USER);
+ session.setClientVersion(IoTDBConstant.ClientVersion.V_1_0);
+ session.setZoneId(ZoneId.systemDefault());
+ session.setSqlDialect(IClientSession.SqlDialect.TABLE);
+
+ SESSION_MANAGER.registerSession(session);
+ try {
+ return Coordinator.getInstance()
+ .executeForTableModel(
+ isGeneratedByPipe ? new PipeEnrichedStatement(statement) : statement,
+ relationalSqlParser,
+ session,
+ SESSION_MANAGER.requestQueryId(),
+ SESSION_MANAGER.getSessionInfoOfPipeReceiver(session, databaseName),
+ "",
+ LocalExecutionPlanner.getInstance().metadata,
+ IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold())
+ .status;
+ } finally {
+ SESSION_MANAGER.removeCurrSession();
+ }
}
public Optional<TSStatus> convertForTreeModel(final LoadTsFileStatement loadTsFileTreeStatement) {
@@ -104,17 +124,32 @@ public class LoadTsFileDataTypeConverter {
}
private TSStatus executeForTreeModel(final Statement statement) {
- return Coordinator.getInstance()
- .executeForTreeModel(
- isGeneratedByPipe ? new PipeEnrichedStatement(statement) : statement,
- SESSION_MANAGER.requestQueryId(),
- SESSION_MANAGER.getSessionInfo(SESSION_MANAGER.getCurrSession()),
- "",
- ClusterPartitionFetcher.getInstance(),
- ClusterSchemaFetcher.getInstance(),
- IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold(),
- false)
- .status;
+ final IClientSession session =
+ new InternalClientSession(
+ String.format(
+ "%s_%s",
+ LoadTsFileDataTypeConverter.class.getSimpleName(),
+ Thread.currentThread().getName()));
+ session.setUsername(AuthorityChecker.SUPER_USER);
+ session.setClientVersion(IoTDBConstant.ClientVersion.V_1_0);
+ session.setZoneId(ZoneId.systemDefault());
+
+ SESSION_MANAGER.registerSession(session);
+ try {
+ return Coordinator.getInstance()
+ .executeForTreeModel(
+ isGeneratedByPipe ? new PipeEnrichedStatement(statement) : statement,
+ SESSION_MANAGER.requestQueryId(),
+ SESSION_MANAGER.getSessionInfo(session),
+ "",
+ ClusterPartitionFetcher.getInstance(),
+ ClusterSchemaFetcher.getInstance(),
+ IoTDBDescriptor.getInstance().getConfig().getQueryTimeoutThreshold(),
+ false)
+ .status;
+ } finally {
+ SESSION_MANAGER.removeCurrSession();
+ }
}
public boolean isSuccessful(final TSStatus status) {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
index 076194dc3ae..b5373e71bd2 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java
@@ -148,6 +148,7 @@ public enum ThreadName {
PIPE_ASYNC_CONNECTOR_CLIENT_POOL("Pipe-Async-Connector-Client-Pool"),
PIPE_RECEIVER_AIR_GAP_AGENT("Pipe-Receiver-Air-Gap-Agent"),
PIPE_AIR_GAP_RECEIVER("Pipe-Air-Gap-Receiver"),
+ LOAD_DATATYPE_CONVERT_POOL("Load-Datatype-Convert-Pool"),
SUBSCRIPTION_EXECUTOR_POOL("Subscription-Executor-Pool"),
SUBSCRIPTION_RUNTIME_META_SYNCER("Subscription-Runtime-Meta-Syncer"),
WINDOW_EVALUATION_SERVICE("WindowEvaluationTaskPoolManager"),