Keep all null rows in table model compaction (#13523)

* keep all null rows in table model compaction

* add ut

* adddeletion ut

* add ut

* revert some ut

* fix ut

* add context in cross space compaction task selection

* fix bug

* spotless

* modify repair task

* update tsfile version

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanOperator.java

@@ -64,8 +64,7 @@ public AlignedSeriesScanOperator(
             seriesScanOptions,
             context.getInstanceContext(),
             queryAllSensors,
-            dataTypes,
-            true);
+            dataTypes);
     this.valueColumnCount = seriesPath.getColumnNum();
     this.maxReturnSize =
         Math.min(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java

@@ -59,7 +59,7 @@ public AlignedSeriesScanUtil(
       Ordering scanOrder,
       SeriesScanOptions scanOptions,
       FragmentInstanceContext context) {
-    this(seriesPath, scanOrder, scanOptions, context, false, null, true);
+    this(seriesPath, scanOrder, scanOptions, context, false, null);
   }
 
   public AlignedSeriesScanUtil(
@@ -68,8 +68,7 @@ public AlignedSeriesScanUtil(
       SeriesScanOptions scanOptions,
       FragmentInstanceContext context,
       boolean queryAllSensors,
-      List<TSDataType> givenDataTypes,
-      boolean ignoreAllNullRows) {
+      List<TSDataType> givenDataTypes) {
     super(seriesPath, scanOrder, scanOptions, context);
     isAligned = true;
     this.dataTypes =
@@ -79,7 +78,7 @@ public AlignedSeriesScanUtil(
                 .map(IMeasurementSchema::getType)
                 .collect(Collectors.toList());
     this.queryAllSensors = queryAllSensors;
-    this.ignoreAllNullRows = ignoreAllNullRows;
+    this.ignoreAllNullRows = context.isIgnoreAllNullRows();
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableScanOperator.java

@@ -306,8 +306,7 @@ private AlignedSeriesScanUtil constructAlignedSeriesScanUtil(DeviceEntry deviceE
         seriesScanOptions,
         operatorContext.getInstanceContext(),
         true,
-        measurementColumnTSDataTypes,
-        false);
+        measurementColumnTSDataTypes);
   }
 
   public static AlignedFullPath constructAlignedPath(

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

@@ -313,6 +313,8 @@ public class DataRegion implements IDataRegionForQuery {
 
   private final AtomicBoolean isCompactionSelecting = new AtomicBoolean(false);
 
+  private boolean isTreeModel;
+
   private static final QueryResourceMetricSet QUERY_RESOURCE_METRIC_SET =
       QueryResourceMetricSet.getInstance();
 
@@ -333,6 +335,7 @@ public DataRegion(
     this.dataRegionId = dataRegionId;
     this.databaseName = databaseName;
     this.fileFlushPolicy = fileFlushPolicy;
+    this.isTreeModel = this.databaseName.startsWith("root.");
     acquireDirectBufferMemory();
 
     dataRegionSysDir = SystemFileFactory.INSTANCE.getFile(systemDir, dataRegionId);
@@ -383,6 +386,7 @@ public DataRegion(String databaseName, String id) {
     this.dataRegionId = id;
     this.tsFileManager = new TsFileManager(databaseName, id, "");
     this.partitionMaxFileVersions = new HashMap<>();
+    this.isTreeModel = this.databaseName.startsWith("root.");
     partitionMaxFileVersions.put(0L, 0L);
   }
 
@@ -2765,7 +2769,7 @@ public int executeCompaction() throws InterruptedException {
       // Sort the time partition from largest to smallest
       timePartitions.sort(Comparator.reverseOrder());
 
-      CompactionScheduleContext context = new CompactionScheduleContext();
+      CompactionScheduleContext context = new CompactionScheduleContext(this.isTreeModel);
 
       // schedule insert compaction
       trySubmitCount += executeInsertionCompaction(timePartitions, context);
@@ -2808,7 +2812,7 @@ public int executeTTLCheck() throws InterruptedException {
     logger.info("[TTL] {}-{} Start ttl checking.", databaseName, dataRegionId);
     int trySubmitCount = 0;
     try {
-      CompactionScheduleContext context = new CompactionScheduleContext();
+      CompactionScheduleContext context = new CompactionScheduleContext(this.isTreeModel);
       List<Long> timePartitions = new ArrayList<>(tsFileManager.getTimePartitions());
       // Sort the time partition from smallest to largest
       Collections.sort(timePartitions);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/ICompactionPerformer.java

@@ -40,6 +40,8 @@ public interface ICompactionPerformer {
 
   void setSummary(CompactionTaskSummary summary);
 
+  void setIgnoreAllNullRows(boolean ignoreAllNullRows);
+
   default void setSourceFiles(List<TsFileResource> files) {
     throw new IllegalSourceFileTypeException(
         "Cannot set single type of source files to this kind of performer");

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java

@@ -91,6 +91,7 @@ public class FastCompactionPerformer
       modificationCache = new ConcurrentHashMap<>();
 
   private final boolean isCrossCompaction;
+  private boolean ignoreAllNullRows = true;
 
   public FastCompactionPerformer(
       List<TsFileResource> seqFiles,
@@ -115,7 +116,7 @@ public FastCompactionPerformer(boolean isCrossCompaction) {
   public void perform() throws Exception {
     this.subTaskSummary.setTemporalFileNum(targetFiles.size());
     try (MultiTsFileDeviceIterator deviceIterator =
-            new MultiTsFileDeviceIterator(seqFiles, unseqFiles, readerCacheMap);
+            new MultiTsFileDeviceIterator(seqFiles, unseqFiles, readerCacheMap, ignoreAllNullRows);
         AbstractCompactionWriter compactionWriter =
             isCrossCompaction
                 ? new FastCrossCompactionWriter(targetFiles, seqFiles, readerCacheMap)
@@ -192,7 +193,7 @@ private void compactAlignedSeries(
     // end offset of each timeseries metadata, in order to facilitate the reading of chunkMetadata
     // directly by this offset later. Instead of deserializing chunk metadata later, we need to
     // deserialize chunk metadata here to get the schemas of all value measurements, because we
-    // should get schemas of all value measurement to startMeasruement() and compaction process is
+    // should get schemas of all value measurement to startMeasurement() and compaction process is
     // to read a batch of overlapped files each time, and we cannot make sure if the first batch of
     // overlapped tsfiles contain all the value measurements.
     for (Map.Entry<String, Pair<MeasurementSchema, Map<TsFileResource, Pair<Long, Long>>>> entry :
@@ -210,7 +211,8 @@ private void compactAlignedSeries(
             sortedSourceFiles,
             measurementSchemas,
             deviceId,
-            taskSummary)
+            taskSummary,
+            ignoreAllNullRows)
         .call();
     subTaskSummary.increase(taskSummary);
   }
@@ -296,6 +298,11 @@ public void setSummary(CompactionTaskSummary summary) {
     this.subTaskSummary = (FastCompactionTaskSummary) summary;
   }
 
+  @Override
+  public void setIgnoreAllNullRows(boolean ignoreAllNullRows) {
+    this.ignoreAllNullRows = ignoreAllNullRows;
+  }
+
   @Override
   public void setSourceFiles(List<TsFileResource> seqFiles, List<TsFileResource> unseqFiles) {
     this.seqFiles = seqFiles;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadChunkCompactionPerformer.java

@@ -55,6 +55,7 @@ public class ReadChunkCompactionPerformer implements ISeqCompactionPerformer {
   private CompactionTsFileWriter currentWriter;
   private long endedFileSize = 0;
   private int currentTargetFileIndex = 0;
+  private boolean ignoreAllNullRows = true;
   // memory budget for file writer is 5% of per compaction task memory budget
   private final long memoryBudgetForFileWriter =
       (long)
@@ -86,7 +87,8 @@ public void perform()
           InterruptedException,
           StorageEngineException,
           PageException {
-    try (MultiTsFileDeviceIterator deviceIterator = new MultiTsFileDeviceIterator(seqFiles)) {
+    try (MultiTsFileDeviceIterator deviceIterator =
+        new MultiTsFileDeviceIterator(seqFiles, ignoreAllNullRows)) {
       schema =
           CompactionTableSchemaCollector.collectSchema(seqFiles, deviceIterator.getReaderMap());
       while (deviceIterator.hasNextDevice()) {
@@ -172,6 +174,11 @@ public void setSummary(CompactionTaskSummary summary) {
     this.summary = summary;
   }
 
+  @Override
+  public void setIgnoreAllNullRows(boolean ignoreAllNullRows) {
+    this.ignoreAllNullRows = ignoreAllNullRows;
+  }
+
   private void compactAlignedSeries(
       IDeviceID device,
       TsFileResource targetResource,
@@ -187,7 +194,7 @@ private void compactAlignedSeries(
     writer.startChunkGroup(device);
     BatchedReadChunkAlignedSeriesCompactionExecutor compactionExecutor =
         new BatchedReadChunkAlignedSeriesCompactionExecutor(
-            device, targetResource, readerAndChunkMetadataList, writer, summary);
+            device, targetResource, readerAndChunkMetadataList, writer, summary, ignoreAllNullRows);
     compactionExecutor.execute();
     for (ChunkMetadata chunkMetadata : writer.getChunkMetadataListOfCurrentDeviceInMemory()) {
       if (chunkMetadata.getMeasurementUid().isEmpty()) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/ReadPointCompactionPerformer.java

@@ -79,6 +79,7 @@ public class ReadPointCompactionPerformer
   private CompactionTaskSummary summary;
 
   protected List<TsFileResource> targetFiles = Collections.emptyList();
+  protected boolean ignoreAllNullRows = true;
 
   public ReadPointCompactionPerformer(
       List<TsFileResource> seqFiles,
@@ -112,7 +113,7 @@ public void perform() throws Exception {
         getCompactionWriter(seqFiles, unseqFiles, targetFiles)) {
       // Do not close device iterator, because tsfile reader is managed by FileReaderManager.
       MultiTsFileDeviceIterator deviceIterator =
-          new MultiTsFileDeviceIterator(seqFiles, unseqFiles);
+          new MultiTsFileDeviceIterator(seqFiles, unseqFiles, ignoreAllNullRows);
       List<Schema> schemas =
           CompactionTableSchemaCollector.collectSchema(
               seqFiles, unseqFiles, deviceIterator.getReaderMap());
@@ -152,6 +153,11 @@ public void setSummary(CompactionTaskSummary summary) {
     this.summary = summary;
   }
 
+  @Override
+  public void setIgnoreAllNullRows(boolean ignoreAllNullRows) {
+    this.ignoreAllNullRows = ignoreAllNullRows;
+  }
+
   private void compactAlignedSeries(
       IDeviceID device,
       MultiTsFileDeviceIterator deviceIterator,
@@ -169,6 +175,8 @@ private void compactAlignedSeries(
         measurementSchemas.stream()
             .map(IMeasurementSchema::getMeasurementId)
             .collect(Collectors.toList());
+
+    fragmentInstanceContext.setIgnoreAllNullRows(ignoreAllNullRows);
     IDataBlockReader dataBlockReader =
         constructReader(
             device,

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/RepairUnsortedFileCompactionPerformer.java

@@ -36,9 +36,10 @@ public class RepairUnsortedFileCompactionPerformer extends ReadPointCompactionPe
 
   private final boolean rewriteFile;
 
-  public RepairUnsortedFileCompactionPerformer(boolean rewriteFile) {
+  public RepairUnsortedFileCompactionPerformer(boolean rewriteFile, boolean ignoreAllNullRows) {
     super();
     this.rewriteFile = rewriteFile;
+    this.ignoreAllNullRows = ignoreAllNullRows;
   }
 
   @Override

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/RepairUnsortedFileCompactionTask.java

@@ -63,13 +63,14 @@ public RepairUnsortedFileCompactionTask(
       TsFileManager tsFileManager,
       TsFileResource sourceFile,
       boolean sequence,
-      long serialId) {
+      long serialId,
+      boolean ignoreAllNullRows) {
     super(
         timePartition,
         tsFileManager,
         Collections.singletonList(sourceFile),
         sequence,
-        new RepairUnsortedFileCompactionPerformer(true),
+        new RepairUnsortedFileCompactionPerformer(true, ignoreAllNullRows),
         serialId);
     this.sourceFile = sourceFile;
     this.innerSpaceEstimator = new RepairUnsortedFileCompactionEstimator();
@@ -82,13 +83,14 @@ public RepairUnsortedFileCompactionTask(
       TsFileResource sourceFile,
       boolean sequence,
       boolean rewriteFile,
-      long serialId) {
+      long serialId,
+      boolean ignoreAllNullRows) {
     super(
         timePartition,
         tsFileManager,
         Collections.singletonList(sourceFile),
         sequence,
-        new RepairUnsortedFileCompactionPerformer(rewriteFile),
+        new RepairUnsortedFileCompactionPerformer(rewriteFile, ignoreAllNullRows),
         serialId);
     this.sourceFile = sourceFile;
     if (rewriteFile) {
@@ -103,13 +105,14 @@ public RepairUnsortedFileCompactionTask(
       TsFileResource sourceFile,
       CountDownLatch latch,
       boolean sequence,
-      long serialId) {
+      long serialId,
+      boolean ignoreAllNullRows) {
     super(
         timePartition,
         tsFileManager,
         Collections.singletonList(sourceFile),
         sequence,
-        new RepairUnsortedFileCompactionPerformer(true),
+        new RepairUnsortedFileCompactionPerformer(true, ignoreAllNullRows),
         serialId);
     this.sourceFile = sourceFile;
     this.innerSpaceEstimator = new RepairUnsortedFileCompactionEstimator();
@@ -124,13 +127,14 @@ public RepairUnsortedFileCompactionTask(
       CountDownLatch latch,
       boolean sequence,
       boolean rewriteFile,
-      long serialId) {
+      long serialId,
+      boolean ignoreAllNullRows) {
     super(
         timePartition,
         tsFileManager,
         Collections.singletonList(sourceFile),
         sequence,
-        new RepairUnsortedFileCompactionPerformer(rewriteFile),
+        new RepairUnsortedFileCompactionPerformer(rewriteFile, ignoreAllNullRows),
         serialId);
     this.sourceFile = sourceFile;
     if (rewriteFile) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/subtask/FastCompactionPerformerSubTask.java

@@ -67,6 +67,8 @@ public class FastCompactionPerformerSubTask implements Callable<Void> {
 
   private final boolean isAligned;
 
+  private final boolean ignoreAllNullRows;
+
   private IDeviceID deviceId;
 
   private List<String> measurements;
@@ -96,6 +98,7 @@ public FastCompactionPerformerSubTask(
     this.sortedSourceFiles = sortedSourceFiles;
     this.measurements = measurements;
     this.summary = summary;
+    this.ignoreAllNullRows = true;
   }
 
   /** Used for aligned timeseries. */
@@ -108,7 +111,8 @@ public FastCompactionPerformerSubTask(
       List<TsFileResource> sortedSourceFiles,
       List<IMeasurementSchema> measurementSchemas,
       IDeviceID deviceId,
-      FastCompactionTaskSummary summary) {
+      FastCompactionTaskSummary summary,
+      boolean ignoreAllNullRows) {
     this.compactionWriter = compactionWriter;
     this.subTaskId = 0;
     this.timeseriesMetadataOffsetMap = timeseriesMetadataOffsetMap;
@@ -119,6 +123,7 @@ public FastCompactionPerformerSubTask(
     this.sortedSourceFiles = sortedSourceFiles;
     this.measurementSchemas = measurementSchemas;
     this.summary = summary;
+    this.ignoreAllNullRows = ignoreAllNullRows;
   }
 
   @Override
@@ -154,7 +159,8 @@ public Void call()
                 deviceId,
                 subTaskId,
                 measurementSchemas,
-                summary);
+                summary,
+                ignoreAllNullRows);
       } else {
         seriesCompactionExecutor =
             new FastAlignedSeriesCompactionExecutor(
@@ -166,7 +172,8 @@ public Void call()
                 deviceId,
                 subTaskId,
                 measurementSchemas,
-                summary);
+                summary,
+                ignoreAllNullRows);
       }
       seriesCompactionExecutor.execute();
     }