Fix pipe history LoadTsFile after deleting aligned measurements

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

@@ -182,6 +182,43 @@ public void testAutoSplitHistoryTsFileWithDeletionWhenReceiverAutoCreateSchemaDi
         new HashSet<>(Arrays.asList("root.sg.aligned,true,null,INF,")));
   }
 
+  @Test
+  public void
+      testAutoSplitHistoryTsFileWithUnflushedAlignedDeletionWhenReceiverAutoCreateSchemaDisabled()
+          throws Exception {
+    TestUtils.executeNonQueries(
+        senderEnv,
+        Arrays.asList(
+            "create database root.sg",
+            "create aligned timeseries root.sg.aligned(s0 INT32, s1 INT64, s2 DOUBLE)",
+            "insert into root.sg.aligned(time, s0, s1, s2) aligned "
+                + "values(1, 1, 10, 1.0), (2, 2, 20, 2.0), (3, 3, 30, 3.0)",
+            "delete timeseries root.sg.aligned.s1"));
+
+    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 s0,s2 from root.sg.aligned",
+        "Time,root.sg.aligned.s0,root.sg.aligned.s2,",
+        new HashSet<>(Arrays.asList("1,1,1.0,", "2,2,2.0,", "3,3,3.0,")));
+    TestUtils.assertDataEventuallyOnEnv(
+        receiverEnv,
+        "count timeseries root.sg.aligned.*",
+        "count(timeseries),",
+        new HashSet<>(Arrays.asList("2,")));
+    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/storageengine/dataregion/memtable/AlignedWritableMemChunk.java

