Lazy apply primitive array for AlignedTVList

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java

@@ -792,6 +792,7 @@
   @Override
   public void addTVListRamCost(long cost) {
     this.tvListRamCost += cost;
+    System.out.println(tvListRamCost);
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java

@@ -24,6 +24,7 @@
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView;
 import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils;
+import org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager;
 import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
 import org.apache.iotdb.db.utils.datastructure.BatchEncodeInfo;
 import org.apache.iotdb.db.utils.datastructure.MemPointIterator;
@@ -57,6 +58,8 @@
 import java.util.concurrent.BlockingQueue;
 
 import static org.apache.iotdb.db.utils.ModificationUtils.isPointDeleted;
+import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_ARRAY_HEADER;
+import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF;
 
 public class AlignedWritableMemChunk extends AbstractWritableMemChunk {
 
@@ -520,7 +523,7 @@
         ioTaskQueue, chunkRange, timeDuplicateInfo, allValueColDeletedMap, maxNumberOfPointsInPage);
   }
 
   private void handleEncoding(
       BlockingQueue<Object> ioTaskQueue,
       List<List<Integer>> chunkRange,
       boolean[] timeDuplicateInfo,
@@ -884,6 +887,61 @@
     return avgPointSizeOfLargestColumn;
   }
 
+  public long getTvListArrayMemCostIncrement1(
+      List<String> insertingMeasurements, List<TSDataType> insertingTypes) {
+    long size = 0;
+    List<List<BitMap>> bitMaps = list.getBitMaps();
+    // value & bitmap array mem size
+    for (int column = 0; column < dataTypes.size(); column++) {
+      if (bitMaps != null && bitMaps.get(column) != null) {
+        size += (long) PrimitiveArrayManager.ARRAY_SIZE / 8 + 1;
+      }
+    }
+    int newMeasurementCount = 0;
+    for (int i = 0; i < insertingMeasurements.size(); i++) {
+      String measurementName = insertingMeasurements.get(i);
+      TSDataType type = insertingTypes.get(i);
+      size += (long) PrimitiveArrayManager.ARRAY_SIZE * (long) type.getDataTypeSize();
+      if (!measurementIndexMap.containsKey(measurementName)) {
+        newMeasurementCount++;
+      }
+    }
+    // size is 0 when all types are null
+    if (size == 0) {
+      return size;
+    }
+    // time array mem size
+    size += PrimitiveArrayManager.ARRAY_SIZE * 8L;
+    // index array mem size
+    size += (list.getIndices() != null) ? PrimitiveArrayManager.ARRAY_SIZE * 4L : 0;
+    // array headers mem size
+    size += (long) NUM_BYTES_ARRAY_HEADER * (2 + insertingTypes.size());
+    // Object references size in ArrayList
+    size += (long) NUM_BYTES_OBJECT_REF * (2 + dataTypes.size() + newMeasurementCount);
+    return size;
+  }
+
+  public long getTvListArrayMemCostIncrement(
+      List<String> insertingMeasurements, List<TSDataType> insertingTypes) {
+    long memCostIncrement = 0;
+    for (int i = 0; i < insertingMeasurements.size(); i++) {
+      String measurementName = insertingMeasurements.get(i);
+      TSDataType dataType = insertingTypes.get(i);
+      Integer columIndex = measurementIndexMap.get(measurementName);
+      if (columIndex == null) {
+        memCostIncrement +=
+            (long) PrimitiveArrayManager.ARRAY_SIZE * (long) dataType.getDataTypeSize();
+      } else {
+        List<Object> columnArries = list.getValues().get(columIndex);
+        if (columnArries.get(columnArries.size() - 1) == null) {
+          memCostIncrement +=
+              (long) PrimitiveArrayManager.ARRAY_SIZE * (long) dataType.getDataTypeSize();
+        }
+      }
+    }
+    return memCostIncrement;
+  }
+
   @Override
   public void setEncryptParameter(EncryptParameter encryptParameter) {
     this.encryptParameter = encryptParameter;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java

@@ -778,7 +778,6 @@
     } else {
       // For existed device of this mem table
       AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk) memChunk;
-      List<TSDataType> dataTypesInTVList = new ArrayList<>();
       for (int i = 0; i < dataTypes.length; i++) {
         // Skip failed Measurements
         if (dataTypes[i] == null
@@ -794,15 +793,13 @@
                   + (alignedMemChunk.alignedListSize() % PrimitiveArrayManager.ARRAY_SIZE > 0
                       ? 1
                       : 0);
-          memTableIncrement += currentArrayNum * AlignedTVList.valueListArrayMemCost(dataTypes[i]);
-          dataTypesInTVList.add(dataTypes[i]);
+          memTableIncrement += currentArrayNum * AlignedTVList.emptyValueListArrayMemCost();
         }
       }
       // this insertion will result in a new array
