Merge branch 'backport-pipe-optimize-memory-17770' of https://github.com/Caideyipi/iotdb into backport-pipe-optimize-memory-17770

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java

@@ -24,6 +24,7 @@
 import org.apache.iotdb.db.exception.DiskSpaceInsufficientException;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils;
 import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
 import org.apache.iotdb.db.pipe.sink.util.PipeTabletEventSorter;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
@@ -361,7 +362,7 @@ private Tablet tryBestToAggregateTablets(
         // Aggregate the current tablet's data
         aggregatedSchemas.addAll(tablet.getSchemas());
         aggregatedValues.addAll(Arrays.asList(tablet.values));
-        aggregatedBitMaps.addAll(Arrays.asList(tablet.bitMaps));
+        aggregatedBitMaps.addAll(Arrays.asList(PipeTabletUtils.copyBitMapsOrCreateEmpty(tablet)));
         // Remove the aggregated tablet
         tablets.pollFirst();
       } else {
@@ -563,7 +564,7 @@ private void writeTabletsIntoOneFile(
                 .map(schema -> (MeasurementSchema) schema)
                 .toArray(MeasurementSchema[]::new);
         Object[] values = Arrays.copyOf(tablet.values, tablet.values.length);
-        BitMap[] bitMaps = Arrays.copyOf(tablet.bitMaps, tablet.bitMaps.length);
+        BitMap[] bitMaps = PipeTabletUtils.copyBitMapsOrCreateEmpty(tablet);
 
         // convert date value to int refer to
         // org.apache.iotdb.db.storageengine.dataregion.memtable.WritableMemChunk.writeNonAlignedTablet

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeTabletUtilsTest.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.db.pipe.event.common.tablet;
 
 import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.BitMap;
 import org.apache.tsfile.write.record.Tablet;
 import org.apache.tsfile.write.schema.MeasurementSchema;
 import org.junit.Assert;
@@ -48,4 +49,24 @@ public void testPutValueUnmarksReusedNullRow() {
     Assert.assertNull(tablet.bitMaps[0]);
     Assert.assertTrue(tablet.bitMaps[1].isMarked(0));
   }
+
+  @Test
+  public void testCopyBitMapsOrCreateEmptyWithNullBitMaps() {
+    final List<MeasurementSchema> schemas =
+        Arrays.asList(
+            new MeasurementSchema("s1", TSDataType.FLOAT),
+            new MeasurementSchema("s2", TSDataType.FLOAT));
+    final Tablet tablet = new Tablet("root.sg.d1", schemas, 2);
+    tablet.addTimestamp(0, 1L);
+    tablet.addValue("s1", 0, 1.0f);
+    tablet.addValue("s2", 0, 2.0f);
+
+    Assert.assertNull(tablet.bitMaps);
+
+    final BitMap[] bitMaps = PipeTabletUtils.copyBitMapsOrCreateEmpty(tablet);
+
+    Assert.assertEquals(schemas.size(), bitMaps.length);
+    Assert.assertNull(bitMaps[0]);
+    Assert.assertNull(bitMaps[1]);
+  }
 }