@@ -336,7 +336,7 @@
     return timestamps;
   }
 
   private void getAnySatisfiedTimestamp(
       AlignedTVList alignedTVList,
       List<List<TimeRange>> valueColumnsDeletionList,
       boolean ignoreAllNullRows,
@@ -542,14 +542,45 @@
 
   @Override
   public IChunkWriter createIChunkWriter() {
-    return new AlignedChunkWriterImpl(schemaList, encryptParameter);
+    return new AlignedChunkWriterImpl(getActiveSchemaList(), encryptParameter);
+  }
+
+  private List<IMeasurementSchema> getActiveSchemaList() {
+    if (measurementIndexMap.size() == schemaList.size()) {
+      return schemaList;
+    }
+    List<IMeasurementSchema> activeSchemaList = new ArrayList<>(measurementIndexMap.size());
+    for (int i = 0; i < schemaList.size(); i++) {
+      IMeasurementSchema schema = schemaList.get(i);
+      Integer activeColumnIndex = measurementIndexMap.get(schema.getMeasurementName());
+      if (activeColumnIndex != null && activeColumnIndex == i) {
+        activeSchemaList.add(schema);
+      }
+    }
+    return activeSchemaList;
+  }
+
+  private List<TSDataType> getDataTypes(List<IMeasurementSchema> schemaList) {
+    if (schemaList == this.schemaList) {
+      return dataTypes;
+    }
+    List<TSDataType> activeDataTypes = new ArrayList<>(schemaList.size());
+    for (IMeasurementSchema schema : schemaList) {
+      activeDataTypes.add(schema.getType());
+    }
+    return activeDataTypes;
   }
 
   @SuppressWarnings({"squid:S6541", "squid:S3776"})
   public void encodeWorkingAlignedTVList(
       BlockingQueue<Object> ioTaskQueue,
       long maxNumberOfPointsInChunk,
       int maxNumberOfPointsInPage) {
+    List<IMeasurementSchema> activeSchemaList = getActiveSchemaList();
+    if (activeSchemaList.isEmpty() && ignoreAllNullRows) {
+      return;
+    }
+
     BitMap allValueColDeletedMap;
     AlignedTVList alignedWorkingListForFlush = (AlignedTVList) workingListForFlush;
 
@@ -615,29 +646,36 @@
     }
 
     handleEncoding(
-        ioTaskQueue, chunkRange, timeDuplicateInfo, allValueColDeletedMap, maxNumberOfPointsInPage);
+        ioTaskQueue,
+        chunkRange,
+        timeDuplicateInfo,
+        allValueColDeletedMap,
+        maxNumberOfPointsInPage,
+        activeSchemaList);
   }
 
   private void handleEncoding(
       BlockingQueue<Object> ioTaskQueue,
       List<List<Integer>> chunkRange,
       boolean[] timeDuplicateInfo,
       BitMap allValueColDeletedMap,
-      int maxNumberOfPointsInPage) {
+      int maxNumberOfPointsInPage,
+      List<IMeasurementSchema> activeSchemaList) {
     AlignedTVList alignedWorkingListForFlush = (AlignedTVList) workingListForFlush;
-    List<TSDataType> dataTypes = alignedWorkingListForFlush.getTsDataTypes();
-    Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new Pair[dataTypes.size()];
+    List<Integer> columnIndexList = buildColumnIndexList(activeSchemaList);
+    Pair<Long, Integer>[] lastValidPointIndexForTimeDupCheck = new Pair[activeSchemaList.size()];
     for (List<Integer> pageRange : chunkRange) {
       AlignedChunkWriterImpl alignedChunkWriter =
-          new AlignedChunkWriterImpl(schemaList, encryptParameter);
+          new AlignedChunkWriterImpl(activeSchemaList, encryptParameter);
       for (int pageNum = 0; pageNum < pageRange.size() / 2; pageNum += 1) {
-        for (int columnIndex = 0; columnIndex < dataTypes.size(); columnIndex++) {
+        for (int columnIndex = 0; columnIndex < activeSchemaList.size(); columnIndex++) {
+          int tvListColumnIndex = columnIndexList.get(columnIndex);
           // Pair of Time and Index
           if (Objects.nonNull(timeDuplicateInfo)
               && lastValidPointIndexForTimeDupCheck[columnIndex] == null) {
             lastValidPointIndexForTimeDupCheck[columnIndex] = new Pair<>(Long.MIN_VALUE, null);
           }
-          TSDataType tsDataType = dataTypes.get(columnIndex);
+          TSDataType tsDataType = activeSchemaList.get(columnIndex).getType();
           for (int sortedRowIndex = pageRange.get(pageNum * 2);
               sortedRowIndex <= pageRange.get(pageNum * 2 + 1);
               sortedRowIndex++) {
@@ -650,8 +688,10 @@
             // skip time duplicated rows
             long time = alignedWorkingListForFlush.getTime(sortedRowIndex);
             if (Objects.nonNull(timeDuplicateInfo)) {
-              if (!alignedWorkingListForFlush.isNullValue(
-                  alignedWorkingListForFlush.getValueIndex(sortedRowIndex), columnIndex)) {
+              if (tvListColumnIndex >= 0
+                  && !alignedWorkingListForFlush.isNullValue(
+                      alignedWorkingListForFlush.getValueIndex(sortedRowIndex),
+                      tvListColumnIndex)) {
                 lastValidPointIndexForTimeDupCheck[columnIndex].left = time;
                 lastValidPointIndexForTimeDupCheck[columnIndex].right =
                     alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
@@ -671,21 +711,24 @@
             // write(T:3,V:null)
 
             int originRowIndex;
-            if (Objects.nonNull(lastValidPointIndexForTimeDupCheck[columnIndex])
+            if (tvListColumnIndex >= 0
+                && Objects.nonNull(lastValidPointIndexForTimeDupCheck[columnIndex])
                 && (time == lastValidPointIndexForTimeDupCheck[columnIndex].left)) {
               originRowIndex = lastValidPointIndexForTimeDupCheck[columnIndex].right;
             } else {
               originRowIndex = alignedWorkingListForFlush.getValueIndex(sortedRowIndex);
             }
 
-            boolean isNull = alignedWorkingListForFlush.isNullValue(originRowIndex, columnIndex);
+            boolean isNull =
+                tvListColumnIndex < 0
+                    || alignedWorkingListForFlush.isNullValue(originRowIndex, tvListColumnIndex);
             switch (tsDataType) {
               case BOOLEAN:
                 alignedChunkWriter.writeByColumn(
                     time,
                     !isNull
                         && alignedWorkingListForFlush.getBooleanByValueIndex(
-                            originRowIndex, columnIndex),
+                            originRowIndex, tvListColumnIndex),
                     isNull);
                 break;
               case INT32:
@@ -695,7 +738,7 @@
                     isNull
                         ? 0
                         : alignedWorkingListForFlush.getIntByValueIndex(
-                            originRowIndex, columnIndex),
+                            originRowIndex, tvListColumnIndex),
                     isNull);
                 break;
               case INT64:
@@ -705,7 +748,7 @@
                     isNull
                         ? 0
                         : alignedWorkingListForFlush.getLongByValueIndex(
-                            originRowIndex, columnIndex),
+                            originRowIndex, tvListColumnIndex),
                     isNull);
                 break;
               case FLOAT:
@@ -714,7 +757,7 @@
                     isNull
                         ? 0
                         : alignedWorkingListForFlush.getFloatByValueIndex(
-                            originRowIndex, columnIndex),
+                            originRowIndex, tvListColumnIndex),
                     isNull);
                 break;
               case DOUBLE:
@@ -723,7 +766,7 @@
                     isNull
                         ? 0
                         : alignedWorkingListForFlush.getDoubleByValueIndex(
-                            originRowIndex, columnIndex),
+                            originRowIndex, tvListColumnIndex),
                     isNull);
                 break;
               case TEXT:
@@ -735,7 +778,7 @@
                     isNull
                         ? null
                         : alignedWorkingListForFlush.getBinaryByValueIndex(
-                            originRowIndex, columnIndex),
+                            originRowIndex, tvListColumnIndex),
                     isNull);
                 break;
               default:
