Ignore data of deprecated table in compaction (apache#16543)

* ignore data of deprecated table in compaction

* fix ut

Author: shuwenwei

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

@@ -125,7 +125,8 @@ public void perform() throws Exception {
                 ? new FastCrossCompactionWriter(targetFiles, seqFiles, readerCacheMap)
                 : new FastInnerCompactionWriter(targetFiles)) {
       List<Schema> schemas =
-          CompactionTableSchemaCollector.collectSchema(seqFiles, unseqFiles, readerCacheMap);
+          CompactionTableSchemaCollector.collectSchema(
+              seqFiles, unseqFiles, readerCacheMap, deviceIterator.getDeprecatedTableSchemaMap());
       compactionWriter.setSchemaForAllTargetFile(schemas);
       readModification(seqFiles);
       readModification(unseqFiles);
@@ -210,6 +211,10 @@ private void compactAlignedSeries(
       measurementSchemas.add(entry.getValue().left);
       timeseriesMetadataOffsetMap.put(entry.getKey(), entry.getValue().right);
     }
+    // current device may be ignored by some conditions
+    if (measurementSchemas.isEmpty()) {
+      return;
+    }
 
     FastCompactionTaskSummary taskSummary = new FastCompactionTaskSummary();
     new FastCompactionPerformerSubTask(

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

@@ -91,7 +91,10 @@ public void perform()
           PageException {
     try (MultiTsFileDeviceIterator deviceIterator = new MultiTsFileDeviceIterator(seqFiles)) {
       schema =
-          CompactionTableSchemaCollector.collectSchema(seqFiles, deviceIterator.getReaderMap());
+          CompactionTableSchemaCollector.collectSchema(
+              seqFiles,
+              deviceIterator.getReaderMap(),
+              deviceIterator.getDeprecatedTableSchemaMap());
       while (deviceIterator.hasNextDevice()) {
         currentWriter = getAvailableCompactionWriter();
         Pair<IDeviceID, Boolean> deviceInfo = deviceIterator.nextDevice();

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

@@ -118,7 +118,10 @@ public void perform() throws Exception {
           new MultiTsFileDeviceIterator(seqFiles, unseqFiles);
       List<Schema> schemas =
           CompactionTableSchemaCollector.collectSchema(
-              seqFiles, unseqFiles, deviceIterator.getReaderMap());
+              seqFiles,
+              unseqFiles,
+              deviceIterator.getReaderMap(),
+              deviceIterator.getDeprecatedTableSchemaMap());
       compactionWriter.setSchemaForAllTargetFile(schemas);
       while (deviceIterator.hasNextDevice()) {
         checkThreadInterrupted();
@@ -271,6 +274,7 @@ public static IDataBlockReader constructReader(
     } else {
       seriesPath = new NonAlignedFullPath(deviceId, measurementSchemas.get(0));
     }
+
     return new SeriesDataBlockReader(
         seriesPath, new HashSet<>(allSensors), fragmentInstanceContext, queryDataSource, true);
   }

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

@@ -33,9 +33,9 @@ public CompactionTableSchema(String tableName) {
     super(tableName);
   }
 
-  public void merge(TableSchema tableSchema) {
+  public boolean merge(TableSchema tableSchema) {
     if (tableSchema == null) {
-      return;
+      return true;
     }
     if (!tableSchema.getTableName().equals(this.tableName)) {
       throw new CompactionTableSchemaNotMatchException(
@@ -60,11 +60,7 @@ public void merge(TableSchema tableSchema) {
       IMeasurementSchema idColumnToMerge = idColumnSchemasToMerge.get(i);
       IMeasurementSchema currentIdColumn = measurementSchemas.get(i);
       if (!idColumnToMerge.getMeasurementName().equals(currentIdColumn.getMeasurementName())) {
-        throw new CompactionTableSchemaNotMatchException(
-            "current id column name is "
-                + currentIdColumn.getMeasurementName()
-                + ", other id column name in same position is "
-                + idColumnToMerge.getMeasurementName());
+        return false;
       }
     }
 
@@ -75,6 +71,7 @@ public void merge(TableSchema tableSchema) {
       columnCategories.add(ColumnCategory.TAG);
       measurementSchemas.add(newIdColumn);
     }
+    return true;
   }
 
   public CompactionTableSchema copy() {

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

@@ -28,8 +28,10 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -39,13 +41,17 @@ private CompactionTableSchemaCollector() {}
   public static List<Schema> collectSchema(
       List<TsFileResource> seqFiles,
       List<TsFileResource> unseqFiles,
-      Map<TsFileResource, TsFileSequenceReader> readerMap)
+      Map<TsFileResource, TsFileSequenceReader> readerMap,
+      Map<TsFileResource, Set<String>> deprecatedTableSchemaMap)
       throws IOException {
     List<Schema> targetSchemas = new ArrayList<>(seqFiles.size());
     Schema schema =
         collectSchema(
-            Stream.concat(seqFiles.stream(), unseqFiles.stream()).collect(Collectors.toList()),
-            readerMap);
+            Stream.concat(seqFiles.stream(), unseqFiles.stream())
+                .sorted(TsFileResource::compareFileName)
+                .collect(Collectors.toList()),
+            readerMap,
+            deprecatedTableSchemaMap);
 
     targetSchemas.add(schema);
     for (int i = 1; i < seqFiles.size(); i++) {
@@ -64,11 +70,14 @@ public static Schema copySchema(Schema source) {
   }
 
   public static Schema collectSchema(
-      List<TsFileResource> sourceFiles, Map<TsFileResource, TsFileSequenceReader> readerMap)
+      List<TsFileResource> sourceFiles,
+      Map<TsFileResource, TsFileSequenceReader> readerMap,
+      Map<TsFileResource, Set<String>> deprecatedTableSchemaMap)
       throws IOException {
     Schema targetSchema = new Schema();
     Map<String, TableSchema> targetTableSchemaMap = new HashMap<>();
-    for (TsFileResource resource : sourceFiles) {
+    for (int i = 0; i < sourceFiles.size(); i++) {
+      TsFileResource resource = sourceFiles.get(i);
       TsFileSequenceReader reader = readerMap.get(resource);
       Map<String, TableSchema> tableSchemaMap = reader.getTableSchemaMap();
       if (tableSchemaMap == null) {
@@ -89,7 +98,19 @@ public static Schema collectSchema(
           collectedTableSchema = new CompactionTableSchema(tableName);
           targetTableSchemaMap.put(tableName, collectedTableSchema);
         }
-        collectedTableSchema.merge(currentTableSchema);
+        boolean canMerge = collectedTableSchema.merge(currentTableSchema);
+        if (!canMerge) {
+          // mark resources with deprecated table schema
+          for (int j = 0; j < i; j++) {
+            deprecatedTableSchemaMap
+                .computeIfAbsent(sourceFiles.get(j), k -> new HashSet<>())
+                .add(tableName);
+          }
+          // replace old table schema in targetTableSchemaMap
+          collectedTableSchema = new CompactionTableSchema(tableName);
+          collectedTableSchema.merge(currentTableSchema);
+          targetTableSchemaMap.put(tableName, collectedTableSchema);
+        }
       }
     }
     targetTableSchemaMap.values().forEach(targetSchema::registerTableSchema);

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

@@ -57,6 +57,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
@@ -69,6 +70,7 @@ public class MultiTsFileDeviceIterator implements AutoCloseable {
   // sort from the oldest to the newest by version (Used by ReadChunkPerformer)
   private List<TsFileResource> tsFileResourcesSortedByAsc;
   private Map<TsFileResource, TsFileSequenceReader> readerMap = new HashMap<>();
+  private final Map<TsFileResource, Set<String>> deprecatedTableSchemaMap = new HashMap<>();
   private final Map<TsFileResource, TsFileDeviceIterator> deviceIteratorMap = new HashMap<>();
   private final Map<TsFileResource, PatternTreeMap<ModEntry, PatternTreeMapFactory.ModsSerializer>>
       modificationCache = new HashMap<>();
@@ -344,6 +346,9 @@ public Map<String, MeasurementSchema> getAllSchemasOfCurrentDevice() throws IOEx
         // which means this tsfile does not contain the current device, then skip it.
         continue;
       }
+      if (isCurrentDeviceDataInDeprecatedTable(resource)) {
+        continue;
+      }
 
       TsFileSequenceReader reader = readerMap.get(resource);
       for (Map.Entry<String, Pair<List<IChunkMetadata>, Pair<Long, Long>>> entrySet :
@@ -408,6 +413,9 @@ public Map<String, MeasurementSchema> getAllSchemasOfCurrentDevice() throws IOEx
       if (!currentDevice.equals(iterator.current())) {
         continue;
       }
+      if (isCurrentDeviceDataInDeprecatedTable(tsFileResource)) {
+        continue;
+      }
       MetadataIndexNode firstMeasurementNodeOfCurrentDevice =
           iterator.getFirstMeasurementNodeOfCurrentDevice();
       TsFileSequenceReader reader = readerMap.get(tsFileResource);
@@ -479,6 +487,10 @@ public Map<TsFileResource, TsFileSequenceReader> getReaderMap() {
     return readerMap;
   }
 
+  public Map<TsFileResource, Set<String>> getDeprecatedTableSchemaMap() {
+    return deprecatedTableSchemaMap;
+  }
+
   @Override
   public void close() throws IOException {
     for (TsFileSequenceReader reader : readerMap.values()) {
@@ -695,4 +707,15 @@ public String nextSeries() throws IllegalPathException {
       return readerAndChunkMetadataForThisSeries;
     }
   }
+
+  // skip data of deleted table
+  private boolean isCurrentDeviceDataInDeprecatedTable(TsFileResource resource) {
+    if (ignoreAllNullRows) {
+      return false;
+    }
+    String tableName = currentDevice.getLeft().getTableName();
+    Set<String> deprecatedTablesInCurrentFile = deprecatedTableSchemaMap.get(resource);
+    return deprecatedTablesInCurrentFile != null
+        && deprecatedTablesInCurrentFile.contains(tableName);
+  }
 }

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableSchemaCollectorTest.java

@@ -19,7 +19,6 @@
 
 package org.apache.iotdb.db.storageengine.dataregion.compaction.tablemodel;
 
-import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionTableSchemaNotMatchException;
 import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionTableSchema;
 
 import org.apache.tsfile.enums.ColumnCategory;
@@ -104,11 +103,6 @@ public void test1() {
     columnTypeList4.add(ColumnCategory.FIELD);
     columnTypeList4.add(ColumnCategory.TAG);
     TableSchema tableSchema4 = new TableSchema("t1", measurementSchemaList4, columnTypeList4);
-    try {
-      compactionTableSchema.merge(tableSchema4);
-    } catch (CompactionTableSchemaNotMatchException e) {
-      return;
-    }
-    Assert.fail();
+    Assert.assertFalse(compactionTableSchema.merge(tableSchema4));
   }
 }