Fix pipe history LoadTsFile schema creation

Allow pipe-generated tree LoadTsFile analysis to create the database and missing timeseries schema while verifying the tsfile, even when the receiver disables global auto schema creation. Keep non-pipe load and insert behavior unchanged. Add a regression test for auto-split history sync with deletion and receiver auto-create disabled.

Author: Caideyipi

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeReceiverAutoCreateDisabledIT.java

@@ -36,6 +36,7 @@
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Set;
 
@@ -61,14 +62,16 @@ protected void setupConfig() {
         .getConfig()
         .getCommonConfig()
         .setDataReplicationFactor(1)
-        .setSchemaReplicationFactor(1);
+        .setSchemaReplicationFactor(1)
+        .setPipeAutoSplitFullEnabled(true);
     receiverEnv
         .getConfig()
         .getCommonConfig()
         .setAutoCreateSchemaEnabled(false)
         .setDatanodeMemoryProportion("3:3:1:1:1:0")
         .setDataReplicationFactor(1)
-        .setSchemaReplicationFactor(1);
+        .setSchemaReplicationFactor(1)
+        .setPipeAutoSplitFullEnabled(true);
   }
 
   @Test
@@ -122,6 +125,49 @@ public void testReceiverAutoCreateSchemaDisabledWithSpecialTimeSeries() throws E
     }
   }
 
