[To dev/1.3] perf: Optimize the mark bitmap operation of AlignedTVList InsertTablet (#16199) (#16590)

(cherry picked from commit 51fe585)

Author: luoluoyuyu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimator.java

@@ -553,10 +553,12 @@ private static long getBinarySize(final Binary[] binaries) {
       return 0L;
     }
 
-    return RamUsageEstimator.shallowSizeOf(binaries)
-        + Arrays.stream(binaries)
-            .mapToLong(InsertNodeMemoryEstimator::sizeOfBinary)
-            .reduce(0L, Long::sum);
+    long size = 0L;
+    for (Binary binary : binaries) {
+      size += InsertNodeMemoryEstimator.sizeOfBinary(binary);
+    }
+
+    return size + RamUsageEstimator.shallowSizeOf(binaries);
   }
 
   public static long sizeOfValues(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java

@@ -712,14 +712,9 @@ public synchronized void putAlignedValues(
           if (indices != null) {
             indices.get(arrayIdx)[elementIdx + i] = rowCount;
           }
-          for (int j = 0; j < values.size(); j++) {
-            if (value[j] == null
-                || bitMaps != null && bitMaps[j] != null && bitMaps[j].isMarked(idx + i)) {
-              markNullValue(j, arrayIdx, elementIdx + i);
-            }
-          }
           rowCount++;
         }
+        markNullBitmapRange(value, bitMaps, idx, elementIdx, inputRemaining, arrayIdx);
         break;
       } else {
         // the remaining inputs cannot fit the last array, fill the last array and create a new
@@ -730,20 +725,31 @@ public synchronized void putAlignedValues(
           if (indices != null) {
             indices.get(arrayIdx)[elementIdx + i] = rowCount;
           }
-          for (int j = 0; j < values.size(); j++) {
-            if (value[j] == null
-                || bitMaps != null && bitMaps[j] != null && bitMaps[j].isMarked(idx + i)) {
-              markNullValue(j, arrayIdx, elementIdx + i);
-            }
-          }
           rowCount++;
         }
+        markNullBitmapRange(value, bitMaps, idx, elementIdx, internalRemaining, arrayIdx);
         idx += internalRemaining;
         checkExpansion();
       }
     }
   }
 
