[To dev/1.3] feat: encode load attributes in active load directories (#16722) (#16758)

Author: luoluoyuyu

integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileIT.java

@@ -998,6 +998,47 @@ public void testLoadLocally() throws Exception {
     }
   }
 
+  @Test
+  public void testAsyncLoadLocally() throws Exception {
+    registerSchema();
+
+    final long writtenPoint1;
+    // device 0, device 1, sg 0
+    try (final TsFileGenerator generator =
+        new TsFileGenerator(new File(tmpDir, "1-0-0-0.tsfile"))) {
+      generator.registerTimeseries(
+          SchemaConfig.DEVICE_0, Collections.singletonList(SchemaConfig.MEASUREMENT_00));
+      generator.generateData(SchemaConfig.DEVICE_0, 1, PARTITION_INTERVAL / 10_000, false);
+      writtenPoint1 = generator.getTotalNumber();
+    }
+
+    try (final Connection connection = EnvFactory.getEnv().getConnection();
+        final Statement statement = connection.createStatement()) {
+
+      statement.execute(
+          String.format(
+              "load \"%s\" with ('async'='true','database-level'='2')", tmpDir.getAbsolutePath()));
+
+      for (int i = 0; i < 20; i++) {
+        try (final ResultSet resultSet =
+            statement.executeQuery("select count(*) from root.** group by level=1,2")) {
+          if (resultSet.next()) {
+            final long sg1Count = resultSet.getLong("count(root.sg.test_0.*.*)");
+            Assert.assertEquals(writtenPoint1, sg1Count);
+          } else {
+            Assert.fail("This ResultSet is empty.");
+          }
+        } catch (final Throwable e) {
+          if (i < 19) {
+            Thread.sleep(1000);
+          } else {
+            throw e;
+          }
+        }
+      }
+    }
+  }
+
   @Test
   @Ignore("Load with conversion is currently banned")
   public void testLoadWithConvertOnTypeMismatch() throws Exception {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java

@@ -83,6 +83,7 @@
 import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
 import org.apache.iotdb.db.queryengine.plan.statement.pipe.PipeEnrichedStatement;
+import org.apache.iotdb.db.storageengine.load.active.ActiveLoadPathHelper;
 import org.apache.iotdb.db.storageengine.load.active.ActiveLoadUtil;
 import org.apache.iotdb.db.storageengine.rescon.disk.FolderManager;
 import org.apache.iotdb.db.storageengine.rescon.disk.strategy.DirectoryStrategyType;
@@ -475,7 +476,14 @@ protected TSStatus loadFileV2(
   }
 
   private TSStatus loadTsFileAsync(final List<String> absolutePaths) throws IOException {
-    if (!ActiveLoadUtil.loadFilesToActiveDir(null, absolutePaths, true)) {
+    final Map<String, String> loadAttributes =
+        ActiveLoadPathHelper.buildAttributes(
+            null,
+            shouldConvertDataTypeOnTypeMismatch,
+            validateTsFile.get(),
+            null,
+            shouldMarkAsPipeRequest.get());
+    if (!ActiveLoadUtil.loadFilesToActiveDir(loadAttributes, absolutePaths, true)) {
       throw new PipeException("Load active listening pipe dir is not set.");
     }
     return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java

@@ -65,6 +65,7 @@
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.FileTimeIndex;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.timeindex.ITimeIndex;
 import org.apache.iotdb.db.storageengine.dataregion.utils.TsFileResourceUtils;
+import org.apache.iotdb.db.storageengine.load.active.ActiveLoadPathHelper;
 import org.apache.iotdb.db.storageengine.load.active.ActiveLoadUtil;
 import org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter;
 import org.apache.iotdb.db.storageengine.load.memory.LoadTsFileMemoryBlock;
@@ -240,7 +241,15 @@ public Analysis analyzeFileByFile(Analysis analysis) {
   private boolean doAsyncLoad(final Analysis analysis) {
     final long startTime = System.nanoTime();
     try {
-      if (ActiveLoadUtil.loadTsFileAsyncToActiveDir(tsFiles, null, isDeleteAfterLoad)) {
+      final Map<String, String> activeLoadAttributes =
+          ActiveLoadPathHelper.buildAttributes(
+              databaseLevel,
+              isConvertOnTypeMismatch,
+              isVerifySchema,
+              tabletConversionThresholdBytes,
+              isGeneratedByPipe);
+      if (ActiveLoadUtil.loadTsFileAsyncToActiveDir(
+          tsFiles, activeLoadAttributes, isDeleteAfterLoad)) {
         analysis.setFinishQueryAfterAnalyze(true);
         analysis.setFailStatus(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS));
         analysis.setStatement(loadTsFileStatement);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/LoadTsFileStatement.java

@@ -51,8 +51,6 @@ public class LoadTsFileStatement extends Statement {
   private boolean isGeneratedByPipe = false;
   private boolean isAsyncLoad = false;
 
-  private Map<String, String> loadAttributes;
-
   private List<File> tsFiles;
   private List<TsFileResource> resources;
   private List<Long> writePointCountList;
@@ -212,15 +210,14 @@ public long getWritePointCount(int resourceIndex) {
   }
 
   public void setLoadAttributes(final Map<String, String> loadAttributes) {
-    this.loadAttributes = loadAttributes;
-    initAttributes();
+    initAttributes(loadAttributes);
   }
 
   public boolean isAsyncLoad() {
     return isAsyncLoad;
   }
 
-  private void initAttributes() {
+  private void initAttributes(final Map<String, String> loadAttributes) {
     this.databaseLevel = LoadTsFileConfigurator.parseOrGetDefaultDatabaseLevel(loadAttributes);
     this.deleteAfterLoad = LoadTsFileConfigurator.parseOrGetDefaultOnSuccess(loadAttributes);
     this.convertOnTypeMismatch =
@@ -229,6 +226,9 @@ private void initAttributes() {
         LoadTsFileConfigurator.parseOrGetDefaultTabletConversionThresholdBytes(loadAttributes);
     this.verifySchema = LoadTsFileConfigurator.parseOrGetDefaultVerify(loadAttributes);
     this.isAsyncLoad = LoadTsFileConfigurator.parseOrGetDefaultAsyncLoad(loadAttributes);
+    if (LoadTsFileConfigurator.parseOrGetDefaultPipeGenerated(loadAttributes)) {
+      markIsGeneratedByPipe();
+    }
   }
 
   public boolean reconstructStatementIfMiniFileConverted(final List<Boolean> isMiniTsFile) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadDirScanner.java

@@ -39,6 +39,7 @@
 import java.nio.file.SimpleFileVisitor;
 import java.nio.file.attribute.BasicFileAttributes;
 import java.util.Arrays;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -101,8 +102,9 @@ private void scan() throws IOException {
 
       final boolean isGeneratedByPipe =
           listeningDir.equals(IOTDB_CONFIG.getLoadActiveListeningPipeDir());
+      final File listeningDirFile = new File(listeningDir);
       try (final Stream<File> fileStream =
-          FileUtils.streamFiles(new File(listeningDir), true, (String[]) null)) {
+          FileUtils.streamFiles(listeningDirFile, true, (String[]) null)) {
         try {
           fileStream
               .filter(file -> !activeLoadTsFileLoader.isFilePendingOrLoading(file))
@@ -115,7 +117,18 @@ private void scan() throws IOException {
               .filter(this::isTsFileCompleted)
               .limit(currentAllowedPendingSize)
               .forEach(
-                  file -> activeLoadTsFileLoader.tryTriggerTsFileLoad(file, isGeneratedByPipe));
+                  filePath -> {
+                    final File tsFile = new File(filePath);
+                    final Map<String, String> attributes =
+                        ActiveLoadPathHelper.parseAttributes(tsFile, listeningDirFile);
+
+                    final File parentFile = tsFile.getParentFile();
+
+                    activeLoadTsFileLoader.tryTriggerTsFileLoad(
+                        tsFile.getAbsolutePath(),
+                        listeningDirFile.getAbsolutePath(),
+                        isGeneratedByPipe);
+                  });
         } catch (UncheckedIOException e) {
           LOGGER.debug("The file has been deleted. Ignore this exception.");
         } catch (final Exception e) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/load/active/ActiveLoadFailedMessageHandler.java

@@ -21,7 +21,6 @@
 
 import org.apache.iotdb.commons.conf.CommonDescriptor;
 
-import org.apache.tsfile.utils.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,72 +40,72 @@ public class ActiveLoadFailedMessageHandler {
               // system is memory constrains
               put(
                   "memory",
-                  filePair ->
+                  entry ->
                       LOGGER.info(
                           "Rejecting auto load tsfile {} (isGeneratedByPipe = {}) due to memory constraints, will retry later.",
-                          filePair.getLeft(),
-                          filePair.getRight()));
+                          entry.getFile(),
+                          entry.isGeneratedByPipe()));
               // system is read only
               put(
                   "read only",
-                  filePair ->
+                  entry ->
                       LOGGER.info(
                           "Rejecting auto load tsfile {} (isGeneratedByPipe = {}) due to the system is read only, will retry later.",
-                          filePair.getLeft(),
-                          filePair.getRight()));
+                          entry.getFile(),
+                          entry.isGeneratedByPipe()));
               // Timed out to wait for procedure return. The procedure is still running.
               put(
                   "procedure return",
-                  filePair ->
+                  entry ->
                       LOGGER.info(
                           "Rejecting auto load tsfile {} (isGeneratedByPipe = {}) due to time out to wait for procedure return, will retry later.",
-                          filePair.getLeft(),
-                          filePair.getRight()));
+                          entry.getFile(),
+                          entry.isGeneratedByPipe()));
               // DataNode is not enough, please register more.
               put(
                   "not enough",
-                  filePair ->
+                  entry ->
                       LOGGER.info(
                           "Rejecting auto load tsfile {} (isGeneratedByPipe = {}) due to the datanode is not enough, will retry later.",
-                          filePair.getLeft(),
-                          filePair.getRight()));
+                          entry.getFile(),
+                          entry.isGeneratedByPipe()));
               // Fail to connect to any config node. Please check status of ConfigNodes or logs of
               // connected DataNode.
               put(
                   "any config node",
-                  filePair ->
+                  entry ->
                       LOGGER.info(
                           "Rejecting auto load tsfile {} (isGeneratedByPipe = {}) due to fail to connect to any config node, will retry later.",
-                          filePair.getLeft(),
-                          filePair.getRight()));
+                          entry.getFile(),
+                          entry.isGeneratedByPipe()));
               // Current query is time out, query start time is 1729653161797, ddl is
               // -3046040214706, current time is 1729653184210, please check your statement or
               // modify timeout parameter
               put(
                   "query is time out",
-                  filePair ->
+                  entry ->
                       LOGGER.info(
                           "Rejecting auto load tsfile {} (isGeneratedByPipe = {}) due to current query is time out, will retry later.",
-                          filePair.getLeft(),
-                          filePair.getRight()));
+                          entry.getFile(),
+                          entry.isGeneratedByPipe()));
             }
           });
 
   @FunctionalInterface
   private interface ExceptionMessageHandler {
-    void handle(final Pair<String, Boolean> filePair);
+    void handle(final ActiveLoadPendingQueue.ActiveLoadEntry activeLoadEntry);
   }
 
   public static boolean isExceptionMessageShouldRetry(
-      final Pair<String, Boolean> filePair, final String message) {
+      final ActiveLoadPendingQueue.ActiveLoadEntry entry, final String message) {
     if (CommonDescriptor.getInstance().getConfig().isReadOnly()) {
-      EXCEPTION_MESSAGE_HANDLER_MAP.get("read only").handle(filePair);
+      EXCEPTION_MESSAGE_HANDLER_MAP.get("read only").handle(entry);
       return true;
     }
 
     for (String key : EXCEPTION_MESSAGE_HANDLER_MAP.keySet()) {
       if (message != null && message.contains(key)) {
-        EXCEPTION_MESSAGE_HANDLER_MAP.get(key).handle(filePair);
+        EXCEPTION_MESSAGE_HANDLER_MAP.get(key).handle(entry);
         return true;
       }
     }