@@ -775,7 +818,7 @@
   }
 
   @Override
   public void encode(BlockingQueue<Object> ioTaskQueue, BatchEncodeInfo encodeInfo, long[] times) {
     encodeInfo.maxNumberOfPointsInChunk =
         Math.min(
             encodeInfo.maxNumberOfPointsInChunk,
@@ -787,16 +830,21 @@
       return;
     }
 
+    List<IMeasurementSchema> activeSchemaList = getActiveSchemaList();
+    if (activeSchemaList.isEmpty() && ignoreAllNullRows) {
+      return;
+    }
+
     AlignedChunkWriterImpl alignedChunkWriter =
-        new AlignedChunkWriterImpl(schemaList, encryptParameter);
+        new AlignedChunkWriterImpl(activeSchemaList, encryptParameter);
 
     // create MergeSortAlignedTVListIterator.
     List<AlignedTVList> alignedTvLists = new ArrayList<>(sortedList);
     alignedTvLists.add((AlignedTVList) workingListForFlush);
-    List<Integer> columnIndexList = buildColumnIndexList(schemaList);
+    List<Integer> columnIndexList = buildColumnIndexList(activeSchemaList);
     MemPointIterator timeValuePairIterator =
         MemPointIteratorFactory.create(
-            dataTypes,
+            getDataTypes(activeSchemaList),
             columnIndexList,
             alignedTvLists,
             ignoreAllNullRows,
@@ -818,7 +866,7 @@
         } catch (InterruptedException e) {
           Thread.currentThread().interrupt();
         }
-        alignedChunkWriter = new AlignedChunkWriterImpl(schemaList, encryptParameter);
+        alignedChunkWriter = new AlignedChunkWriterImpl(activeSchemaList, encryptParameter);
         encodeInfo.reset();
       }
     }