+  private void markNullBitmapRange(
+      Object[] values, BitMap[] bitMaps, int idx, int elementIdx, int len, int arrayIndex) {
+    for (int j = 0; j < values.length; j++) {
+      /* Fast-path: column is entirely null */
+      if (values[j] == null) {
+        getBitMap(j, arrayIndex).markRange(elementIdx, len);
+        continue;
+      }
+
+      /* 2.mask the column bitmap */
+      if (bitMaps != null && bitMaps[j] != null) {
+        getBitMap(j, arrayIndex).merge(bitMaps[j], idx, elementIdx, len);
+      }
+    }
+  }
+
   private void arrayCopy(Object[] value, int idx, int arrayIndex, int elementIndex, int remaining) {
     for (int i = 0; i < values.size(); i++) {
       if (value[i] == null) {
@@ -791,7 +797,7 @@ private void arrayCopy(Object[] value, int idx, int arrayIndex, int elementIndex
     }
   }
 
-  private void markNullValue(int columnIndex, int arrayIndex, int elementIndex) {
+  private BitMap getBitMap(int columnIndex, int arrayIndex) {
     // init BitMaps if doesn't have
     if (bitMaps == null) {
       List<List<BitMap>> localBitMaps = new ArrayList<>(dataTypes.size());
@@ -805,18 +811,22 @@ private void markNullValue(int columnIndex, int arrayIndex, int elementIndex) {
     if (bitMaps.get(columnIndex) == null) {
       List<BitMap> columnBitMaps = new ArrayList<>();
       for (int i = 0; i < values.get(columnIndex).size(); i++) {
-        columnBitMaps.add(new BitMap(ARRAY_SIZE));
+        columnBitMaps.add(new BitMap(ARRAY_SIZE, new byte[ARRAY_SIZE]));
       }
       bitMaps.set(columnIndex, columnBitMaps);
     }
 
     // if the bitmap in arrayIndex is null, init the bitmap
     if (bitMaps.get(columnIndex).get(arrayIndex) == null) {
-      bitMaps.get(columnIndex).set(arrayIndex, new BitMap(ARRAY_SIZE));
+      bitMaps.get(columnIndex).set(arrayIndex, new BitMap(ARRAY_SIZE, new byte[ARRAY_SIZE]));
     }
 
+    return bitMaps.get(columnIndex).get(arrayIndex);
+  }
+
+  private void markNullValue(int columnIndex, int arrayIndex, int elementIndex) {
     // mark the null value in the current bitmap
-    bitMaps.get(columnIndex).get(arrayIndex).mark(elementIndex);
+    getBitMap(columnIndex, arrayIndex).mark(elementIndex);
   }
 
   @Override
@@ -1257,33 +1267,50 @@ public BitMap getAllValueColDeletedMap() {
     }
 
     byte[] rowBitsArr = new byte[rowCount / Byte.SIZE + 1];
-    for (int row = 0; row < rowCount; row += Byte.SIZE) {
-      boolean isFirstColumn = true;
-      byte rowBits = 0x00;
-      for (int columnIndex = 0; columnIndex < values.size(); columnIndex++) {
-        List<BitMap> columnBitMaps = bitMaps.get(columnIndex);
-        byte columnBits;
-        if (values.get(columnIndex) == null) {
-          columnBits = (byte) 0xFF;
-        } else if (columnBitMaps == null || columnBitMaps.get(row / ARRAY_SIZE) == null) {
-          // row exists when any column value exists
-          rowBits = 0x00;
-          break;
-        } else {
-          columnBits =
-              columnBitMaps.get(row / ARRAY_SIZE).getByteArray()[(row % ARRAY_SIZE) / Byte.SIZE];
+    int bitsMapSize =
+        rowCount % ARRAY_SIZE == 0 ? rowCount / ARRAY_SIZE : rowCount / ARRAY_SIZE + 1;
+    boolean[] allNotNullArray = new boolean[bitsMapSize];
+    Arrays.fill(rowBitsArr, (byte) 0xFF);
+    for (int columnIndex = 0; columnIndex < values.size(); columnIndex++) {
+      List<BitMap> columnBitMaps = bitMaps.get(columnIndex);
+      if (columnBitMaps == null) {
+        Arrays.fill(rowBitsArr, (byte) 0x00);
+        break;
+      } else if (values.get(columnIndex) != null) {
+        int row = 0;
+        boolean isEnd = true;
+        for (int i = 0; i < bitsMapSize; i++) {
+          if (allNotNullArray[i]) {
+            row += ARRAY_SIZE;
+            continue;
+          }
+
+          BitMap bitMap = columnBitMaps.get(i);
+          int index = row / Byte.SIZE;
+          int size = ((Math.min((rowCount - row), ARRAY_SIZE)) + 7) >>> 3;
+          row += ARRAY_SIZE;
+
+          if (bitMap == null) {
+            Arrays.fill(rowBitsArr, index, index + size, (byte) 0x00);
+            allNotNullArray[i] = true;
+            continue;
+          }
+
+          byte bits = (byte) 0X00;
+          for (int j = 0; j < size; j++) {
+            rowBitsArr[index] &= bitMap.getByteArray()[j];
+            bits |= rowBitsArr[index++];
+            isEnd = false;
+          }
+
+          allNotNullArray[i] = bits == (byte) 0;
         }
-        // set row to null when all column values are null
-        if (isFirstColumn) {
-          rowBits = columnBits;
-          isFirstColumn = false;
-        } else {
-          rowBits &= columnBits;
+
+        if (isEnd) {
+          break;
         }
       }
-      rowBitsArr[row / Byte.SIZE] = rowBits;
     }
-
     return new BitMap(rowCount, rowBitsArr);
   }