This is an automated email from the ASF dual-hosted git repository. ColinLeeo pushed a commit to branch fix/tablet-out-of-order-seq-overlap in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f56f2f7ef3f7ef1a20cc65b13d8ddac21ade2abc Author: ColinLee <[email protected]> AuthorDate: Wed Oct 28 11:13:56 2026 +0800 Fix sequence space overlap caused by out-of-order tablets DataRegion.split() classified the rows of a tablet by latching the sequence flag: once a row was newer than the device's last flush time, every following row of that device was written into the sequence space as well. That only holds while the rows are sorted by time, an expectation the engine does not verify - its own INSERT INTO ... SELECT path feeds the storage layer with rows in the child operator's output order. With an out-of-order tablet such as [100, 15] for a device whose last flushed time is 39, the older row was written into a new sequence TsFile, which then overlapped the existing sequence TsFile of the same device. Subsequent queries returned duplicated rows and broke the time order, without any error. Track whether the rows are still in ascending time order and, when the order is broken, classify each remaining row against the device's last flush time individually. Sorted tablets keep the previous behaviour and cost only one extra comparison per row. Fixes #19020 --- .../db/storageengine/dataregion/DataRegion.java | 25 ++++++--- .../storageengine/dataregion/DataRegionTest.java | 63 ++++++++++++++++++++++ 2 files changed, 81 insertions(+), 7 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index ead75db6bac..e3bf400e0ea 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -1267,9 +1267,20 @@ public class DataRegion implements IDataRegionForQuery { // if is sequence boolean isSequence = false; + // Rows of a device are expected to arrive in ascending time order. Only then does "all rows + // after the first sequence row are sequence rows" hold. If the order is broken, an older row + // placed after a newer one must still be written into the unsequence space: classifying it as + // sequence would write data older than the device's last flushed time into a new sequence + // file, which then overlaps the existing sequence files. + boolean timeOrdered = true; + long previousTime = Long.MIN_VALUE; while (loc < endOffset) { long time = insertTabletNode.getTimes()[loc]; final long timePartitionId = TimePartitionUtils.getTimePartitionId(time); + if (time < previousTime) { + timeOrdered = false; + } + previousTime = time; long lastFlushTime; // judge if we should insert sequence @@ -1280,17 +1291,17 @@ public class DataRegion implements IDataRegionForQuery { before = loc; beforeTimePartition = timePartitionId; isSequence = time > lastFlushTime; - } else if (!isSequence) { + } else if (!isSequence || !timeOrdered) { lastFlushTime = getLastFlushTime(timePartitionId, insertTabletNode.getDeviceID(loc)); - if (time > lastFlushTime) { - // the same partition and switch to sequence data - // insert previous range into unsequence - updateSplitInfo(splitInfo, beforeTimePartition, false, new int[] {before, loc}); + boolean rowIsSequence = time > lastFlushTime; + if (rowIsSequence != isSequence) { + // the same partition and the sequence flag of this row changes + updateSplitInfo(splitInfo, beforeTimePartition, isSequence, new int[] {before, loc}); before = loc; - isSequence = true; + isSequence = rowIsSequence; } } - // else: the same partition and isSequence not changed, just move the cursor forward + // else: the same partition, the rows are in time order and isSequence not changed loc++; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java index 4e2ca24a750..cbd5aa60963 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java @@ -61,6 +61,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.CompactionUtils; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.utils.log.CompactionLogger; +import org.apache.iotdb.db.storageengine.dataregion.compaction.repair.RepairDataFileScanUtil; import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionTaskManager; import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.constant.InnerSequenceCompactionSelector; import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.constant.InnerUnsequenceCompactionSelector; @@ -91,6 +92,7 @@ import org.apache.tsfile.file.metadata.enums.TSEncoding; import org.apache.tsfile.read.TimeValuePair; import org.apache.tsfile.read.common.TimeRange; import org.apache.tsfile.read.reader.IPointReader; +import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.write.record.TSRecord; import org.apache.tsfile.write.record.datapoint.DataPoint; @@ -107,6 +109,7 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; import java.lang.reflect.Field; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -2305,4 +2308,64 @@ public class DataRegionTest { } } } + + /** + * The rows of a tablet are expected to be sorted by the client. When they are not, an older row + * that follows a newer one must still be written into the unsequence space: classifying it as + * sequence would write data older than the device's last flushed time into a new sequence file, + * which then overlaps the existing sequence file of the same device. + */ + @Test + public void testOutOfOrderTabletKeepsSequenceSpaceNonOverlapped() + throws QueryProcessException, WriteProcessException { + long[] orderedTimes = new long[30]; + for (int i = 0; i < orderedTimes.length; i++) { + orderedTimes[i] = 10 + i; + } + dataRegion.insertTablet(genTabletWithTimes(orderedTimes)); + dataRegion.syncCloseAllWorkingTsFileProcessors(); + + // the first row is newer than the device's last flush time, the second one is older + dataRegion.insertTablet(genTabletWithTimes(100L, 15L)); + dataRegion.syncCloseAllWorkingTsFileProcessors(); + + List<TsFileResource> seqFiles = dataRegion.getTsFileManager().getTsFileList(true); + List<TsFileResource> unseqFiles = dataRegion.getTsFileManager().getTsFileList(false); + List<TsFileResource> overlappedSeqFiles = + RepairDataFileScanUtil.checkTimePartitionHasOverlap(seqFiles, false); + Assert.assertTrue( + "sequence files should not overlap, but got " + overlappedSeqFiles, + overlappedSeqFiles.isEmpty()); + // the older row must be written into the unsequence space + Assert.assertEquals(1, unseqFiles.size()); + Assert.assertEquals(2, seqFiles.size()); + } + + private static RelationalInsertTabletNode genTabletWithTimes(long... times) { + return new RelationalInsertTabletNode( + new PlanNodeId("tablet-" + times[0] + "-" + times.length), + new PartialPath(new String[] {StatementTestUtils.tableName()}), + true, + StatementTestUtils.genColumnNames(), + StatementTestUtils.genDataTypes(), + StatementTestUtils.genMeasurementSchemas(), + times, + null, + genSingleDeviceColumns(times.length), + times.length, + StatementTestUtils.genColumnCategories()); + } + + /** All rows belong to the same device, so that their times are classified together. */ + private static Object[] genSingleDeviceColumns(int rowCnt) { + Binary[] ids = new Binary[rowCnt]; + Binary[] attributes = new Binary[rowCnt]; + double[] values = new double[rowCnt]; + for (int i = 0; i < rowCnt; i++) { + ids[i] = new Binary("id:0".getBytes(StandardCharsets.UTF_8)); + attributes[i] = new Binary("attr:0".getBytes(StandardCharsets.UTF_8)); + values[i] = i; + } + return new Object[] {ids, attributes, values}; + } }