-      if ((alignedMemChunk.alignedListSize() % PrimitiveArrayManager.ARRAY_SIZE) == 0) {
-        dataTypesInTVList.addAll(alignedMemChunk.getWorkingTVList().getTsDataTypes());
-        memTableIncrement += alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost();
-      }
+      //      if ((alignedMemChunk.alignedListSize() % PrimitiveArrayManager.ARRAY_SIZE) == 0) {
+      //        memTableIncrement += alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost();
+      //      }
     }
 
     for (int i = 0; i < dataTypes.length; i++) {
@@ -816,7 +813,7 @@
   }
 
   @SuppressWarnings("squid:S3776") // high Cognitive Complexity
   private long[] checkAlignedMemCostAndAddToTspInfoForRows(List<InsertRowNode> insertRowNodeList)
       throws WriteProcessException {
     // Memory of increased PrimitiveArray and TEXT values, e.g., add a long[128], add 128*8
     long memTableIncrement = 0L;
@@ -882,8 +879,7 @@
                             > 0
                         ? 1
                         : 0);
-            memTableIncrement +=
-                currentArrayNum * AlignedTVList.valueListArrayMemCost(dataTypes[i]);
+            memTableIncrement += currentArrayNum * AlignedTVList.emptyValueListArrayMemCost();
           }
         }
         int addingPointNum = addingPointNumInfo.right;
@@ -893,11 +889,11 @@
             dataTypesInTVList.addAll(alignedMemChunk.getWorkingTVList().getTsDataTypes());
           }
           dataTypesInTVList.addAll(addingPointNumInfo.left.values());
-          memTableIncrement +=
-              alignedMemChunk != null
-                  ? alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost()
-                  : AlignedTVList.alignedTvListArrayMemCost(
-                      dataTypesInTVList.toArray(new TSDataType[0]), null);
+          //          memTableIncrement +=
+          //              alignedMemChunk != null
+          //                  ? alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost()
+          //                  : AlignedTVList.alignedTvListArrayMemCost(
+          //                      dataTypesInTVList.toArray(new TSDataType[0]), null);
         }
         addingPointNumInfo.setRight(addingPointNum + 1);
       }
@@ -1020,7 +1016,7 @@
     }
   }
 
   private void updateAlignedMemCost(
       TSDataType[] dataTypes,
       IDeviceID deviceId,
       String[] measurementIds,
@@ -1072,9 +1068,18 @@
           numArraysToAdd * AlignedTVList.alignedTvListArrayMemCost(dataTypes, columnCategories);
     } else {
       AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk) memChunk;
-      List<TSDataType> dataTypesInTVList = new ArrayList<>();
       int currentPointNum = alignedMemChunk.alignedListSize();
       int newPointNum = currentPointNum + incomingPointNum;
+      // calculate how many new arrays will be added after this insertion
+      int currentArrayCnt =
+          currentPointNum / PrimitiveArrayManager.ARRAY_SIZE
+              + (currentPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0);
+      int newArrayCnt =
+          newPointNum / PrimitiveArrayManager.ARRAY_SIZE
+              + (newPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0);
+      long acquireArray = newArrayCnt - currentArrayCnt;
+      List<String> insertingMeasurements = new ArrayList<>();
+      List<TSDataType> insertingTypes = new ArrayList<>();
       for (int i = 0; i < dataTypes.length; i++) {
         TSDataType dataType = dataTypes[i];
         String measurement = measurementIds[i];
@@ -1085,30 +1090,24 @@
             || (columnCategories != null && columnCategories[i] != TsTableColumnCategory.FIELD)) {
           continue;
         }
+        insertingMeasurements.add(measurement);
+        insertingTypes.add(dataType);
 
         if (!alignedMemChunk.containsMeasurement(measurementIds[i])) {
           // add a new column in the TVList, the new column should be as long as existing ones
-          memIncrements[0] +=
-              (currentPointNum / PrimitiveArrayManager.ARRAY_SIZE + 1)
-                  * AlignedTVList.valueListArrayMemCost(dataType);
-          dataTypesInTVList.add(dataType);
+          memIncrements[0] += newArrayCnt * AlignedTVList.emptyValueListArrayMemCost();
         }
       }
 
-      // calculate how many new arrays will be added after this insertion
-      int currentArrayCnt =
-          currentPointNum / PrimitiveArrayManager.ARRAY_SIZE
-              + (currentPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0);
-      int newArrayCnt =
-          newPointNum / PrimitiveArrayManager.ARRAY_SIZE
-              + (newPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0);
-      long acquireArray = newArrayCnt - currentArrayCnt;
-
       if (acquireArray != 0) {
         // memory of extending the TVList
-        dataTypesInTVList.addAll(alignedMemChunk.getWorkingTVList().getTsDataTypes());
         memIncrements[0] +=
-            acquireArray * alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost();
+            acquireArray
+                * alignedMemChunk.getTvListArrayMemCostIncrement1(
+                    insertingMeasurements, insertingTypes);
+      } else {
+        memIncrements[0] +=
+            alignedMemChunk.getTvListArrayMemCostIncrement(insertingMeasurements, insertingTypes);
       }
     }