[iotdb] branch rel/1.1 updated: [To rel/1.1] Fix compaction force decoding check (#10803)

[xingtanzjr]

This is an automated email from the ASF dual-hosted git repository.

xingtanzjr pushed a commit to branch rel/1.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/rel/1.1 by this push:
     new f580f5fac10 [To rel/1.1] Fix compaction force decoding check (#10803)
f580f5fac10 is described below

commit f580f5fac10b4f532b2b64873f2fa294e632ece5
Author: shuwenwei <55970239+shuwen...@users.noreply.github.com>
AuthorDate: Tue Aug 8 17:04:42 2023 +0800

    [To rel/1.1] Fix compaction force decoding check (#10803)
---
 .../fast/AlignedSeriesCompactionExecutor.java      | 28 ++++++++++++++++------
 1 file changed, 21 insertions(+), 7 deletions(-)

diff --git 
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/utils/executor/fast/AlignedSeriesCompactionExecutor.java
 
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/utils/executor/fast/AlignedSeriesCompactionExecutor.java
index a6c11dd5a2a..c5b75ccf0bf 100644
--- 
a/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/utils/executor/fast/AlignedSeriesCompactionExecutor.java
+++ 
b/server/src/main/java/org/apache/iotdb/db/engine/compaction/execute/utils/executor/fast/AlignedSeriesCompactionExecutor.java
@@ -46,6 +46,7 @@ import 
org.apache.iotdb.tsfile.write.schema.IMeasurementSchema;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -57,6 +58,8 @@ public class AlignedSeriesCompactionExecutor extends 
SeriesCompactionExecutor {
   private final Map<String, Map<TsFileResource, Pair<Long, Long>>> 
timeseriesMetadataOffsetMap;
 
   private final List<IMeasurementSchema> measurementSchemas;
+  private final IMeasurementSchema timeColumnMeasurementSchema;
+  private final Map<String, IMeasurementSchema> measurementSchemaMap;
 
   public AlignedSeriesCompactionExecutor(
       AbstractCompactionWriter compactionWriter,
@@ -72,6 +75,10 @@ public class AlignedSeriesCompactionExecutor extends 
SeriesCompactionExecutor {
         compactionWriter, readerCacheMap, modificationCacheMap, deviceId, 
true, subTaskId, summary);
     this.timeseriesMetadataOffsetMap = timeseriesMetadataOffsetMap;
     this.measurementSchemas = measurementSchemas;
+    this.timeColumnMeasurementSchema = measurementSchemas.get(0);
+    this.measurementSchemaMap = new HashMap<>();
+    this.measurementSchemas.forEach(
+        schema -> measurementSchemaMap.put(schema.getMeasurementId(), schema));
     // get source files which are sorted by the startTime of current device 
from old to new,
     // files that do not contain the current device have been filtered out as 
well.
     sortedSourceFiles.forEach(x -> fileList.add(new FileElement(x)));
@@ -315,18 +322,25 @@ public class AlignedSeriesCompactionExecutor extends 
SeriesCompactionExecutor {
   }
 
   void setForceDecoding(ChunkMetadataElement chunkMetadataElement) {
-    IMeasurementSchema timeChunkSchema = measurementSchemas.get(0);
-    if (timeChunkSchema.getCompressor()
+    if (timeColumnMeasurementSchema.getCompressor()
             != chunkMetadataElement.chunk.getHeader().getCompressionType()
-        || timeChunkSchema.getEncodingType()
+        || timeColumnMeasurementSchema.getEncodingType()
             != chunkMetadataElement.chunk.getHeader().getEncodingType()) {
       chunkMetadataElement.needForceDecoding = true;
       return;
     }
-    for (int i = 1; i < measurementSchemas.size(); i++) {
-      ChunkHeader header = chunkMetadataElement.valueChunks.get(i - 
1).getHeader();
-      if (header.getCompressionType() != 
measurementSchemas.get(i).getCompressor()
-          || header.getEncodingType() != 
measurementSchemas.get(i).getEncodingType()) {
+    for (Chunk chunk : chunkMetadataElement.valueChunks) {
+      if (chunk == null) {
+        continue;
+      }
+      ChunkHeader header = chunk.getHeader();
+      String measurementId = header.getMeasurementID();
+      IMeasurementSchema measurementSchema = 
measurementSchemaMap.get(measurementId);
+      if (measurementSchema == null) {
+        continue;
+      }
+      if (measurementSchema.getCompressor() != header.getCompressionType()
+          || measurementSchema.getEncodingType() != header.getEncodingType()) {
         chunkMetadataElement.needForceDecoding = true;
         return;
       }