diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java
index b0cccba6f6cb9..10e476705b5f4 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java
@@ -87,7 +87,11 @@ public static void setUpClass() {
         .getConfig()
         .getCommonConfig()
         .setPartitionInterval(1000)
-        .setMemtableSizeThreshold(10000);
+        .setMemtableSizeThreshold(10000)
+        // The out-of-order tablet test needs a sequence file with several pages: when the earlier
+        // file only has one page, the read side merges the overlapping page before the earlier
+        // pages are emitted and the wrong order is masked.
+        .setMaxNumberOfPointsInPage(10);
     EnvFactory.getEnv().initClusterEnvironment();
   }
 
@@ -167,6 +171,83 @@ public void testSimple() throws SQLException {
     }
   }
 
+  /**
+   * End-to-end check for an {@code INSERT INTO ... SELECT} whose source tablet is not in ascending
+   * time order. The destination device is flushed at time 39, then the query feeds it another
+   * tablet: on the unfixed code the row at time 15 is classified as sequence data, a sequence
+   * TsFile overlapping [10,39] is created, and the query returns 15 after 39 (or twice).
+   *
+   * <p>The out-of-order input depends on the nested-loop cross join emitting the rows in probe
+   * order, i.e. {@code [100, 15]}. If the optimizer ever emits those rows in time order this test
+   * no longer reproduces the situation and would pass trivially; the guards that do not depend on
+   * the plan shape are {@code InsertTabletStatementGeneratorTest} (the tablet is sorted before it
+   * leaves the engine) and {@code
+   * DataRegionTest#testOutOfOrderTabletKeepsSequenceSpaceNonOverlapped} (the storage-layer
+   * classification of an unsorted tablet).
+   *
+   * <p>The first sequence file also has to hold several pages (see {@code
+   * max_number_of_points_in_page} in {@link #setUpClass()}): with a single page the overlapping
+   * page is merged before the earlier pages are emitted, which hides the wrong order as well.
+   */
+  @Test
+  public void testInsertIntoQueryWithOutOfOrderTabletDoesNotOverlapSequenceFiles()
+      throws SQLException {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute("CREATE TABLE overlap_dst(id STRING TAG, s INT32 FIELD)");
+      statement.execute("CREATE TABLE overlap_src(id STRING TAG, s INT32 FIELD)");
+
+      for (int time = 10; time <= 39; time++) {
+        statement.execute(
+            String.format("INSERT INTO overlap_dst(time,id,s) VALUES(%d,'x',%d)", time, time));
+      }
+      statement.execute("FLUSH");
+
+      statement.execute("INSERT INTO overlap_src(time,id,s) VALUES(100,'x',1),(15,'x',2)");
+      statement.execute(
+          "INSERT INTO overlap_dst (SELECT b.time, b.id, b.s FROM overlap_src a, overlap_src b "
+              + "WHERE a.id='x' AND b.id='x' AND a.time<>b.time)");
+      statement.execute("FLUSH");
+
+      List<Long> expectedTimes = new ArrayList<>();
+      for (long time = 10; time <= 39; time++) {
+        expectedTimes.add(time);
+      }
+      expectedTimes.add(100L);
+
+      try (ResultSet resultSet =
+          statement.executeQuery("SELECT time,s FROM overlap_dst WHERE id='x'")) {
+        List<Long> actualTimes = new ArrayList<>();
+        while (resultSet.next()) {
+          long time = resultSet.getLong(1);
+          actualTimes.add(time);
+          if (time == 15) {
+            Assert.assertEquals(2, resultSet.getInt(2));
+          }
+        }
+        Assert.assertEquals(expectedTimes, actualTimes);
+      }
+
+      // descending: the latest row 100 first, then 39 down to 10
+      List<Long> expectedDescTimes = new ArrayList<>();
+      expectedDescTimes.add(100L);
+      for (long time = 39; time >= 10; time--) {
+        expectedDescTimes.add(time);
+      }
+
+      try (ResultSet resultSet =
+          statement.executeQuery(
+              "SELECT time,s FROM overlap_dst WHERE id='x' ORDER BY time DESC")) {
+        List<Long> actualTimes = new ArrayList<>();
+        while (resultSet.next()) {
+          actualTimes.add(resultSet.getLong(1));
+        }
+        Assert.assertEquals(expectedDescTimes, actualTimes);
+      }
+    }
+  }
+
   @Test
   public void testInsertIntoSourceTable() throws SQLException {
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/InsertTabletStatementGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/InsertTabletStatementGenerator.java
index d32f8bca7d5ce..cd7ac65b73374 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/InsertTabletStatementGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/InsertTabletStatementGenerator.java
@@ -33,7 +33,9 @@
 import org.apache.tsfile.utils.BitMap;
 import org.apache.tsfile.utils.RamUsageEstimator;
 
+import java.lang.reflect.Array;
 import java.util.Arrays;
+import java.util.Comparator;
 
 public abstract class InsertTabletStatementGenerator implements Accountable {
   protected final int rowLimit;
@@ -112,6 +114,8 @@ public InsertTabletStatement constructInsertTabletStatement() {
       }
     }
 
+    sortRowsByTimeIfNeeded();
+
     insertTabletStatement.setTimes(times);
     insertTabletStatement.setBitMaps(bitMaps);
     insertTabletStatement.setColumns(columns);
@@ -157,6 +161,64 @@ protected void processColumn(
         .write(sourceTypeConvertor, valueColumn, rowIndex, columns, rowCount);
   }
 
+  /**
+   * The storage layer assumes the rows of a tablet are sorted by time: the sequence/unsequence
+   * classification, the time boundaries of the tsfile resource, the ttl check, the time partition
+   * routing and the last value composition all rely on it. The child operator of an {@code INSERT
+   * INTO ... SELECT} does not guarantee that order, so restore it before the tablet leaves the
+   * engine.
+   */
+  private void sortRowsByTimeIfNeeded() {
+    boolean sorted = true;
+    for (int i = 1; i < rowCount; i++) {
+      if (times[i] < times[i - 1]) {
+        sorted = false;
+        break;
+      }
+    }
+    if (sorted) {
+      return;
+    }
+
+    Integer[] order = new Integer[rowCount];
+    for (int i = 0; i < rowCount; i++) {
+      order[i] = i;
+    }
+    // stable, so rows sharing a timestamp keep their relative order
+    Arrays.sort(order, Comparator.comparingLong(index -> times[index]));
+
+    int[] permutation = new int[rowCount];
+    long[] sortedTimes = new long[rowCount];
+    for (int i = 0; i < rowCount; i++) {
+      permutation[i] = order[i];
+      sortedTimes[i] = times[order[i]];
+    }
+    times = sortedTimes;
+    for (int i = 0; i < columns.length; i++) {
+      columns[i] = permuteColumn(columns[i], permutation);
+      bitMaps[i] = permuteBitMap(bitMaps[i], permutation);
+    }
+  }
+
+  private Object permuteColumn(Object column, int[] permutation) {
+    int length = Array.getLength(column);
+    Object sortedColumn = Array.newInstance(column.getClass().getComponentType(), length);
+    for (int i = 0; i < length; i++) {
+      Array.set(sortedColumn, i, Array.get(column, permutation[i]));
+    }
+    return sortedColumn;
+  }
+
+  private BitMap permuteBitMap(BitMap bitMap, int[] permutation) {
+    BitMap sortedBitMap = new BitMap(rowCount);
+    for (int i = 0; i < rowCount; i++) {
+      if (bitMap.isMarked(permutation[i])) {
+        sortedBitMap.mark(i);
+      }
+    }
+    return sortedBitMap;
+  }
+
   protected long sizeOf(Object[] arr, Class<?> clazz) {
     if (arr == null) {
       return 0;
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 52000fe97fce3..e7e2733c9c3f7 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
@@ -1355,9 +1355,20 @@ private void split(
 
     // 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
@@ -1368,17 +1379,17 @@ private void split(
         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/queryengine/execution/operator/process/InsertTabletStatementGeneratorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/InsertTabletStatementGeneratorTest.java
new file mode 100644
index 0000000000000..663af66d12580
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/InsertTabletStatementGeneratorTest.java
@@ -0,0 +1,314 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.execution.operator.process;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.parameter.InputLocation;
+import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.utils.Binary;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+public class InsertTabletStatementGeneratorTest {
+
+  @Test
+  public void testOutOfOrderRowsAreSortedByTime() throws IllegalPathException {
+    TreeInsertTabletStatementGenerator generator = createGenerator();
+
+    TsBlockBuilder builder = new TsBlockBuilder(Collections.singletonList(TSDataType.INT32));
+    append(builder, 100, 100);
+    append(builder, 15, 15);
+    generator.processTsBlock(builder.build(), 0);
+
+    InsertTabletStatement statement = generator.constructInsertTabletStatement();
+
+    Assert.assertArrayEquals(new long[] {15, 100}, statement.getTimes());
+    Assert.assertArrayEquals(new int[] {15, 100}, (int[]) statement.getColumns()[0]);
+  }
+
+  @Test
+  public void testNullValueMovesWithItsRow() throws IllegalPathException {
+    TreeInsertTabletStatementGenerator generator = createGenerator();
+
+    TsBlockBuilder builder = new TsBlockBuilder(Collections.singletonList(TSDataType.INT32));
+    append(builder, 100, 100);
+    builder.getTimeColumnBuilder().writeLong(15);
+    builder.getColumnBuilder(0).appendNull();
+    builder.declarePosition();
+    generator.processTsBlock(builder.build(), 0);
+
+    InsertTabletStatement statement = generator.constructInsertTabletStatement();
+
+    Assert.assertArrayEquals(new long[] {15, 100}, statement.getTimes());
+    Assert.assertArrayEquals(new int[] {0, 100}, (int[]) statement.getColumns()[0]);
+    // row 0 (time 15) has no value, row 1 (time 100) does
+    Assert.assertTrue(statement.getBitMaps()[0].isMarked(0));
+    Assert.assertFalse(statement.getBitMaps()[0].isMarked(1));
+  }
+
+  @Test
+  public void testSortedRowsAreKeptAsIs() throws IllegalPathException {
+    TreeInsertTabletStatementGenerator generator = createGenerator();
+
+    TsBlockBuilder builder = new TsBlockBuilder(Collections.singletonList(TSDataType.INT32));
+    append(builder, 15, 15);
+    append(builder, 100, 100);
+    generator.processTsBlock(builder.build(), 0);
+
+    InsertTabletStatement statement = generator.constructInsertTabletStatement();
+
+    Assert.assertArrayEquals(new long[] {15, 100}, statement.getTimes());
+    Assert.assertArrayEquals(new int[] {15, 100}, (int[]) statement.getColumns()[0]);
+  }
+
+  /**
+   * A full batch ({@code rowCount == rowLimit}) takes the other branch of {@code
+   * constructInsertTabletStatement}: the generator's own arrays are handed to the statement without
+   * being copied first, so sorting has to keep the arrays and the statement consistent.
+   */
+  @Test
+  public void testFullBatchOfMixedTypesIsSorted() throws IllegalPathException {
+    String[] measurements = {"s1", "s2", "s3", "s4", "s5", "s6"};
+    TSDataType[] dataTypes = {
+      TSDataType.INT32,
+      TSDataType.INT64,
+      TSDataType.FLOAT,
+      TSDataType.DOUBLE,
+      TSDataType.BOOLEAN,
+      TSDataType.TEXT
+    };
+    Map<String, InputLocation> inputLocations = new LinkedHashMap<>();
+    for (int i = 0; i < measurements.length; i++) {
+      inputLocations.put(measurements[i], new InputLocation(0, i));
+    }
+    Map<String, TSDataType> dataTypeMap = new LinkedHashMap<>();
+    for (int i = 0; i < measurements.length; i++) {
+      dataTypeMap.put(measurements[i], dataTypes[i]);
+    }
+    TreeInsertTabletStatementGenerator generator =
+        new TreeInsertTabletStatementGenerator(
+            new PartialPath("root.sg.d1"),
+            inputLocations,
+            dataTypeMap,
+            Arrays.asList(dataTypes),
+            false,
+            3);
+
+    TsBlockBuilder builder = new TsBlockBuilder(Arrays.asList(dataTypes));
+    // row order: 30, 10, 20; the TEXT value of the row at time 10 is null
+    appendMixedRow(builder, 30, false);
+    appendMixedRow(builder, 10, true);
+    appendMixedRow(builder, 20, false);
+    generator.processTsBlock(builder.build(), 0);
+
+    InsertTabletStatement statement = generator.constructInsertTabletStatement();
+
+    Assert.assertEquals(3, statement.getRowCount());
+    Assert.assertArrayEquals(new long[] {10, 20, 30}, statement.getTimes());
+    Assert.assertArrayEquals(new int[] {10, 20, 30}, (int[]) statement.getColumns()[0]);
+    Assert.assertArrayEquals(new long[] {10L, 20L, 30L}, (long[]) statement.getColumns()[1]);
+    Assert.assertArrayEquals(
+        new float[] {10.0F, 20.0F, 30.0F}, (float[]) statement.getColumns()[2], 0.0F);
+    Assert.assertArrayEquals(
+        new double[] {10.0, 20.0, 30.0}, (double[]) statement.getColumns()[3], 0.0);
+    Assert.assertArrayEquals(
+        new boolean[] {false, true, false}, (boolean[]) statement.getColumns()[4]);
+    Binary[] textColumn = (Binary[]) statement.getColumns()[5];
+    Assert.assertTrue(statement.getBitMaps()[5].isMarked(0));
+    Assert.assertFalse(statement.getBitMaps()[5].isMarked(1));
+    Assert.assertFalse(statement.getBitMaps()[5].isMarked(2));
+    Assert.assertEquals("v20", textColumn[1].toString());
+    Assert.assertEquals("v30", textColumn[2].toString());
+  }
+
+  private void appendMixedRow(TsBlockBuilder builder, long time, boolean textIsNull) {
+    builder.getTimeColumnBuilder().writeLong(time);
+    builder.getColumnBuilder(0).writeInt((int) time);
+    builder.getColumnBuilder(1).writeLong(time);
+    builder.getColumnBuilder(2).writeFloat(time);
+    builder.getColumnBuilder(3).writeDouble(time);
+    builder.getColumnBuilder(4).writeBoolean(time % 20 != 10);
+    if (textIsNull) {
+      builder.getColumnBuilder(5).appendNull();
+    } else {
+      builder
+          .getColumnBuilder(5)
+          .writeBinary(new Binary(("v" + time).getBytes(StandardCharsets.UTF_8)));
+    }
+    builder.declarePosition();
+  }
+
+  /**
+   * The table-model generator writes the tag value of every row (not only of the first one), so a
+   * tag, including a null tag carried by the bitmap, has to move together with the row it belongs
+   * to.
+   */
+  @Test
+  public void testTagColumnMovesWithItsRow() throws IllegalPathException {
+    Map<String, InputLocation> inputLocations = new LinkedHashMap<>();
+    inputLocations.put("time", new InputLocation(0, 0));
+    inputLocations.put("id1", new InputLocation(0, 1));
+    inputLocations.put("m1", new InputLocation(0, 2));
+    Map<String, TSDataType> dataTypeMap = new LinkedHashMap<>();
+    dataTypeMap.put("id1", TSDataType.STRING);
+    dataTypeMap.put("m1", TSDataType.DOUBLE);
+    TableInsertTabletStatementGenerator generator =
+        new TableInsertTabletStatementGenerator(
+            "root.sg",
+            new PartialPath(new String[] {"table1"}),
+            inputLocations,
+            dataTypeMap,
+            Arrays.asList(TSDataType.STRING, TSDataType.DOUBLE),
+            Arrays.asList(TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD),
+            true,
+            10,
+            "time");
+
+    TsBlockBuilder builder =
+        new TsBlockBuilder(Arrays.asList(TSDataType.INT64, TSDataType.STRING, TSDataType.DOUBLE));
+    appendTableRow(builder, 100, "b", 1.0);
+    appendTableRow(builder, 15, "a", 2.0);
+    appendTableRow(builder, 50, null, 3.0);
+    generator.processTsBlock(builder.build(), 0);
+
+    InsertTabletStatement statement = generator.constructInsertTabletStatement();
+
+    Assert.assertEquals(2, statement.getColumns().length);
+    Assert.assertArrayEquals(new long[] {15, 50, 100}, statement.getTimes());
+    Binary[] tagColumn = (Binary[]) statement.getColumns()[0];
+    Assert.assertEquals("a", tagColumn[0].toString());
+    Assert.assertTrue(statement.getBitMaps()[0].isMarked(1));
+    Assert.assertEquals("b", tagColumn[2].toString());
+    Assert.assertArrayEquals(
+        new double[] {2.0, 3.0, 1.0}, (double[]) statement.getColumns()[1], 0.0);
+  }
+
+  private void appendTableRow(TsBlockBuilder builder, long time, String tag, double value) {
+    // the time column of the target table is carried as value column 0 of the input TsBlock (the
+    // generator reads it through getValueColumns()[timeColumnIndex]), the block itself also needs
+    // its own time column
+    builder.getTimeColumnBuilder().writeLong(time);
+    builder.getColumnBuilder(0).writeLong(time);
+    if (tag == null) {
+      builder.getColumnBuilder(1).appendNull();
+    } else {
+      builder.getColumnBuilder(1).writeBinary(new Binary(tag.getBytes(StandardCharsets.UTF_8)));
+    }
+    builder.getColumnBuilder(2).writeDouble(value);
+    builder.declarePosition();
+  }
+
+  /**
+   * The generator is reused through {@link InsertTabletStatementGenerator#reset()} after a write.
+   */
+  @Test
+  public void testGeneratorIsReusableAfterReset() throws IllegalPathException {
+    TreeInsertTabletStatementGenerator generator = createGenerator();
+
+    TsBlockBuilder first = new TsBlockBuilder(Collections.singletonList(TSDataType.INT32));
+    append(first, 100, 100);
+    append(first, 15, 15);
+    generator.processTsBlock(first.build(), 0);
+    InsertTabletStatement firstStatement = generator.constructInsertTabletStatement();
+    Assert.assertArrayEquals(new long[] {15, 100}, firstStatement.getTimes());
+
+    generator.reset();
+    TsBlockBuilder second = new TsBlockBuilder(Collections.singletonList(TSDataType.INT32));
+    append(second, 200, 200);
+    append(second, 150, 150);
+    generator.processTsBlock(second.build(), 0);
+    InsertTabletStatement secondStatement = generator.constructInsertTabletStatement();
+
+    // the first statement is not affected by the reset, and the second batch is sorted as well
+    Assert.assertArrayEquals(new long[] {15, 100}, firstStatement.getTimes());
+    Assert.assertArrayEquals(new long[] {150, 200}, secondStatement.getTimes());
+    Assert.assertArrayEquals(new int[] {150, 200}, (int[]) secondStatement.getColumns()[0]);
+  }
+
+  /**
+   * Rows sharing a timestamp keep their relative order, so the last one still wins when the storage
+   * layer resolves the duplicate.
+   */
+  @Test
+  public void testDuplicateTimestampsKeepTheirRelativeOrder() throws IllegalPathException {
+    TreeInsertTabletStatementGenerator generator = createGenerator();
+
+    TsBlockBuilder builder = new TsBlockBuilder(Collections.singletonList(TSDataType.INT32));
+    append(builder, 100, 1);
+    append(builder, 15, 2);
+    append(builder, 15, 3);
+    generator.processTsBlock(builder.build(), 0);
+
+    InsertTabletStatement statement = generator.constructInsertTabletStatement();
+
+    Assert.assertArrayEquals(new long[] {15, 15, 100}, statement.getTimes());
+    Assert.assertArrayEquals(new int[] {2, 3, 1}, (int[]) statement.getColumns()[0]);
+  }
+
+  @Test
+  public void testAlignedTabletRowsAreSorted() throws IllegalPathException {
+    TreeInsertTabletStatementGenerator generator =
+        new TreeInsertTabletStatementGenerator(
+            new PartialPath("root.sg.d1"),
+            Collections.singletonMap("s1", new InputLocation(0, 0)),
+            Collections.singletonMap("s1", TSDataType.INT32),
+            Collections.singletonList(TSDataType.INT32),
+            true,
+            10);
+
+    TsBlockBuilder builder = new TsBlockBuilder(Collections.singletonList(TSDataType.INT32));
+    append(builder, 100, 100);
+    append(builder, 15, 15);
+    generator.processTsBlock(builder.build(), 0);
+
+    InsertTabletStatement statement = generator.constructInsertTabletStatement();
+
+    Assert.assertTrue(statement.isAligned());
+    Assert.assertArrayEquals(new long[] {15, 100}, statement.getTimes());
+    Assert.assertArrayEquals(new int[] {15, 100}, (int[]) statement.getColumns()[0]);
+  }
+
+  private TreeInsertTabletStatementGenerator createGenerator() throws IllegalPathException {
+    return new TreeInsertTabletStatementGenerator(
+        new PartialPath("root.sg.d1"),
+        Collections.singletonMap("s1", new InputLocation(0, 0)),
+        Collections.singletonMap("s1", TSDataType.INT32),
+        Collections.singletonList(TSDataType.INT32),
+        false,
+        10);
+  }
+
+  private void append(TsBlockBuilder builder, long time, int value) {
+    builder.getTimeColumnBuilder().writeLong(time);
+    builder.getColumnBuilder(0).writeInt(value);
+    builder.declarePosition();
+  }
+}
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 c82a65d7389fa..9e1bf9d020676 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.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;
@@ -93,8 +94,10 @@
 
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.TimeseriesMetadata;
 import org.apache.tsfile.file.metadata.enums.CompressionType;
 import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.file.metadata.statistics.Statistics;
 import org.apache.tsfile.read.TimeValuePair;
 import org.apache.tsfile.read.TsFileReader;
 import org.apache.tsfile.read.TsFileSequenceReader;
@@ -104,6 +107,7 @@
 import org.apache.tsfile.read.expression.QueryExpression;
 import org.apache.tsfile.read.query.dataset.QueryDataSet;
 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;
@@ -119,7 +123,9 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.io.Serializable;
 import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -2762,4 +2768,174 @@ public void testSyncCloseWaitsForAlreadyClosingProcessor() throws Exception {
       }
     }
   }
+
+  /**
+   * 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.
+   *
+   * <p>The written rows are asserted through the chunk statistics of each file, not only through
+   * the device time index: the index is updated from the first and last row of every inserted range
+   * ({@code TsFileProcessor#insertTablet}), so a range that physically contains an older row can
+   * still be indexed as if it did not, and a resource-index based overlap check would miss it.
+   */
+  @Test
+  public void testOutOfOrderTabletKeepsSequenceSpaceNonOverlapped()
+      throws QueryProcessException, WriteProcessException, IOException {
+    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 written rows themselves: the sequence space keeps 10..39 in the first file and only 100
+    // in the second one, while the older row 15 landed in the unsequence file
+    List<long[]> seqStatistics = new ArrayList<>();
+    for (TsFileResource seqFile : seqFiles) {
+      seqStatistics.add(readFieldStatistics(seqFile));
+    }
+    Assert.assertTrue(
+        "expected a sequence file with 10..39 and one with 100 only, but got "
+            + formatStatistics(seqStatistics),
+        seqStatistics.stream().anyMatch(statistics -> statisticsIs(statistics, 10, 39, 30))
+            && seqStatistics.stream()
+                .anyMatch(statistics -> statisticsIs(statistics, 100, 100, 1)));
+    // the older row must be written into the unsequence space
+    Assert.assertEquals(1, unseqFiles.size());
+    Assert.assertTrue(
+        "the unsequence file should hold only time 15, but got "
+            + formatStatistics(Collections.singletonList(readFieldStatistics(unseqFiles.get(0)))),
+        statisticsIs(readFieldStatistics(unseqFiles.get(0)), 15, 15, 1));
+    Assert.assertEquals(2, seqFiles.size());
+  }
+
+  /**
+   * Once a descending step is seen, every following row is classified on its own, so a row that is
+   * newer than the device's last flush time after the inversion goes back to the sequence space
+   * without creating an overlapping sequence file.
+   */
+  @Test
+  public void testOutOfOrderTabletWithMultipleInversionsKeepsSequenceSpaceNonOverlapped()
+      throws QueryProcessException, WriteProcessException, IOException {
+    long[] orderedTimes = new long[30];
+    for (int i = 0; i < orderedTimes.length; i++) {
+      orderedTimes[i] = 10 + i;
+    }
+    dataRegion.insertTablet(genTabletWithTimes(orderedTimes));
+    dataRegion.syncCloseAllWorkingTsFileProcessors();
+
+    // sequence, then unsequence, then sequence again
+    dataRegion.insertTablet(genTabletWithTimes(100L, 15L, 200L));
+    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());
+    List<long[]> seqStatistics = new ArrayList<>();
+    for (TsFileResource seqFile : seqFiles) {
+      seqStatistics.add(readFieldStatistics(seqFile));
+    }
+    Assert.assertTrue(
+        "expected a sequence file with 10..39 and one with 100..200, but got "
+            + formatStatistics(seqStatistics),
+        seqStatistics.stream().anyMatch(statistics -> statisticsIs(statistics, 10, 39, 30))
+            && seqStatistics.stream()
+                .anyMatch(statistics -> statisticsIs(statistics, 100, 200, 2)));
+    // only the inverted row goes to the unsequence space
+    Assert.assertEquals(1, unseqFiles.size());
+    Assert.assertTrue(
+        "the unsequence file should hold only time 15, but got "
+            + formatStatistics(Collections.singletonList(readFieldStatistics(unseqFiles.get(0)))),
+        statisticsIs(readFieldStatistics(unseqFiles.get(0)), 15, 15, 1));
+  }
+
+  private static boolean statisticsIs(long[] statistics, long startTime, long endTime, long count) {
+    return statistics[0] == startTime && statistics[1] == endTime && statistics[2] == count;
+  }
+
+  private static String formatStatistics(List<long[]> statisticsList) {
+    StringBuilder builder = new StringBuilder("[");
+    for (int i = 0; i < statisticsList.size(); i++) {
+      if (i > 0) {
+        builder.append(", ");
+      }
+      builder
+          .append("startTime=")
+          .append(statisticsList.get(i)[0])
+          .append(", endTime=")
+          .append(statisticsList.get(i)[1])
+          .append(", count=")
+          .append(statisticsList.get(i)[2]);
+    }
+    return builder.append("]").toString();
+  }
+
+  /**
+   * Returns {startTime, endTime, pointCount} of measurement {@code m1} as recorded by the chunk
+   * statistics of the file, i.e. derived from the data written into it instead of from the device
+   * time index.
+   */
+  private static long[] readFieldStatistics(TsFileResource resource) throws IOException {
+    long startTime = Long.MAX_VALUE;
+    long endTime = Long.MIN_VALUE;
+    long count = 0;
+    try (TsFileSequenceReader reader = new TsFileSequenceReader(resource.getTsFilePath())) {
+      for (IDeviceID device : resource.getDevices()) {
+        TimeseriesMetadata metadata = reader.readDeviceMetadata(device).get("m1");
+        if (metadata == null || metadata.getStatistics() == null) {
+          continue;
+        }
+        Statistics<? extends Serializable> statistics = metadata.getStatistics();
+        startTime = Math.min(startTime, statistics.getStartTime());
+        endTime = Math.max(endTime, statistics.getEndTime());
+        count += statistics.getCount();
+      }
+    }
+    return new long[] {startTime, endTime, count};
+  }
+
+  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};
+  }
 }