+  @Test
+  public void testAutoSplitHistoryTsFileWithDeletionWhenReceiverAutoCreateSchemaDisabled()
+      throws Exception {
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "create database root.sg",
+            "create timeseries root.sg.non_aligned.s1 with datatype=INT32",
+            "create timeseries root.sg.non_aligned.s2 with datatype=DOUBLE",
+            "create aligned timeseries root.sg.aligned(s1 INT32, s2 DOUBLE)",
+            "insert into root.sg.non_aligned(time, s1, s2) values(1, 1, 1.0), (2, 2, 2.0), (3, 3, 3.0)",
+            "insert into root.sg.aligned(time, s1, s2) values(1, 10, 10.0), (2, 20, 20.0), (3, 30, 30.0)",
+            "flush",
+            "delete from root.sg.non_aligned.s1 where time > 2",
+            "delete from root.sg.aligned.* where time > 2",
+            "flush"));
+
+    awaitUntilFlush(senderEnv);
+
+    TestUtils.executeNonQuery(
+        senderEnv,
+        String.format(
+            "create pipe test with source ('inclusion'='all', 'source.history.enable'='true', 'source.realtime.mode'='batch') "
+                + "with sink ('sink'='iotdb-thrift-sink', 'sink.node-urls'='%s')",
+            receiverEnv.getDataNodeWrapper(0).getIpAndPortString()));
+
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "select * from root.sg.non_aligned",
+        "Time,root.sg.non_aligned.s1,root.sg.non_aligned.s2,",
+        new HashSet<>(Arrays.asList("1,1,1.0,", "2,2,2.0,", "3,null,3.0,")));
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "select * from root.sg.aligned",
+        "Time,root.sg.aligned.s1,root.sg.aligned.s2,",
+        new HashSet<>(Arrays.asList("1,10,10.0,", "2,20,20.0,")));
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "show devices root.sg.aligned",
+        "Device,IsAligned,Template,TTL(ms),",
+        new HashSet<>(Arrays.asList("root.sg.aligned,true,null,INF,")));
+  }
+
   private QueryResult queryForResult(final Statement statement, final String sql)
       throws SQLException {
     try (final ResultSet resultSet = statement.executeQuery(sql)) {

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

@@ -191,6 +191,10 @@ protected boolean isConvertOnTypeMismatch() {
     return isConvertOnTypeMismatch;
   }
 
+  protected boolean isGeneratedByPipe() {
+    return isGeneratedByPipe;
+  }
+
   public IAnalysis analyzeFileByFile(IAnalysis analysis) {
     if (!checkBeforeAnalyzeFileByFile(analysis)) {
       return analysis;

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

@@ -215,12 +215,10 @@ private void doAutoCreateAndVerify()
         makeSureNoDuplicatedMeasurementsInDevices();
       }
 
-      if (loadTsFileAnalyzer.isAutoCreateDatabase()) {
+      if (loadTsFileAnalyzer.isAutoCreateDatabase() || loadTsFileAnalyzer.isGeneratedByPipe()) {
         autoCreateDatabase();
       }
 
-      // schema fetcher will not auto create if config set
-      // isAutoCreateSchemaEnabled is false.
       final ISchemaTree schemaTree = autoCreateSchema();
 
       if (loadTsFileAnalyzer.isVerifySchema()) {
@@ -432,7 +430,8 @@ private ISchemaTree autoCreateSchema() throws IllegalPathException {
         encodingsList,
         compressionTypesList,
         isAlignedList,
-        loadTsFileAnalyzer.context);
+        loadTsFileAnalyzer.context,
+        loadTsFileAnalyzer.isGeneratedByPipe());
   }
 
   private void verifySchema(ISchemaTree schemaTree)

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetcher.java

@@ -289,6 +289,27 @@ public ISchemaTree fetchSchemaListWithAutoCreate(
       final List<CompressionType[]> compressionTypesList,
       final List<Boolean> isAlignedList,
       final MPPQueryContext context) {
+    return fetchSchemaListWithAutoCreate(
+        devicePathList,
+        measurementsList,
+        tsDataTypesList,
+        encodingsList,
+        compressionTypesList,
+        isAlignedList,
+        context,
+        false);
+  }
+
+  @Override
+  public ISchemaTree fetchSchemaListWithAutoCreate(
+      final List<PartialPath> devicePathList,
+      final List<String[]> measurementsList,
+      final List<TSDataType[]> tsDataTypesList,
+      final List<TSEncoding[]> encodingsList,
+      final List<CompressionType[]> compressionTypesList,
+      final List<Boolean> isAlignedList,
+      final MPPQueryContext context,
+      final boolean forceAutoCreate) {
     // The schema cache R/W and fetch operation must be locked together thus the cache clean
     // operation executed by delete timeSeries will be effective.
     DataNodeSchemaLockManager.getInstance()
@@ -326,7 +347,7 @@ public ISchemaTree fetchSchemaListWithAutoCreate(
         schemaTree.mergeSchemaTree(remoteSchemaTree);
       }
 
-      if (!config.isAutoCreateSchemaEnabled()) {
+      if (!forceAutoCreate && !config.isAutoCreateSchemaEnabled()) {
         // disable auto-create for non-system series
         indexOfDevicesWithMissingMeasurements.removeIf(
             i ->

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ISchemaFetcher.java

@@ -121,6 +121,19 @@ ISchemaTree fetchSchemaListWithAutoCreate(
       List<Boolean> aligned,
       MPPQueryContext context);
 
+  default ISchemaTree fetchSchemaListWithAutoCreate(
+      List<PartialPath> devicePath,
+      List<String[]> measurements,
+      List<TSDataType[]> tsDataTypes,
+      List<TSEncoding[]> encodings,
+      List<CompressionType[]> compressionTypes,
+      List<Boolean> aligned,
+      MPPQueryContext context,
+      boolean forceAutoCreate) {
+    return fetchSchemaListWithAutoCreate(
+        devicePath, measurements, tsDataTypes, encodings, compressionTypes, aligned, context);
+  }
+
   Pair<Template, PartialPath> checkTemplateSetInfo(PartialPath devicePath);
 
   Pair<Template, PartialPath> checkTemplateSetAndPreSetInfo(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/SchemaValidator.java

@@ -97,4 +97,25 @@ public static ISchemaTree validate(
     return schemaFetcher.fetchSchemaListWithAutoCreate(
         devicePaths, measurements, dataTypes, encodings, compressionTypes, isAlignedList, context);
   }
+
+  public static ISchemaTree validate(
+      ISchemaFetcher schemaFetcher,
+      List<PartialPath> devicePaths,
+      List<String[]> measurements,
+      List<TSDataType[]> dataTypes,
+      List<TSEncoding[]> encodings,
+      List<CompressionType[]> compressionTypes,
+      List<Boolean> isAlignedList,
+      MPPQueryContext context,
+      boolean forceAutoCreate) {
+    return schemaFetcher.fetchSchemaListWithAutoCreate(
+        devicePaths,
+        measurements,
+        dataTypes,
+        encodings,
+        compressionTypes,
+        isAlignedList,
+        context,
+        forceAutoCreate);
+  }
 }