Fix sequence space overlap caused by out-of-order tablets

integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBInsertQueryIT.java

@@ -167,6 +167,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);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java

@@ -26,8 +26,8 @@
 import org.apache.iotdb.db.pipe.sink.util.builder.PipeTableModelTsFileBuilderV2;
 import org.apache.iotdb.db.pipe.sink.util.builder.PipeTreeModelTsFileBuilderV2;
 import org.apache.iotdb.db.pipe.sink.util.builder.PipeTsFileBuilder;
-import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTableModelTabletEventSorter;
-import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTreeModelTabletEventSorter;
+import org.apache.iotdb.db.utils.sorter.TableModelTabletEventSorter;
+import org.apache.iotdb.db.utils.sorter.TreeModelTabletEventSorter;
 import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent;
 
 import org.apache.tsfile.exception.write.WriteProcessException;
@@ -274,7 +274,7 @@ private void bufferTreeModelTablet(
       final long creationTime,
       final Tablet tablet,
       final boolean isAligned) {
-    new PipeTreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
+    new TreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
 
     pipeName2WeightMap.compute(
         new Pair<>(pipeName, creationTime),
@@ -285,7 +285,7 @@ private void bufferTreeModelTablet(
 
   private void bufferTableModelTablet(
       final String pipeName, final long creationTime, final Tablet tablet, final String dataBase) {
-    new PipeTableModelTabletEventSorter(tablet).sortAndDeduplicateByDevIdTimestamp();
+    new TableModelTabletEventSorter(tablet).sortAndDeduplicateByDevIdTimestamp();
 
     pipeName2WeightMap.compute(
         new Pair<>(pipeName, creationTime),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReq.java

@@ -26,8 +26,8 @@
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils.TabletStringInternPool;
 import org.apache.iotdb.db.pipe.sink.util.TabletStatementConverter;
-import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTreeModelTabletEventSorter;
 import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
+import org.apache.iotdb.db.utils.sorter.TreeModelTabletEventSorter;
 import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
 
 import org.apache.tsfile.enums.TSDataType;
@@ -81,12 +81,12 @@ public boolean getIsAligned() {
    */
   public InsertTabletStatement constructStatement() {
     if (statement != null) {
-      new PipeTreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary();
+      new TreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary();
       return statement;
     }
 
     // Sort and deduplicate tablet before converting
-    new PipeTreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
+    new TreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
 
     try {
       if (isTabletEmpty(tablet)) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReqV2.java

@@ -27,9 +27,9 @@
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils;
 import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils.TabletStringInternPool;
 import org.apache.iotdb.db.pipe.sink.util.TabletStatementConverter;
-import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTableModelTabletEventSorter;
-import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTreeModelTabletEventSorter;
 import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
+import org.apache.iotdb.db.utils.sorter.TableModelTabletEventSorter;
+import org.apache.iotdb.db.utils.sorter.TreeModelTabletEventSorter;
 import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
 
 import org.apache.tsfile.utils.PublicBAOS;
@@ -62,10 +62,10 @@ public InsertTabletStatement constructStatement() {
 
     if (statement != null) {
       if (isTableModel) {
-        new PipeTableModelTabletEventSorter(statement).sortByTimestampIfNecessary();
+        new TableModelTabletEventSorter(statement).sortByTimestampIfNecessary();
         statement.setWriteToTable(true);
       } else {
-        new PipeTreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary();
+        new TreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary();
       }
       if (Objects.nonNull(dataBaseName)) {
         statement.setDatabaseName(dataBaseName);
@@ -75,9 +75,9 @@ public InsertTabletStatement constructStatement() {
     }
 
     if (isTableModel) {
-      new PipeTableModelTabletEventSorter(tablet).sortByTimestampIfNecessary();
+      new TableModelTabletEventSorter(tablet).sortByTimestampIfNecessary();
     } else {
-      new PipeTreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
+      new TreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
     }
 
     try {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaServerHandle.java

@@ -20,8 +20,8 @@
 package org.apache.iotdb.db.pipe.sink.protocol.opcda;
 
 import org.apache.iotdb.db.i18n.DataNodePipeMessages;
-import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTreeModelTabletEventSorter;
 import org.apache.iotdb.db.utils.TypeServices;
+import org.apache.iotdb.db.utils.sorter.TreeModelTabletEventSorter;
 import org.apache.iotdb.pipe.api.exception.PipeException;
 
 import com.sun.jna.Memory;
@@ -186,7 +186,7 @@ static String getClsIDFromProgID(final String progID) {
   }
 
   void transfer(final Tablet tablet) {
-    new PipeTreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
+    new TreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary();
     final List<IMeasurementSchema> schemas = tablet.getSchemas();
 
     for (int i = 0; i < schemas.size(); ++i) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableInsertTabletStatementGenerator.java

@@ -24,6 +24,7 @@
 import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
 import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
 import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement;
+import org.apache.iotdb.db.utils.sorter.TableModelTabletEventSorter;
 
 import org.apache.tsfile.block.column.Column;
 import org.apache.tsfile.enums.TSDataType;
@@ -106,6 +107,14 @@ public InsertTabletStatement constructInsertTabletStatement() {
     InsertTabletStatement insertTabletStatement = super.constructInsertTabletStatement();
     insertTabletStatement.setDatabaseName(databaseName);
     insertTabletStatement.setWriteToTable(true);
+    // The rows of the SELECT output are not ordered: a join, an ORDER BY DESC or a scan over
+    // several devices can emit them in any order, while the storage layer documents that it
+    // assumes the rows of a device are ascending in time. Restore that order here, before the
+    // statement leaves the INSERT INTO path.
+    final TableModelTabletEventSorter sorter =
+        new TableModelTabletEventSorter(insertTabletStatement);
+    sorter.sortByTimestampIfNecessary();
+    insertTabletStatement.setOriginalRowIndexes(sorter.getRowPermutation());
     return insertTabletStatement;
   }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java

@@ -1362,6 +1362,12 @@ protected int checkTTLInternal(TSStatus[] results, long ttl, boolean breakOnFirs
 
     /*
      * assume that batch has been sorted by client
+     *
+     * A table-model tablet holds the rows of several devices, so the out-of-ttl rows are not
+     * necessarily a prefix: a row placed after firstAliveLoc can still be out of ttl, and the write
+     * path below does not reject it. SeriesScanUtil applies the ttl lower bound when it scans, so
+     * such a row stays invisible instead of being read back. Dropping it needs a change in the
+     * write path as well.
      */
     int loc = 0;
     int firstAliveLoc = -1;

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertTablet.java

@@ -98,7 +98,9 @@
       deviceID2AttributeValues.put(deviceID, new Object[attrColumnIndices.size()]);
     }
 
-    for (int rowIndex = 0; rowIndex < insertTabletStatement.getRowCount(); rowIndex++) {
+    final int[] submittedRowOrder = getSubmittedRowOrder();
+    for (int k = 0; k < insertTabletStatement.getRowCount(); k++) {
+      final int rowIndex = submittedRowOrder == null ? k : submittedRowOrder[k];
       final Object[] attrValues =
           deviceID2AttributeValues.get(insertTabletStatement.getTableDeviceID(rowIndex));
       for (int attrColNum = 0; attrColNum < attrColumnIndices.size(); attrColNum++) {
@@ -120,16 +122,36 @@
     return new ArrayList<>(deviceID2AttributeValues.values());
   }
 
+  /**
+   * The row indexes in the order the statement was submitted in. The coordinator sorts the rows
+   * before planning, but the attribute of a device is the last non-null value of the rows as the
+   * client sent them, so that order has to be restored here and in {@link
+   * #prepareDeviceID2LastIdxMap()}.
+   */
+  private int[] getSubmittedRowOrder() {
+    final int[] originalRowIndexes = getInnerTreeStatement().getOriginalRowIndexes();
+    if (originalRowIndexes == null) {
+      return null;
+    }
+    final int[] submittedRowOrder = new int[originalRowIndexes.length];
+    for (int row = 0; row < originalRowIndexes.length; row++) {
+      submittedRowOrder[originalRowIndexes[row]] = row;
+    }
+    return submittedRowOrder;
+  }
+
   // The map cannot be maintained during construction because the IDeviceID may be reset later.
   private void prepareDeviceID2LastIdxMap() {
     if (deviceID2LastIdxMap != null) {
       return;
     }
     InsertTabletStatement insertTabletStatement = getInnerTreeStatement();
     deviceID2LastIdxMap = new LinkedHashMap<>(insertTabletStatement.getRowCount());
-    for (int i = 0; i < insertTabletStatement.getRowCount(); i++) {
-      IDeviceID deviceID = insertTabletStatement.getTableDeviceID(i);
-      deviceID2LastIdxMap.put(deviceID, i);
+    final int[] submittedRowOrder = getSubmittedRowOrder();
+    for (int k = 0; k < insertTabletStatement.getRowCount(); k++) {
+      final int rowIndex = submittedRowOrder == null ? k : submittedRowOrder[k];
+      IDeviceID deviceID = insertTabletStatement.getTableDeviceID(rowIndex);
+      deviceID2LastIdxMap.put(deviceID, rowIndex);
     }
     if (deviceID2LastIdxMap.size() == 1) {
       insertTabletStatement.setSingleDevice();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertTabletStatement.java

@@ -96,6 +96,13 @@ public static long getInstanceSize() {
 
   protected boolean singleDevice;
 
+  /**
+   * Set when the rows were reordered before planning: the row at position {@code i} came from row
+   * {@code originalRowIndexes[i]} of the order the statement was submitted in. {@code null} when
+   * nothing was moved.
+   */
+  protected int[] originalRowIndexes;
+
   protected int rowCount = 0;
 
   /**
@@ -218,6 +225,20 @@ public void setColumnCategories(TsTableColumnCategory[] columnCategories) {
     deviceIDs = null;
   }
 
+  /** Drops the device IDs cached from the tag columns. Called after the rows were reordered. */
+  public void clearDeviceIdCache() {
+    this.deviceIDs = null;
+  }
+
+  /** Records the permutation applied to the rows, or {@code null} when nothing was moved. */
+  public void setOriginalRowIndexes(final int[] originalRowIndexes) {
+    this.originalRowIndexes = originalRowIndexes;
+  }
+
+  public int[] getOriginalRowIndexes() {
+    return originalRowIndexes;
+  }
+
   public long[] getTimes() {
     return times;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java

@@ -730,6 +730,10 @@ public void insertTablet(
         int start = rangePair[0];
         int end = rangePair[1];
         try {
+          // Rows already marked out-of-ttl are not skipped here: the results are only forwarded
+          // when a previous fragment failed, so a device-grouped tablet can persist them. The query
+          // engine applies the ttl lower bound, so they stay invisible instead of being read back.
+          // See InsertTabletNode#checkTTLInternal.
           if (insertTabletNode.isAligned()) {
             pointInserted +=
                 workMemTable.insertAlignedTablet(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/InsertEventDataAdapter.java → iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/InsertEventDataAdapter.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.iotdb.db.pipe.sink.util.sorter;
+package org.apache.iotdb.db.utils.sorter;
 
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.file.metadata.IDeviceID;