This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_flat_device in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit fb00c8335d8195dae354bf71eb5238616eac6f4a Author: Tian Jiang <[email protected]> AuthorDate: Thu Sep 17 11:11:05 2026 +0800 fix compaction --- .../performer/impl/FastCompactionPerformer.java | 12 +- .../CompactionTableModelTestFileWriter.java | 15 ++ .../tablemodel/FlatDeviceCompactionTest.java | 226 +++++++++++++++++++++ 3 files changed, 252 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java index b8e0c3ed8fd..efd8c3e3736 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/performer/impl/FastCompactionPerformer.java @@ -76,6 +76,7 @@ import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; +import java.util.stream.Stream; public class FastCompactionPerformer implements ICrossCompactionPerformer, ISeqCompactionPerformer, IUnseqCompactionPerformer { @@ -162,8 +163,17 @@ public class FastCompactionPerformer || FlatDeviceCompactionUtils.containsFlatDevice(unseqFiles)) { // Fast compaction deduplicates by timestamp and can split TAG columns across batches. // Keep the complete flat-device row key together in the read-point path. + // The single-list setter stores both kinds of inner-space sources in seqFiles. Restore + // their actual classification before the query reader assumes sequence files don't overlap. + List<TsFileResource> readPointSeqFiles = new ArrayList<>(); + List<TsFileResource> readPointUnseqFiles = new ArrayList<>(); + Stream.concat(seqFiles.stream(), unseqFiles.stream()) + .forEach( + resource -> + (resource.isSeq() ? readPointSeqFiles : readPointUnseqFiles).add(resource)); ReadPointCompactionPerformer performer = - new ReadPointCompactionPerformer(seqFiles, unseqFiles, targetFiles, encryptParameter); + new ReadPointCompactionPerformer( + readPointSeqFiles, readPointUnseqFiles, targetFiles, encryptParameter); performer.setSummary(subTaskSummary); performer.perform(); return; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java index 302b33d2145..31fde52a9de 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/CompactionTableModelTestFileWriter.java @@ -97,6 +97,17 @@ public class CompactionTableModelTestFileWriter extends CompactionTestFileWriter public void generateFlatDeviceAlignedRows( long[] timestamps, String[] tag1Values, long[] fieldValues, String[] tag2Values) throws IOException { + generateFlatDeviceAlignedRows( + timestamps, tag1Values, fieldValues, tag2Values, Integer.MAX_VALUE); + } + + public void generateFlatDeviceAlignedRows( + long[] timestamps, + String[] tag1Values, + long[] fieldValues, + String[] tag2Values, + int rowsPerPage) + throws IOException { if (timestamps.length != tag1Values.length || timestamps.length != fieldValues.length || timestamps.length != tag2Values.length) { @@ -123,6 +134,10 @@ public class CompactionTableModelTestFileWriter extends CompactionTestFileWriter chunkWriter .getValueChunkWriterByIndex(2) .write(timestamp, new Binary(tag2Values[row], TSFileConfig.STRING_CHARSET), false); + // Seal all columns together so tests can split equal timestamps across physical pages. + if ((row + 1) % rowsPerPage == 0) { + chunkWriter.sealCurrentPage(); + } } chunkWriter.writeToFileWriter(fileWriter); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/FlatDeviceCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/FlatDeviceCompactionTest.java index 5d60777593c..2ae57a24fe9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/FlatDeviceCompactionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/tablemodel/FlatDeviceCompactionTest.java @@ -28,9 +28,11 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOpt import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; import org.apache.iotdb.db.storageengine.dataregion.compaction.AbstractCompactionTest; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.constant.InnerUnseqCompactionPerformer; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadChunkCompactionPerformer; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadPointCompactionPerformer; +import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CrossSpaceCompactionTask; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.RepairUnsortedFileCompactionTask; import org.apache.iotdb.db.storageengine.dataregion.compaction.repair.RepairDataFileScanUtil; @@ -39,6 +41,7 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileRepairStatus; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.common.conf.TSFileDescriptor; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.StringArrayDeviceID; @@ -62,6 +65,8 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.TreeMap; public class FlatDeviceCompactionTest extends AbstractCompactionTest { @@ -88,6 +93,227 @@ public class FlatDeviceCompactionTest extends AbstractCompactionTest { assertFlatDeviceRowsSurviveCompaction(new ReadPointCompactionPerformer()); } + @Test + public void testInnerUnseqCompactionWithOverlappingFlatDeviceFiles() throws Exception { + assertInnerUnseqFlatDeviceCompaction(3, 20, 8, 37, 11, false); + } + + @Test + public void testFastInnerUnseqCompactionWithOverlappingFlatDeviceFiles() throws Exception { + // INNER_UNSEQ defaults to FAST. Its flat fallback must preserve the unseq classification + // when handing overlapping source files to the read-point performer. + assertInnerUnseqFlatDeviceCompaction( + 3, 20, 8, 37, 11, false, new FastCompactionPerformer(false)); + } + + @Test + public void testFastInnerUnseqCompactionWithFlatDevicePageAndChunkBoundaries() throws Exception { + // Exercise the default performer with equal timestamps split across physical pages/chunks. + assertInnerUnseqFlatDeviceCompaction( + 8, 67, 32, 113, 31, false, new FastCompactionPerformer(false)); + } + + @Test + public void testFastInnerUnseqCompactionWithSparseFlatDeviceRows() throws Exception { + // Unequal client progress leaves gaps in overlapping flushes. Use the INNER_UNSEQ factory + // and task entry point so the test exercises FAST's single-list setter and flat fallback. + assertInnerUnseqFlatDeviceCompaction( + 8, 67, 32, 113, 31, true, InnerUnseqCompactionPerformer.FAST.createInstance()); + } + + @Test + public void testFastInnerUnseqCompactionWithEarlierRowsInNewerFlatDeviceFile() throws Exception { + // File version order differs from time order. The newer file both backfills earlier data + // and overwrites (3, a, x); (3, b, x) and (3, c, x) must remain distinct rows. + long[][] timestamps = {{3, 3, 4, 5}, {1, 3, 3, 6}}; + String[][] firstTags = {{"a", "b", "a", "a"}, {"a", "a", "c", "a"}}; + for (int file = 0; file < timestamps.length; file++) { + TsFileResource source = createEmptyFileAndResource(false); + unseqResources.add(source); + long fieldOffset = (file + 1) * 10L; + try (CompactionTableModelTestFileWriter writer = + new CompactionTableModelTestFileWriter(source)) { + writer.registerFlatDeviceSchema("flat_table"); + writer.startChunkGroup("flat_table", Collections.emptyList()); + writer.generateFlatDeviceAlignedRows( + timestamps[file], + firstTags[file], + new long[] {fieldOffset, fieldOffset + 1, fieldOffset + 2, fieldOffset + 3}, + new String[] {"x", "x", "x", "x"}, + 2); + writer.endChunkGroup(); + writer.endFile(); + } + assertFlatDeviceFileSorted(source); + } + tsFileManager.addAll(unseqResources, false); + InnerSpaceCompactionTask task = + new InnerSpaceCompactionTask( + 0, + tsFileManager, + unseqResources, + false, + InnerUnseqCompactionPerformer.FAST.createInstance(), + 0); + Assert.assertTrue("Flat-device INNER_UNSEQ backfill compaction must succeed", task.start()); + Assert.assertTrue(tsFileManager.getTsFileList(true).isEmpty()); + Assert.assertEquals(1, tsFileManager.getTsFileList(false).size()); + TsFileResource target = tsFileManager.getTsFileList(false).get(0); + Assert.assertFalse(target.isSeq()); + assertFlatDeviceFileSorted(target); + Assert.assertEquals( + Arrays.asList( + "1:a:20:x", "3:a:21:x", "3:b:11:x", "3:c:22:x", "4:a:12:x", "5:a:13:x", "6:a:23:x"), + readFlatDeviceRows(target)); + IDeviceID tableDevice = new StringArrayDeviceID(new String[] {"flat_table"}); + Assert.assertEquals(Collections.singleton(tableDevice), target.getDevices()); + Assert.assertEquals(1L, target.getStartTime(tableDevice).get().longValue()); + Assert.assertEquals(6L, target.getEndTime(tableDevice).get().longValue()); + } + + @Test + public void testFastCrossCompactionPreservesFlatDeviceSourceSpaces() throws Exception { + // The fallback must retain both source spaces so unseq updates win duplicate full row keys + // and different TAGs at the same timestamp survive across both sequence targets. + seqResources.add( + createFlatDeviceFile(true, 10, new String[] {"a", "b"}, new String[] {"x", "y"}, 1)); + seqResources.add( + createFlatDeviceFile(true, 30, new String[] {"a", "b"}, new String[] {"x", "y"}, 3)); + unseqResources.add( + createFlatDeviceFile(false, 20, new String[] {"a", "c"}, new String[] {"x", "z"}, 1)); + unseqResources.add( + createFlatDeviceFile(false, 40, new String[] {"a", "c"}, new String[] {"x", "z"}, 3)); + tsFileManager.addAll(seqResources, true); + tsFileManager.addAll(unseqResources, false); + CrossSpaceCompactionTask task = + new CrossSpaceCompactionTask( + 0, + tsFileManager, + seqResources, + unseqResources, + new FastCompactionPerformer(true), + 0, + 0); + Assert.assertTrue(task.start()); + Assert.assertTrue(tsFileManager.getTsFileList(false).isEmpty()); + Assert.assertEquals(2, tsFileManager.getTsFileList(true).size()); + List<String> rows = new ArrayList<>(); + for (TsFileResource target : tsFileManager.getTsFileList(true)) { + assertFlatDeviceFileSorted(target); + rows.addAll(readFlatDeviceRows(target)); + } + Assert.assertEquals( + Arrays.asList("1:a:20:x", "1:b:11:y", "1:c:21:z", "3:a:40:x", "3:b:31:y", "3:c:41:z"), + rows); + } + + @Test + public void testInnerUnseqCompactionWithFlatDevicePageAndChunkBoundaries() throws Exception { + assertInnerUnseqFlatDeviceCompaction(8, 67, 32, 113, 31, false); + } + + @Test + public void testInnerUnseqCompactionWithSparseFlatDeviceRows() throws Exception { + assertInnerUnseqFlatDeviceCompaction(8, 67, 32, 113, 31, true); + } + + private void assertInnerUnseqFlatDeviceCompaction( + int fileCount, + int timesPerFile, + int deviceCount, + int rowsPerChunk, + int rowsPerPage, + boolean sparse) + throws Exception { + assertInnerUnseqFlatDeviceCompaction( + fileCount, + timesPerFile, + deviceCount, + rowsPerChunk, + rowsPerPage, + sparse, + new ReadPointCompactionPerformer()); + } + + private void assertInnerUnseqFlatDeviceCompaction( + int fileCount, + int timesPerFile, + int deviceCount, + int rowsPerChunk, + int rowsPerPage, + boolean sparse, + ICompactionPerformer performer) + throws Exception { + // Model overlapping unseq flushes from clients at different time offsets. Each source is + // ordered by (time, tag1, tag2), while equal timestamps cross page/chunk and output-block + // boundaries. Duplicate full keys must retain the value from the newer file. + TSFileDescriptor.getInstance().getConfig().setMaxNumberOfPointsInPage(rowsPerPage); + Map<Long, String> expectedRows = new TreeMap<>(); + for (int file = 0; file < fileCount; file++) { + TsFileResource source = createEmptyFileAndResource(false); + unseqResources.add(source); + List<Long> keys = new ArrayList<>(); + for (long time = file * 11L; time < file * 11L + timesPerFile; time++) { + for (int device = 0; device < deviceCount; device++) { + // Deterministic gaps emulate unequal client progress without making a source unsorted. + if (sparse && (time + device * 7L + file * 3L) % 13 < 5) { + continue; + } + keys.add(time * deviceCount + device); + } + } + try (CompactionTableModelTestFileWriter writer = + new CompactionTableModelTestFileWriter(source)) { + writer.registerFlatDeviceSchema("flat_table"); + writer.startChunkGroup("flat_table", Collections.emptyList()); + for (int offset = 0; offset < keys.size(); offset += rowsPerChunk) { + int count = Math.min(rowsPerChunk, keys.size() - offset); + long[] times = new long[count]; + String[] tag1 = new String[count]; + String[] tag2 = new String[count]; + long[] fields = new long[count]; + for (int row = 0; row < count; row++) { + long key = keys.get(offset + row); + int device = (int) (key % deviceCount); + times[row] = key / deviceCount; + // Both TAG columns contribute to ordering; fixed-width values preserve lexical order. + tag1[row] = "device_" + (100 + device / 4); + tag2[row] = "sensor_" + device % 4; + fields[row] = file * 1_000_000L + key; + expectedRows.put( + key, times[row] + ":" + tag1[row] + ":" + fields[row] + ":" + tag2[row]); + } + writer.generateFlatDeviceAlignedRows(times, tag1, fields, tag2, rowsPerPage); + } + writer.endChunkGroup(); + writer.endFile(); + } + assertFlatDeviceFileSorted(source); + } + + tsFileManager.addAll(unseqResources, false); + InnerSpaceCompactionTask task = + new InnerSpaceCompactionTask(0, tsFileManager, unseqResources, false, performer, 0); + Assert.assertTrue("Flat-device INNER_UNSEQ compaction must succeed", task.start()); + Assert.assertTrue(tsFileManager.getTsFileList(true).isEmpty()); + Assert.assertEquals(1, tsFileManager.getTsFileList(false).size()); + TsFileResource target = tsFileManager.getTsFileList(false).get(0); + assertFlatDeviceFileSorted(target); + Assert.assertEquals(new ArrayList<>(expectedRows.values()), readFlatDeviceRows(target)); + IDeviceID tableDevice = new StringArrayDeviceID(new String[] {"flat_table"}); + Assert.assertEquals(Collections.singleton(tableDevice), target.getDevices()); + Assert.assertEquals(0L, target.getStartTime(tableDevice).get().longValue()); + Assert.assertEquals( + (fileCount - 1) * 11L + timesPerFile - 1, target.getEndTime(tableDevice).get().longValue()); + } + + private void assertFlatDeviceFileSorted(TsFileResource resource) throws IOException { + RepairDataFileScanUtil scan = new RepairDataFileScanUtil(resource); + scan.scanTsFile(true); + Assert.assertFalse(resource.getTsFilePath(), scan.isBrokenFile()); + Assert.assertFalse(resource.getTsFilePath(), scan.hasUnsortedDataOrWrongStatistics()); + } + @Test public void testRepairChecksFlatDeviceRowKeys() throws Exception { // TAGs are separated by a FIELD in the physical schema. Compare both TAGs, including
