jt2594838 commented on code in PR #18421:
URL: https://github.com/apache/iotdb/pull/18421#discussion_r3734451709
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBDataRegionSyncSink.java:
##########
@@ -533,11 +537,20 @@ private void doTransfer(
final Map<Pair<String, Long>, Double> pipeName2WeightMap,
final File tsFile,
final File modFile,
- final String dataBaseName)
+ final String dataBaseName,
+ final Iterable<? extends EnrichedEvent> events,
+ final int outputIndex)
throws PipeException, IOException {
final Pair<IoTDBSyncClient, Boolean> clientAndStatus =
clientManager.getClient();
final TPipeTransferResp resp;
+ final String conversionTaskId =
+ PipeTransferTsFileSealWithModReq.generateConversionTaskId(
+ sinkTaskId,
+ events,
+ dataBaseName,
+ outputIndex,
+ Objects.nonNull(modFile) &&
clientManager.supportModsIfIsDataNodeReceiver());
Review Comment:
Generate the conversion key only when necessary
(shouldAsyncLoadTsFileOnTypeMismatch == true)?
The calculation seems costly.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java:
##########
@@ -765,6 +766,9 @@ private void
setFailAnalysisForTemporaryUnavailablePipeSchema(
private void executeTabletConversionOnException(
final IAnalysis analysis, final LoadAnalyzeException e) {
+ if (e instanceof LoadAnalyzeTypeMismatchException) {
+ PipeTsFileConversionTaskManager.markTypeMismatchDetected();
+ }
Review Comment:
The exception does not point to a specific task.
If one load triggers the conversion, will all tasks be moved into async path?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTableStatementDataTypeConvertExecutionVisitor.java:
##########
@@ -101,46 +132,98 @@ public Optional<TSStatus> visitLoadTsFile(
databaseName)
.constructStatement(),
loadTsFileStatement.isConvertOnTypeMismatch());
+ conversionContext.pendingStatement = statement;
+ conversionContext.pendingTabletInsertionEvent = null;
final TSStatus status = executeInsertTabletWithRetry(statement,
databaseName);
if (!handleTSStatus(status, loadTsFileStatement)) {
+ shouldReleaseContext = !isManagedTask ||
!isTemporaryUnavailable(status);
return Optional.of(status);
}
+ conversionContext.pendingStatement = null;
}
- } catch (final Exception e) {
- LOGGER.warn(
- StorageEngineMessages
-
.STORAGE_LOG_FAILED_TO_CONVERT_DATA_TYPE_FOR_LOADTSFILESTATEMENT_5D132E57,
- loadTsFileStatement,
- e);
- return Optional.of(
-
LoadTsFileDataTypeConverter.TABLE_STATEMENT_EXCEPTION_VISITOR.process(
- loadTsFileStatement, e));
+
+ conversionContext.closeParser();
+ conversionContext.fileIndex++;
}
- }
- if (loadTsFileStatement.isDeleteAfterLoad()) {
- loadTsFileStatement
- .getTsFiles()
- .forEach(
- tsfile -> {
-
org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(tsfile);
- final String tsFilePath = tsfile.getAbsolutePath();
- org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(
- new File(LoadUtil.getTsFileResourcePath(tsFilePath)));
- org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(
- new File(LoadUtil.getTsFileModsV1Path(tsFilePath)));
- org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(
- new File(LoadUtil.getTsFileModsV2Path(tsFilePath)));
- });
+ shouldReleaseContext = true;
+ if (loadTsFileStatement.isDeleteAfterLoad()) {
+ deleteSourceFiles(loadTsFileStatement);
+ }
+
+ LOGGER.info(
+ StorageEngineMessages
+
.STORAGE_LOG_DATA_TYPE_CONVERSION_FOR_LOADTSFILESTATEMENT_IS_SUCCESSFUL_99016326,
+ loadTsFileStatement);
+ return Optional.of(new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ } catch (final Exception e) {
+ LOGGER.warn(
+ StorageEngineMessages
+
.STORAGE_LOG_FAILED_TO_CONVERT_DATA_TYPE_FOR_LOADTSFILESTATEMENT_5D132E57,
+ loadTsFileStatement,
+ e);
+ final TSStatus status =
+
LoadTsFileDataTypeConverter.TABLE_STATEMENT_EXCEPTION_VISITOR.process(
+ loadTsFileStatement, e);
+ shouldReleaseContext =
+ !isManagedTask ||
!LoadTsFileDataTypeConverter.isMemoryPressureException(e);
+ return Optional.of(status);
+ } finally {
+ if (shouldReleaseContext) {
+ if (isManagedTask) {
+ PipeTsFileConversionTaskManager.clearCurrentContext();
+ } else {
+ conversionContext.close();
+ }
+ }
}
+ }
- LOGGER.info(
- StorageEngineMessages
-
.STORAGE_LOG_DATA_TYPE_CONVERSION_FOR_LOADTSFILESTATEMENT_IS_SUCCESSFUL_99016326,
- loadTsFileStatement);
+ private static boolean isTemporaryUnavailable(final TSStatus status) {
+ return status != null
+ && (status.getCode() ==
TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode()
+ || status.getCode()
+ ==
TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode());
+ }
- return Optional.of(new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ private static void deleteSourceFiles(final LoadTsFile statement) {
+ statement
+ .getTsFiles()
+ .forEach(
+ tsFile -> {
+
org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(tsFile);
Review Comment:
Mind the import
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/LoadTableStatementDataTypeConvertExecutionVisitor.java:
##########
@@ -101,46 +132,98 @@ public Optional<TSStatus> visitLoadTsFile(
databaseName)
.constructStatement(),
loadTsFileStatement.isConvertOnTypeMismatch());
+ conversionContext.pendingStatement = statement;
+ conversionContext.pendingTabletInsertionEvent = null;
final TSStatus status = executeInsertTabletWithRetry(statement,
databaseName);
if (!handleTSStatus(status, loadTsFileStatement)) {
+ shouldReleaseContext = !isManagedTask ||
!isTemporaryUnavailable(status);
return Optional.of(status);
}
+ conversionContext.pendingStatement = null;
}
- } catch (final Exception e) {
- LOGGER.warn(
- StorageEngineMessages
-
.STORAGE_LOG_FAILED_TO_CONVERT_DATA_TYPE_FOR_LOADTSFILESTATEMENT_5D132E57,
- loadTsFileStatement,
- e);
- return Optional.of(
-
LoadTsFileDataTypeConverter.TABLE_STATEMENT_EXCEPTION_VISITOR.process(
- loadTsFileStatement, e));
+
+ conversionContext.closeParser();
+ conversionContext.fileIndex++;
}
- }
- if (loadTsFileStatement.isDeleteAfterLoad()) {
- loadTsFileStatement
- .getTsFiles()
- .forEach(
- tsfile -> {
-
org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(tsfile);
- final String tsFilePath = tsfile.getAbsolutePath();
- org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(
- new File(LoadUtil.getTsFileResourcePath(tsFilePath)));
- org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(
- new File(LoadUtil.getTsFileModsV1Path(tsFilePath)));
- org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(
- new File(LoadUtil.getTsFileModsV2Path(tsFilePath)));
- });
+ shouldReleaseContext = true;
+ if (loadTsFileStatement.isDeleteAfterLoad()) {
+ deleteSourceFiles(loadTsFileStatement);
+ }
+
+ LOGGER.info(
+ StorageEngineMessages
+
.STORAGE_LOG_DATA_TYPE_CONVERSION_FOR_LOADTSFILESTATEMENT_IS_SUCCESSFUL_99016326,
+ loadTsFileStatement);
+ return Optional.of(new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ } catch (final Exception e) {
+ LOGGER.warn(
+ StorageEngineMessages
+
.STORAGE_LOG_FAILED_TO_CONVERT_DATA_TYPE_FOR_LOADTSFILESTATEMENT_5D132E57,
+ loadTsFileStatement,
+ e);
+ final TSStatus status =
+
LoadTsFileDataTypeConverter.TABLE_STATEMENT_EXCEPTION_VISITOR.process(
+ loadTsFileStatement, e);
+ shouldReleaseContext =
+ !isManagedTask ||
!LoadTsFileDataTypeConverter.isMemoryPressureException(e);
+ return Optional.of(status);
+ } finally {
+ if (shouldReleaseContext) {
+ if (isManagedTask) {
+ PipeTsFileConversionTaskManager.clearCurrentContext();
+ } else {
+ conversionContext.close();
+ }
+ }
}
+ }
- LOGGER.info(
- StorageEngineMessages
-
.STORAGE_LOG_DATA_TYPE_CONVERSION_FOR_LOADTSFILESTATEMENT_IS_SUCCESSFUL_99016326,
- loadTsFileStatement);
+ private static boolean isTemporaryUnavailable(final TSStatus status) {
+ return status != null
+ && (status.getCode() ==
TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode()
+ || status.getCode()
+ ==
TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode());
+ }
- return Optional.of(new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ private static void deleteSourceFiles(final LoadTsFile statement) {
+ statement
+ .getTsFiles()
+ .forEach(
+ tsFile -> {
+
org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(tsFile);
+ final String tsFilePath = tsFile.getAbsolutePath();
+ org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(
+ new File(LoadUtil.getTsFileResourcePath(tsFilePath)));
+ org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(
+ new File(LoadUtil.getTsFileModsV1Path(tsFilePath)));
+ org.apache.iotdb.commons.utils.FileUtils.deleteFileIfExist(
+ new File(LoadUtil.getTsFileModsV2Path(tsFilePath)));
+ });
+ }
+
+ private static final class TableConversionContext implements AutoCloseable {
Review Comment:
TableConversionContext -> TabletConversionContext
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/util/LoadUtil.java:
##########
@@ -186,53 +189,160 @@ public static boolean loadFilesToActiveDir(
return false;
}
final Map<String, String> attributes =
appendCurrentUserIfAbsent(loadAttributes);
- final File targetDir =
ActiveLoadPathHelper.resolveTargetDir(targetFilePath, attributes);
+ final File targetDir =
+ ActiveLoadPathHelper.resolvePipeTransferTargetDir(targetFilePath,
attributes);
final List<File> sourceFiles = new ArrayList<>(files.size());
for (final String file : files) {
sourceFiles.add(new File(file));
}
sourceFiles.sort(Comparator.comparing(LoadUtil::isTsFile));
- transferFilesToActiveDir(targetDir, sourceFiles, isDeleteAfterLoad);
+ transferFilesToActiveDir(
+ targetDir,
+ sourceFiles,
+ isDeleteAfterLoad,
+ attributes.get(ActiveLoadPathHelper.PIPE_CONVERSION_TASK_ID_KEY));
return true;
}
static void transferFilesToActiveDir(
final File targetDir, final List<File> sourceFiles, final boolean
isDeleteAfterLoad)
throws IOException {
+ transferFilesToActiveDir(targetDir, sourceFiles, isDeleteAfterLoad, null);
+ }
+
+ static void transferFilesToActiveDir(
+ final File targetDir,
+ final List<File> sourceFiles,
+ final boolean isDeleteAfterLoad,
+ final String deterministicDirectoryName)
+ throws IOException {
+ final File transferDir =
+ new File(
+ targetDir,
+ deterministicDirectoryName == null
+ ? UUID.randomUUID().toString()
+ : ActiveLoadPathHelper.formatPipeTaskTransferDirectoryName(
+ deterministicDirectoryName));
final List<File> existingSourceFiles = new ArrayList<>(sourceFiles.size());
for (final File sourceFile : sourceFiles) {
if (sourceFile.exists()) {
existingSourceFiles.add(sourceFile);
}
}
+
+ if (deterministicDirectoryName != null && transferDir.exists()) {
+ if (!isExistingTaskComplete(transferDir, sourceFiles)) {
+ throw new
IOException(StorageEngineMessages.FAIL_TO_LOAD_TSFILE_TO_ACTIVE_DIR);
+ }
+ if (isDeleteAfterLoad) {
+ deleteSourceFiles(existingSourceFiles);
+ }
+ return;
+ }
if (existingSourceFiles.isEmpty()) {
+ if (deterministicDirectoryName != null) {
+ // A retry is successful only when either the source or the published
deterministic target
+ // proves that the handoff completed. Reporting success for two
missing paths would make
+ // the receiver claim ownership of a task whose TsFile was lost.
+ throw new
IOException(StorageEngineMessages.FAIL_TO_LOAD_TSFILE_TO_ACTIVE_DIR);
+ }
return;
}
- final File transferDir = new File(targetDir, UUID.randomUUID().toString());
+ final File stagingDir =
+ new File(
+ targetDir,
+
ActiveLoadPathHelper.formatTransferStagingDirectoryName(UUID.randomUUID().toString()));
try {
- Files.createDirectories(transferDir.toPath());
+ Files.createDirectories(stagingDir.toPath());
for (final File sourceFile : existingSourceFiles) {
- final File targetFile = new File(transferDir, sourceFile.getName());
+ final File targetFile = new File(stagingDir, sourceFile.getName());
RetryUtils.retryOnException(
() -> {
transferFile(sourceFile, targetFile, isDeleteAfterLoad);
return null;
});
}
+ try {
+ publishTransferDirectory(stagingDir, transferDir);
+ } catch (final IOException e) {
+ // Another retry may have published the same deterministic task
between the existence
+ // check above and this rename. Reuse that complete handoff instead of
overwriting it.
+ if (deterministicDirectoryName == null
+ || !isExistingTaskComplete(transferDir, sourceFiles)) {
+ throw e;
+ }
+ }
} catch (final IOException | RuntimeException e) {
- if (transferDir.exists()) {
- FileUtils.deleteFileOrDirectoryWithRetry(transferDir);
+ if (stagingDir.exists()) {
+ FileUtils.deleteFileOrDirectoryWithRetry(stagingDir);
}
throw e;
}
+ if (stagingDir.exists()) {
+ FileUtils.deleteFileOrDirectoryWithRetry(stagingDir);
+ }
if (isDeleteAfterLoad) {
deleteSourceFiles(existingSourceFiles);
}
}
+ private static boolean isExistingTaskComplete(
+ final File transferDir, final List<File> sourceFiles) {
+ if (!transferDir.isDirectory()) {
+ return false;
+ }
+
+ final File[] targetFiles = transferDir.listFiles(File::isFile);
+ if (targetFiles == null || targetFiles.length == 0) {
+ return false;
+ }
Review Comment:
Why does a non-existent or empty dir indicate not-complete?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadPathHelper.java:
##########
@@ -289,6 +384,20 @@ private static void validateAttributeValue(final String
key, final String value)
case LoadTsFileConfigurator.VERIFY_KEY:
LoadTsFileConfigurator.validateVerifyParam(value);
break;
+ case PIPE_CONVERSION_TASK_ID_KEY:
+ if (value == null || value.isEmpty()) {
+ throw new
SemanticException(StorageEngineMessages.USER_NAME_MUST_NOT_BE_EMPTY);
Review Comment:
Wrong message
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java:
##########
@@ -348,6 +385,16 @@ private void handleOtherException(
}
}
+ private String getConversionTaskId(final
ActiveLoadPendingQueue.ActiveLoadEntry entry) {
+ final File tsFile = new File(entry.getFile());
+ final File pendingDir =
+ entry.getPendingDir() == null
+ ? ActiveLoadPathHelper.findPendingDirectory(tsFile)
+ : new File(entry.getPendingDir());
+ return ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir)
+ .get(ActiveLoadPathHelper.PIPE_CONVERSION_TASK_ID_KEY);
+ }
Review Comment:
Is it possible to store the taskId in the entry?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadTsFileLoader.java:
##########
@@ -270,27 +271,52 @@ private TSStatus loadTsFile(
? ActiveLoadPathHelper.findPendingDirectory(tsFile)
: new File(entry.getPendingDir());
final Map<String, String> attributes =
ActiveLoadPathHelper.parseAttributes(tsFile, pendingDir);
- ActiveLoadPathHelper.applyAttributesToStatement(attributes, statement,
isVerify);
- final String userName =
- attributes.getOrDefault(ActiveLoadPathHelper.USER_KEY,
AuthorityChecker.SUPER_USER);
- final Optional<Long> userId = AuthorityChecker.getUserId(userName);
- if (!userId.isPresent()) {
- return new TSStatus(TSStatusCode.USER_NOT_EXIST.getStatusCode())
-
.setMessage(StorageEngineMessages.USER_IN_ACTIVE_LOAD_PATH_DOES_NOT_EXIST);
- }
- session.setUserId(userId.get());
- session.setUsername(userName);
-
- final File parentFile;
- if (statement.getDatabase() == null && entry.isTableModel()) {
- statement.setDatabase(
- files.isEmpty() || (parentFile = files.get(0).getParentFile()) ==
null
- ? null
- : parentFile.getName());
- }
+ final String conversionTaskId =
+ attributes.get(ActiveLoadPathHelper.PIPE_CONVERSION_TASK_ID_KEY);
+ final boolean asyncLoadOnTypeMismatch =
+ Boolean.parseBoolean(
+ attributes.getOrDefault(
+ ActiveLoadPathHelper.PIPE_ASYNC_LOAD_ON_TYPE_MISMATCH_KEY,
"true"));
Review Comment:
Why is asyncLoadOnTypeMismatch used by ActiveLoader?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/PipeTsFileConversionTaskManager.java:
##########
@@ -0,0 +1,534 @@
+/*
+ * 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.storageengine.load.converter;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+
+/**
+ * Deduplicates pipe TsFile conversion tasks. The status table is bounded, and
only a bounded number
+ * of tasks retain an in-memory parser checkpoint. The active-load directory
remains the durable
+ * source of each receiver-owned file.
+ */
+public final class PipeTsFileConversionTaskManager {
+
+ public enum State {
+ PENDING,
+ RUNNING,
+ PAUSED,
+ SUCCESS,
+ FAILED
+ }
+
+ public static final class Task {
+ private final String taskId;
+ private final boolean asyncLoadOnTypeMismatch;
+ private volatile State state = State.PENDING;
+ private volatile TSStatus status;
+ private volatile boolean typeMismatchDetected;
+ private volatile boolean receiverOwned;
+ private boolean retrySealAllowed;
+ private Object conversionContext;
+
+ private Task(final String taskId, final boolean asyncLoadOnTypeMismatch) {
+ this.taskId = taskId;
+ this.asyncLoadOnTypeMismatch = asyncLoadOnTypeMismatch;
+ }
+
+ public String getTaskId() {
+ return taskId;
+ }
+
+ public boolean isAsyncLoadOnTypeMismatch() {
+ return asyncLoadOnTypeMismatch;
+ }
+
+ public State getState() {
+ return state;
+ }
+
+ public TSStatus getStatus() {
+ return status;
+ }
+
+ public boolean isTypeMismatchDetected() {
+ return typeMismatchDetected;
+ }
+
+ public boolean isReceiverOwned() {
+ return receiverOwned;
+ }
+ }
+
+ private static final class UnretainedContext {
+ private final String taskId;
+ private final Object context;
+
+ private UnretainedContext(final String taskId, final Object context) {
+ this.taskId = taskId;
+ this.context = context;
+ }
+ }
+
+ private static final int MAX_TASKS = 4096;
+ private static final int MAX_CONTEXTS =
+ LoadTsFileDataTypeConverter.getTabletConversionPermitCount();
+ private static final Map<String, Task> TASKS = new LinkedHashMap<>(128,
0.75F, true);
+ private static final ThreadLocal<String> CURRENT_TASK_ID = new
ThreadLocal<>();
+ // Keeps legacy seal requests (which predate conversion task ids) eligible
for receiver takeover.
+ private static final ThreadLocal<Boolean> CURRENT_TYPE_MISMATCH = new
ThreadLocal<>();
+ private static final ThreadLocal<UnretainedContext>
CURRENT_UNRETAINED_CONTEXT =
+ new ThreadLocal<>();
+
+ private PipeTsFileConversionTaskManager() {
+ // utility class
+ }
+
+ public static Task registerIfAbsent(final String taskId, final boolean
asyncLoadOnTypeMismatch) {
+ if (taskId == null || taskId.isEmpty()) {
+ return null;
+ }
+ synchronized (TASKS) {
+ Task task = TASKS.get(taskId);
+ if (task == null) {
+ if (!hasTaskCapacity()) {
+ return null;
+ }
+ task = new Task(taskId, asyncLoadOnTypeMismatch);
+ TASKS.put(taskId, task);
+ }
+ return task;
+ }
+ }
+
+ public static Task get(final String taskId) {
+ if (taskId == null || taskId.isEmpty()) {
+ return null;
+ }
+ synchronized (TASKS) {
+ return TASKS.get(taskId);
+ }
+ }
+
+ /** Returns a response for a duplicate seal, or {@code null} when no task is
known. */
+ public static TSStatus getDuplicateStatus(
+ final String taskId, final boolean asyncLoadOnTypeMismatch) {
+ if (taskId == null || taskId.isEmpty()) {
+ return null;
+ }
+ synchronized (TASKS) {
+ final Task task = TASKS.get(taskId);
+ return task == null ? null : getDuplicateStatus(task,
asyncLoadOnTypeMismatch);
+ }
+ }
+
+ /** Atomically claims a new/retryable seal or returns the status of its
existing task. */
+ public static TSStatus registerAndGetDuplicateStatus(
+ final String taskId, final boolean asyncLoadOnTypeMismatch) {
+ if (taskId == null || taskId.isEmpty()) {
+ return null;
+ }
+ synchronized (TASKS) {
+ final Task task = TASKS.get(taskId);
+ if (task == null) {
+ if (!hasTaskCapacity()) {
+ return createReceiverTemporaryUnavailableStatus(null);
+ }
+ TASKS.put(taskId, new Task(taskId, asyncLoadOnTypeMismatch));
+ return null;
+ }
+ if (task.retrySealAllowed && !task.receiverOwned) {
+ task.retrySealAllowed = false;
+ task.status = null;
+ task.state = State.PENDING;
+ return null;
+ }
+ return getDuplicateStatus(task, asyncLoadOnTypeMismatch);
+ }
+ }
+
+ private static TSStatus getDuplicateStatus(
+ final Task task, final boolean asyncLoadOnTypeMismatch) {
+ if (asyncLoadOnTypeMismatch || task.isAsyncLoadOnTypeMismatch()) {
+ if (task.receiverOwned || task.state == State.SUCCESS) {
+ // Once the receiver owns the file, the sender must not create a
second conversion task.
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+ if (task.state == State.FAILED && task.status != null) {
+ return toReceiverStatus(task.status);
+ }
+ if (task.state == State.PAUSED && task.status != null) {
+ return toReceiverStatus(task.status);
+ }
+ return createReceiverTemporaryUnavailableStatus(null);
+ }
+ if (task.state == State.SUCCESS) {
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+ if ((task.state == State.PAUSED || task.state == State.FAILED) &&
task.status != null) {
+ return toReceiverStatus(task.status);
+ }
+ return createReceiverTemporaryUnavailableStatus(null);
+ }
+
+ private static TSStatus toReceiverStatus(final TSStatus status) {
+ if (status == null
+ || status.getCode() !=
TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode()) {
+ return status;
+ }
+ return createReceiverTemporaryUnavailableStatus(status.getMessage());
+ }
+
+ private static TSStatus createReceiverTemporaryUnavailableStatus(final
String message) {
+ return new
TSStatus(TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode())
+ .setMessage(message);
+ }
+
+ public static void enter(final String taskId) {
+ CURRENT_TYPE_MISMATCH.set(false);
+ if (taskId != null && !taskId.isEmpty()) {
+ final String previousTaskId = CURRENT_TASK_ID.get();
+ if (previousTaskId != null && !previousTaskId.equals(taskId)) {
+ clearCurrentUnretainedContext(previousTaskId);
+ }
+ CURRENT_TASK_ID.set(taskId);
+ } else {
+ clearCurrentUnretainedContext(CURRENT_TASK_ID.get());
+ CURRENT_TASK_ID.remove();
+ }
+ }
+
+ public static String getCurrentTaskId() {
+ return CURRENT_TASK_ID.get();
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> T getOrCreateCurrentContext(final Supplier<T> supplier) {
+ final String currentTaskId = CURRENT_TASK_ID.get();
+ if (currentTaskId == null) {
+ return supplier.get();
+ }
+
+ final UnretainedContext currentUnretainedContext =
CURRENT_UNRETAINED_CONTEXT.get();
+ if (currentUnretainedContext != null) {
+ if (currentTaskId.equals(currentUnretainedContext.taskId)) {
+ return (T) currentUnretainedContext.context;
+ }
+ clearCurrentUnretainedContext(currentUnretainedContext.taskId);
+ }
+
+ Object evictedContext = null;
+ final T context;
+ boolean retained = false;
+ synchronized (TASKS) {
+ final Task task = TASKS.get(currentTaskId);
+ if (task == null) {
+ context = supplier.get();
+ } else if (task.state == State.SUCCESS || task.state == State.FAILED) {
+ // A parser callback that races with terminal completion may finish
its current call, but
+ // it must not recreate a checkpoint for a task that is already
terminal.
+ context = supplier.get();
+ } else if (task.conversionContext != null) {
+ context = (T) task.conversionContext;
+ retained = true;
+ } else {
+ context = supplier.get();
+ final ContextReservation reservation = reserveContextSlot(task);
+ evictedContext = reservation.evictedContext;
+ if (reservation.slotAvailable) {
+ task.conversionContext = context;
+ retained = true;
+ }
+ }
+ }
+ closeContext(evictedContext);
+ if (!retained) {
+ CURRENT_UNRETAINED_CONTEXT.set(new UnretainedContext(currentTaskId,
context));
+ }
+ return context;
+ }
+
+ public static void clearCurrentContext() {
+ clearContext(CURRENT_TASK_ID.get());
+ }
+
+ public static void clearContext(final String taskId) {
+ if (taskId == null || taskId.isEmpty()) {
+ return;
+ }
+ Object context = null;
+ synchronized (TASKS) {
+ final Task task = TASKS.get(taskId);
+ if (task != null) {
+ context = task.conversionContext;
+ task.conversionContext = null;
+ }
+ }
+ closeContext(context);
+ clearCurrentUnretainedContext(taskId);
+ }
+
+ private static ContextReservation reserveContextSlot(final Task currentTask)
{
+ int contextCount = 0;
+ for (final Task task : TASKS.values()) {
+ if (task.conversionContext != null) {
+ contextCount++;
+ }
+ }
+ if (contextCount < MAX_CONTEXTS) {
+ return new ContextReservation(true, null);
+ }
+ for (final Task task : TASKS.values()) {
+ if (task != currentTask && task.conversionContext != null && task.state
!= State.RUNNING) {
+ final Object context = task.conversionContext;
+ task.conversionContext = null;
+ return new ContextReservation(true, context);
+ }
+ }
+ return new ContextReservation(false, null);
+ }
+
+ private static final class ContextReservation {
+ private final boolean slotAvailable;
+ private final Object evictedContext;
+
+ private ContextReservation(final boolean slotAvailable, final Object
evictedContext) {
+ this.slotAvailable = slotAvailable;
+ this.evictedContext = evictedContext;
+ }
+ }
+
+ private static void closeContext(final Object context) {
+ if (!(context instanceof AutoCloseable)) {
+ return;
+ }
+ try {
+ ((AutoCloseable) context).close();
+ } catch (final Exception ignored) {
+ // Best-effort cleanup. The task state is still authoritative.
+ }
Review Comment:
May leave a warn log for debugging.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/converter/PipeTsFileConversionTaskManager.java:
##########
@@ -0,0 +1,534 @@
+/*
+ * 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.storageengine.load.converter;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Supplier;
+
+/**
+ * Deduplicates pipe TsFile conversion tasks. The status table is bounded, and
only a bounded number
+ * of tasks retain an in-memory parser checkpoint. The active-load directory
remains the durable
+ * source of each receiver-owned file.
+ */
+public final class PipeTsFileConversionTaskManager {
+
+ public enum State {
+ PENDING,
+ RUNNING,
+ PAUSED,
+ SUCCESS,
+ FAILED
+ }
+
+ public static final class Task {
+ private final String taskId;
+ private final boolean asyncLoadOnTypeMismatch;
+ private volatile State state = State.PENDING;
+ private volatile TSStatus status;
+ private volatile boolean typeMismatchDetected;
+ private volatile boolean receiverOwned;
+ private boolean retrySealAllowed;
+ private Object conversionContext;
+
+ private Task(final String taskId, final boolean asyncLoadOnTypeMismatch) {
+ this.taskId = taskId;
+ this.asyncLoadOnTypeMismatch = asyncLoadOnTypeMismatch;
+ }
+
+ public String getTaskId() {
+ return taskId;
+ }
+
+ public boolean isAsyncLoadOnTypeMismatch() {
+ return asyncLoadOnTypeMismatch;
+ }
+
+ public State getState() {
+ return state;
+ }
+
+ public TSStatus getStatus() {
+ return status;
+ }
+
+ public boolean isTypeMismatchDetected() {
+ return typeMismatchDetected;
+ }
+
+ public boolean isReceiverOwned() {
+ return receiverOwned;
+ }
+ }
+
+ private static final class UnretainedContext {
+ private final String taskId;
+ private final Object context;
+
+ private UnretainedContext(final String taskId, final Object context) {
+ this.taskId = taskId;
+ this.context = context;
+ }
+ }
+
+ private static final int MAX_TASKS = 4096;
+ private static final int MAX_CONTEXTS =
+ LoadTsFileDataTypeConverter.getTabletConversionPermitCount();
+ private static final Map<String, Task> TASKS = new LinkedHashMap<>(128,
0.75F, true);
+ private static final ThreadLocal<String> CURRENT_TASK_ID = new
ThreadLocal<>();
+ // Keeps legacy seal requests (which predate conversion task ids) eligible
for receiver takeover.
+ private static final ThreadLocal<Boolean> CURRENT_TYPE_MISMATCH = new
ThreadLocal<>();
+ private static final ThreadLocal<UnretainedContext>
CURRENT_UNRETAINED_CONTEXT =
+ new ThreadLocal<>();
+
+ private PipeTsFileConversionTaskManager() {
+ // utility class
+ }
+
+ public static Task registerIfAbsent(final String taskId, final boolean
asyncLoadOnTypeMismatch) {
+ if (taskId == null || taskId.isEmpty()) {
+ return null;
+ }
+ synchronized (TASKS) {
+ Task task = TASKS.get(taskId);
+ if (task == null) {
+ if (!hasTaskCapacity()) {
+ return null;
+ }
+ task = new Task(taskId, asyncLoadOnTypeMismatch);
+ TASKS.put(taskId, task);
+ }
+ return task;
+ }
+ }
+
+ public static Task get(final String taskId) {
+ if (taskId == null || taskId.isEmpty()) {
+ return null;
+ }
+ synchronized (TASKS) {
+ return TASKS.get(taskId);
+ }
+ }
+
+ /** Returns a response for a duplicate seal, or {@code null} when no task is
known. */
+ public static TSStatus getDuplicateStatus(
+ final String taskId, final boolean asyncLoadOnTypeMismatch) {
+ if (taskId == null || taskId.isEmpty()) {
+ return null;
+ }
+ synchronized (TASKS) {
+ final Task task = TASKS.get(taskId);
+ return task == null ? null : getDuplicateStatus(task,
asyncLoadOnTypeMismatch);
+ }
+ }
+
+ /** Atomically claims a new/retryable seal or returns the status of its
existing task. */
+ public static TSStatus registerAndGetDuplicateStatus(
+ final String taskId, final boolean asyncLoadOnTypeMismatch) {
+ if (taskId == null || taskId.isEmpty()) {
+ return null;
+ }
+ synchronized (TASKS) {
+ final Task task = TASKS.get(taskId);
+ if (task == null) {
+ if (!hasTaskCapacity()) {
+ return createReceiverTemporaryUnavailableStatus(null);
+ }
+ TASKS.put(taskId, new Task(taskId, asyncLoadOnTypeMismatch));
+ return null;
+ }
+ if (task.retrySealAllowed && !task.receiverOwned) {
+ task.retrySealAllowed = false;
+ task.status = null;
+ task.state = State.PENDING;
+ return null;
+ }
+ return getDuplicateStatus(task, asyncLoadOnTypeMismatch);
+ }
+ }
+
+ private static TSStatus getDuplicateStatus(
+ final Task task, final boolean asyncLoadOnTypeMismatch) {
+ if (asyncLoadOnTypeMismatch || task.isAsyncLoadOnTypeMismatch()) {
+ if (task.receiverOwned || task.state == State.SUCCESS) {
+ // Once the receiver owns the file, the sender must not create a
second conversion task.
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+ if (task.state == State.FAILED && task.status != null) {
+ return toReceiverStatus(task.status);
+ }
+ if (task.state == State.PAUSED && task.status != null) {
+ return toReceiverStatus(task.status);
+ }
+ return createReceiverTemporaryUnavailableStatus(null);
+ }
+ if (task.state == State.SUCCESS) {
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+ if ((task.state == State.PAUSED || task.state == State.FAILED) &&
task.status != null) {
+ return toReceiverStatus(task.status);
+ }
+ return createReceiverTemporaryUnavailableStatus(null);
+ }
+
+ private static TSStatus toReceiverStatus(final TSStatus status) {
+ if (status == null
+ || status.getCode() !=
TSStatusCode.LOAD_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode()) {
+ return status;
+ }
+ return createReceiverTemporaryUnavailableStatus(status.getMessage());
+ }
+
+ private static TSStatus createReceiverTemporaryUnavailableStatus(final
String message) {
+ return new
TSStatus(TSStatusCode.PIPE_RECEIVER_TEMPORARY_UNAVAILABLE_EXCEPTION.getStatusCode())
+ .setMessage(message);
+ }
+
+ public static void enter(final String taskId) {
+ CURRENT_TYPE_MISMATCH.set(false);
+ if (taskId != null && !taskId.isEmpty()) {
+ final String previousTaskId = CURRENT_TASK_ID.get();
+ if (previousTaskId != null && !previousTaskId.equals(taskId)) {
+ clearCurrentUnretainedContext(previousTaskId);
+ }
+ CURRENT_TASK_ID.set(taskId);
+ } else {
+ clearCurrentUnretainedContext(CURRENT_TASK_ID.get());
+ CURRENT_TASK_ID.remove();
+ }
+ }
+
+ public static String getCurrentTaskId() {
+ return CURRENT_TASK_ID.get();
+ }
+
+ @SuppressWarnings("unchecked")
+ public static <T> T getOrCreateCurrentContext(final Supplier<T> supplier) {
+ final String currentTaskId = CURRENT_TASK_ID.get();
+ if (currentTaskId == null) {
+ return supplier.get();
+ }
+
+ final UnretainedContext currentUnretainedContext =
CURRENT_UNRETAINED_CONTEXT.get();
+ if (currentUnretainedContext != null) {
+ if (currentTaskId.equals(currentUnretainedContext.taskId)) {
+ return (T) currentUnretainedContext.context;
+ }
+ clearCurrentUnretainedContext(currentUnretainedContext.taskId);
+ }
+
+ Object evictedContext = null;
+ final T context;
+ boolean retained = false;
+ synchronized (TASKS) {
+ final Task task = TASKS.get(currentTaskId);
+ if (task == null) {
+ context = supplier.get();
+ } else if (task.state == State.SUCCESS || task.state == State.FAILED) {
+ // A parser callback that races with terminal completion may finish
its current call, but
+ // it must not recreate a checkpoint for a task that is already
terminal.
+ context = supplier.get();
+ } else if (task.conversionContext != null) {
+ context = (T) task.conversionContext;
+ retained = true;
+ } else {
+ context = supplier.get();
+ final ContextReservation reservation = reserveContextSlot(task);
+ evictedContext = reservation.evictedContext;
+ if (reservation.slotAvailable) {
+ task.conversionContext = context;
+ retained = true;
+ }
+ }
+ }
+ closeContext(evictedContext);
+ if (!retained) {
+ CURRENT_UNRETAINED_CONTEXT.set(new UnretainedContext(currentTaskId,
context));
+ }
+ return context;
+ }
+
+ public static void clearCurrentContext() {
+ clearContext(CURRENT_TASK_ID.get());
+ }
+
+ public static void clearContext(final String taskId) {
+ if (taskId == null || taskId.isEmpty()) {
+ return;
+ }
+ Object context = null;
+ synchronized (TASKS) {
+ final Task task = TASKS.get(taskId);
+ if (task != null) {
+ context = task.conversionContext;
+ task.conversionContext = null;
+ }
+ }
+ closeContext(context);
+ clearCurrentUnretainedContext(taskId);
+ }
+
+ private static ContextReservation reserveContextSlot(final Task currentTask)
{
+ int contextCount = 0;
+ for (final Task task : TASKS.values()) {
+ if (task.conversionContext != null) {
+ contextCount++;
+ }
+ }
Review Comment:
Is it possible to use an atomic var?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
