From 7320ce91bb8cbb43eb475191c5e980ae435b16e3 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 4 Nov 2026 15:56:52 +0800 Subject: [PATCH 1/3] Move the insert-event sorter out of the pipe package The sortedness check, the stable sort and the typed reorder of an insert tablet are needed by the write path too, so the sorter moves to org.apache.iotdb.db.utils.sorter and drops the Pipe prefix. It also used to reach into the pipe package for compactBitMaps; it now calls org.apache.iotdb.db.utils.BitMapUtils directly, so the query engine can use it without depending on pipe. The time-only sort moves from TableModelTabletEventSorter to its base class, because a statement sorter of either model can use it, and the base class now exposes the row permutation a sort applied, so a caller can keep the order the rows were submitted in. No behavior change otherwise: the pipe classes and tests only follow the new package. --- .../batch/PipeTabletEventTsFileBatch.java | 8 +-- .../request/PipeTransferTabletRawReq.java | 6 +- .../request/PipeTransferTabletRawReqV2.java | 12 ++-- .../protocol/opcda/OpcDaServerHandle.java | 4 +- .../sorter/InsertEventDataAdapter.java | 2 +- .../sorter/InsertEventSorter.java} | 69 +++++++++++++++++-- .../sorter/InsertTabletStatementAdapter.java | 2 +- .../sorter/TableModelTabletEventSorter.java} | 48 ++----------- .../util => utils}/sorter/TabletAdapter.java | 2 +- .../sorter/TreeModelTabletEventSorter.java} | 18 ++--- .../sink/PipeStatementEventSorterTest.java | 14 ++-- .../pipe/sink/PipeTabletEventSorterTest.java | 14 ++-- .../tagfilter/TabletTagFilterTest.java | 4 +- 13 files changed, 104 insertions(+), 99 deletions(-) rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/{pipe/sink/util => utils}/sorter/InsertEventDataAdapter.java (98%) rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/{pipe/sink/util/sorter/PipeInsertEventSorter.java => utils/sorter/InsertEventSorter.java} (73%) rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/{pipe/sink/util => utils}/sorter/InsertTabletStatementAdapter.java (98%) rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/{pipe/sink/util/sorter/PipeTableModelTabletEventSorter.java => utils/sorter/TableModelTabletEventSorter.java} (81%) rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/{pipe/sink/util => utils}/sorter/TabletAdapter.java (98%) rename iotdb-core/datanode/src/main/java/org/apache/iotdb/db/{pipe/sink/util/sorter/PipeTreeModelTabletEventSorter.java => utils/sorter/TreeModelTabletEventSorter.java} (83%) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java index 5ca18623e12ed..e1f39e545808b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/batch/PipeTabletEventTsFileBatch.java +++ b/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), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReq.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReq.java index 00907ed8302f5..4d57cac059d6c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReq.java +++ b/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)) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReqV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReqV2.java index dd6941d7f7d8f..ab14177a37083 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReqV2.java +++ b/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 { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaServerHandle.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaServerHandle.java index 8bd84d1db72a0..566d72776fbd2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcda/OpcDaServerHandle.java +++ b/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 schemas = tablet.getSchemas(); for (int i = 0; i < schemas.size(); ++i) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/InsertEventDataAdapter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/InsertEventDataAdapter.java similarity index 98% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/InsertEventDataAdapter.java rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/InsertEventDataAdapter.java index 62111cb4129ca..e8d8130d4d54a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/InsertEventDataAdapter.java +++ b/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; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/PipeInsertEventSorter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/InsertEventSorter.java similarity index 73% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/PipeInsertEventSorter.java rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/InsertEventSorter.java index 73c35b8cc7adb..27bde4d8bbe64 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/PipeInsertEventSorter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/InsertEventSorter.java @@ -17,10 +17,10 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.util.sorter; +package org.apache.iotdb.db.utils.sorter; -import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; +import org.apache.iotdb.db.utils.BitMapUtils; import org.apache.iotdb.db.utils.TypeServices; import org.apache.tsfile.enums.TSDataType; @@ -29,9 +29,11 @@ import org.apache.tsfile.write.record.Tablet; import java.lang.reflect.Array; +import java.util.Arrays; +import java.util.Comparator; import java.util.Objects; -public class PipeInsertEventSorter { +public class InsertEventSorter { protected final InsertEventDataAdapter dataAdapter; @@ -46,7 +48,7 @@ public class PipeInsertEventSorter { * * @param tablet the tablet to sort */ - public PipeInsertEventSorter(final Tablet tablet) { + public InsertEventSorter(final Tablet tablet) { this.dataAdapter = new TabletAdapter(tablet); } @@ -55,7 +57,7 @@ public PipeInsertEventSorter(final Tablet tablet) { * * @param statement the insert tablet statement to sort */ - public PipeInsertEventSorter(final InsertTabletStatement statement) { + public InsertEventSorter(final InsertTabletStatement statement) { this.dataAdapter = new InsertTabletStatementAdapter(statement); } @@ -64,7 +66,7 @@ public PipeInsertEventSorter(final InsertTabletStatement statement) { * * @param adapter the data adapter */ - protected PipeInsertEventSorter(final InsertEventDataAdapter adapter) { + protected InsertEventSorter(final InsertEventDataAdapter adapter) { this.dataAdapter = adapter; } @@ -107,7 +109,7 @@ protected void sortAndMayDeduplicateValuesAndBitMaps() { } if (bitMapsModified) { - dataAdapter.setBitMaps(PipeTabletUtils.compactBitMaps(bitMaps, deDuplicatedSize)); + dataAdapter.setBitMaps(BitMapUtils.compactBitMaps(bitMaps, deDuplicatedSize)); } } @@ -166,4 +168,57 @@ private int getLastNonnullIndex( } return index[lastNonnullIndex]; } + + /** Sort by time only, keeping the relative order of rows that share a timestamp. */ + public void sortByTimestampIfNecessary() { + if (dataAdapter == null || dataAdapter.getRowSize() == 0) { + return; + } + + final long[] timestamps = dataAdapter.getTimestamps(); + final int rowSize = dataAdapter.getRowSize(); + for (int i = 1; i < rowSize; ++i) { + if (timestamps[i] < timestamps[i - 1]) { + isSorted = false; + break; + } + } + + if (isSorted) { + return; + } + + index = new Integer[rowSize]; + for (int i = 0; i < rowSize; i++) { + index[i] = i; + } + sortTimestamps(); + + deDuplicatedIndex = null; + deDuplicatedSize = rowSize; + sortAndMayDeduplicateValuesAndBitMaps(); + } + + /** + * The permutation the last sort applied: the row at position {@code i} came from row {@code + * permutation[i]} of the input. {@code null} when the rows were already in order and nothing was + * moved. + */ + public int[] getRowPermutation() { + if (index == null) { + return null; + } + final int[] permutation = new int[index.length]; + for (int i = 0; i < index.length; i++) { + permutation[i] = index[i]; + } + return permutation; + } + + protected void sortTimestamps() { + // Index is sorted stably because it is Integer[] + Arrays.sort(this.index, Comparator.comparingLong(dataAdapter::getTimestamp)); + final long[] timestamps = dataAdapter.getTimestamps(); + Arrays.sort(timestamps, 0, dataAdapter.getRowSize()); + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/InsertTabletStatementAdapter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/InsertTabletStatementAdapter.java similarity index 98% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/InsertTabletStatementAdapter.java rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/InsertTabletStatementAdapter.java index 30f74a22965a8..fa4960321d6d3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/InsertTabletStatementAdapter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/InsertTabletStatementAdapter.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.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/PipeTableModelTabletEventSorter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/TableModelTabletEventSorter.java similarity index 81% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/PipeTableModelTabletEventSorter.java rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/TableModelTabletEventSorter.java index ba034cae3a310..4785c4cf20ad2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/PipeTableModelTabletEventSorter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/TableModelTabletEventSorter.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.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; @@ -33,7 +33,7 @@ import java.util.List; import java.util.Map; -public class PipeTableModelTabletEventSorter extends PipeInsertEventSorter { +public class TableModelTabletEventSorter extends InsertEventSorter { private int initIndexSize; /** @@ -41,7 +41,7 @@ public class PipeTableModelTabletEventSorter extends PipeInsertEventSorter { * * @param tablet the tablet to sort */ - public PipeTableModelTabletEventSorter(final Tablet tablet) { + public TableModelTabletEventSorter(final Tablet tablet) { super(tablet); deDuplicatedSize = tablet == null ? 0 : tablet.getRowSize(); } @@ -51,7 +51,7 @@ public PipeTableModelTabletEventSorter(final Tablet tablet) { * * @param statement the insert tablet statement to sort */ - public PipeTableModelTabletEventSorter(final InsertTabletStatement statement) { + public TableModelTabletEventSorter(final InsertTabletStatement statement) { super(statement); deDuplicatedSize = statement == null ? 0 : statement.getRowCount(); } @@ -171,44 +171,4 @@ private void deDuplicateTimestamps(final int startIndex, final int endIndex) { } deDuplicatedIndex[deDuplicatedSize++] = endIndex - 1; } - - /** Sort by time only. */ - public void sortByTimestampIfNecessary() { - if (dataAdapter == null || dataAdapter.getRowSize() == 0) { - return; - } - - final long[] timestamps = dataAdapter.getTimestamps(); - final int rowSize = dataAdapter.getRowSize(); - for (int i = 1; i < rowSize; ++i) { - final long currentTimestamp = timestamps[i]; - final long previousTimestamp = timestamps[i - 1]; - - if (currentTimestamp < previousTimestamp) { - isSorted = false; - break; - } - } - - if (isSorted) { - return; - } - - index = new Integer[rowSize]; - for (int i = 0; i < rowSize; i++) { - index[i] = i; - } - - if (!isSorted) { - sortTimestamps(); - } - - sortAndMayDeduplicateValuesAndBitMaps(); - } - - private void sortTimestamps() { - Arrays.sort(this.index, Comparator.comparingLong(dataAdapter::getTimestamp)); - final long[] timestamps = dataAdapter.getTimestamps(); - Arrays.sort(timestamps, 0, dataAdapter.getRowSize()); - } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/TabletAdapter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/TabletAdapter.java similarity index 98% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/TabletAdapter.java rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/TabletAdapter.java index b200127a5d451..d899e03d77fbf 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/TabletAdapter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/TabletAdapter.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; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/PipeTreeModelTabletEventSorter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/TreeModelTabletEventSorter.java similarity index 83% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/PipeTreeModelTabletEventSorter.java rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/TreeModelTabletEventSorter.java index 2a56b706463ff..675d691fd7a2f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/util/sorter/PipeTreeModelTabletEventSorter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/sorter/TreeModelTabletEventSorter.java @@ -17,23 +17,20 @@ * under the License. */ -package org.apache.iotdb.db.pipe.sink.util.sorter; +package org.apache.iotdb.db.utils.sorter; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; import org.apache.tsfile.write.record.Tablet; -import java.util.Arrays; -import java.util.Comparator; - -public class PipeTreeModelTabletEventSorter extends PipeInsertEventSorter { +public class TreeModelTabletEventSorter extends InsertEventSorter { /** * Constructor for Tablet. * * @param tablet the tablet to sort */ - public PipeTreeModelTabletEventSorter(final Tablet tablet) { + public TreeModelTabletEventSorter(final Tablet tablet) { super(tablet); deDuplicatedSize = tablet == null ? 0 : tablet.getRowSize(); } @@ -43,7 +40,7 @@ public PipeTreeModelTabletEventSorter(final Tablet tablet) { * * @param statement the insert tablet statement to sort */ - public PipeTreeModelTabletEventSorter(final InsertTabletStatement statement) { + public TreeModelTabletEventSorter(final InsertTabletStatement statement) { super(statement); deDuplicatedSize = statement == null ? 0 : statement.getRowCount(); } @@ -94,13 +91,6 @@ public void deduplicateAndSortTimestampsIfNecessary() { sortAndMayDeduplicateValuesAndBitMaps(); } - private void sortTimestamps() { - // Index is sorted stably because it is Integer[] - Arrays.sort(index, Comparator.comparingLong(dataAdapter::getTimestamp)); - final long[] timestamps = dataAdapter.getTimestamps(); - Arrays.sort(timestamps, 0, dataAdapter.getRowSize()); - } - private void deduplicateTimestamps() { deDuplicatedSize = 0; long[] timestamps = dataAdapter.getTimestamps(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeStatementEventSorterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeStatementEventSorterTest.java index 7c13d9764b3cf..d9c1af7a263c1 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeStatementEventSorterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeStatementEventSorterTest.java @@ -19,9 +19,9 @@ package org.apache.iotdb.db.pipe.sink; -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.tsfile.enums.TSDataType; import org.apache.tsfile.utils.Binary; @@ -82,7 +82,7 @@ public void testTreeModelDeduplicateAndSort() throws Exception { InsertTabletStatement statement = new InsertTabletStatement(tablet, true, null); // Sort using Statement - new PipeTreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary(); + new TreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary(); Assert.assertEquals(indices.size(), statement.getRowCount()); @@ -133,7 +133,7 @@ public void testTreeModelDeduplicate() throws Exception { InsertTabletStatement statement = new InsertTabletStatement(tablet, true, null); // Sort using Statement - new PipeTreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary(); + new TreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary(); Assert.assertEquals(indices.size(), statement.getRowCount()); @@ -198,7 +198,7 @@ public void testTreeModelSort() throws Exception { InsertTabletStatement statement = new InsertTabletStatement(tablet, true, null); // Sort using Statement - new PipeTreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary(); + new TreeModelTabletEventSorter(statement).deduplicateAndSortTimestampsIfNecessary(); Assert.assertEquals(indices.size(), statement.getRowCount()); @@ -246,7 +246,7 @@ public void doTableModelTest(final boolean hasDuplicates, final boolean isUnSort InsertTabletStatement statement = new InsertTabletStatement(tablet, false, "test_db"); // Sort using Statement - new PipeTableModelTabletEventSorter(statement).sortAndDeduplicateByDevIdTimestamp(); + new TableModelTabletEventSorter(statement).sortAndDeduplicateByDevIdTimestamp(); long[] timestamps = statement.getTimes(); final Object[] columns = statement.getColumns(); @@ -282,7 +282,7 @@ public void doTableModelTest1() throws Exception { InsertTabletStatement statement = new InsertTabletStatement(tablet, false, "test_db"); // Sort using Statement - new PipeTableModelTabletEventSorter(statement).sortByTimestampIfNecessary(); + new TableModelTabletEventSorter(statement).sortByTimestampIfNecessary(); long[] timestamps = statement.getTimes(); final Object[] columns = statement.getColumns(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeTabletEventSorterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeTabletEventSorterTest.java index eb5ffd475d361..3e02fdb3149af 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeTabletEventSorterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeTabletEventSorterTest.java @@ -19,8 +19,8 @@ package org.apache.iotdb.db.pipe.sink; -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.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; @@ -81,7 +81,7 @@ public void testTreeModelDeduplicateAndSort() { Assert.assertFalse(tablet.isSorted()); - new PipeTreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary(); + new TreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary(); Assert.assertTrue(tablet.isSorted()); @@ -129,7 +129,7 @@ public void testTreeModelDeduplicate() { Assert.assertTrue(tablet.isSorted()); - new PipeTreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary(); + new TreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary(); Assert.assertTrue(tablet.isSorted()); @@ -191,7 +191,7 @@ public void testTreeModelSort() { } } - new PipeTreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary(); + new TreeModelTabletEventSorter(tablet).deduplicateAndSortTimestampsIfNecessary(); Assert.assertTrue(tablet.isSorted()); @@ -233,7 +233,7 @@ public void testTableModelSort1() { public void doTableModelTest(final boolean hasDuplicates, final boolean isUnSorted) { final Tablet tablet = generateTablet("test", 10, hasDuplicates, isUnSorted); - new PipeTableModelTabletEventSorter(tablet).sortAndDeduplicateByDevIdTimestamp(); + new TableModelTabletEventSorter(tablet).sortAndDeduplicateByDevIdTimestamp(); long[] timestamps = tablet.getTimestamps(); for (int i = 1; i < tablet.getRowSize(); i++) { long time = timestamps[i]; @@ -256,7 +256,7 @@ public void doTableModelTest(final boolean hasDuplicates, final boolean isUnSort public void doTableModelTest1() { final Tablet tablet = generateTablet("test", 10, false, true); - new PipeTableModelTabletEventSorter(tablet).sortByTimestampIfNecessary(); + new TableModelTabletEventSorter(tablet).sortByTimestampIfNecessary(); long[] timestamps = tablet.getTimestamps(); for (int i = 1; i < tablet.getRowSize(); i++) { long time = timestamps[i]; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/tagfilter/TabletTagFilterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/tagfilter/TabletTagFilterTest.java index f7c8d02a9f30a..95ed152257d82 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/tagfilter/TabletTagFilterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/tagfilter/TabletTagFilterTest.java @@ -19,7 +19,7 @@ package org.apache.iotdb.db.subscription.tagfilter; -import org.apache.iotdb.db.pipe.sink.util.sorter.PipeTableModelTabletEventSorter; +import org.apache.iotdb.db.utils.sorter.TableModelTabletEventSorter; import org.apache.iotdb.rpc.subscription.config.TopicConfig; import org.apache.iotdb.rpc.subscription.config.TopicConstant; @@ -59,7 +59,7 @@ public void testPrefixSharesColumnDataWithoutChangingSourceWhenSorted() throws E // Sorting replaces columns and bitmap entries in place, so the outer arrays must be private. final Object originalColumn = tablet.getValues()[0]; final BitMap originalBitMap = tablet.getBitMaps()[0]; - new PipeTableModelTabletEventSorter(filtered).sortAndDeduplicateByDevIdTimestamp(); + new TableModelTabletEventSorter(filtered).sortAndDeduplicateByDevIdTimestamp(); Assert.assertEquals(1, filtered.getTimestamp(0)); Assert.assertNotSame(originalColumn, filtered.getValues()[0]); Assert.assertEquals(4, tablet.getRowSize()); From 7807aaa26fe7029f37eeb6811dcb0f04d3ea8782 Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 4 Nov 2026 15:56:59 +0800 Subject: [PATCH 2/3] Sort the tablets a table-model INSERT INTO writes `INSERT INTO ... SELECT` copies the rows of the child operator verbatim into the insert tablet, while the storage layer documents that it assumes the batch has been sorted by the client (InsertTabletNode#checkTTLInternal). The rows of a SELECT are not ordered: a join, an ORDER BY DESC or a scan over several devices can emit them in any order, so a tablet whose rows go backwards in time could write a row older than the device's last flush time into the sequence space. The resulting sequence TsFile overlapped the existing one, and queries returned duplicated rows and broke the time order, silently. Sort the tablet by time in TableInsertTabletStatementGenerator, i.e. only on the table-model INSERT INTO path, before the statement is submitted. The sort is stable, so rows that share a timestamp keep their relative order and the last one still wins, and it reuses the sorter that the pipe already had instead of a second implementation. A batch that is already ordered only costs one pass over its timestamps. Sorting moves rows, but an attribute is device metadata: it is the last non-null value of the rows as the SELECT emitted them. The statement records the permutation the sort applied and InsertTablet uses it, so the attribute still comes from the last submitted row. Other ingresses (the tree model, REST, MQTT, the session API) are out of scope: they are expected to submit ordered batches, as the storage layer documents. --- .../TableInsertTabletStatementGenerator.java | 9 + .../plan/node/write/InsertTabletNode.java | 6 + .../plan/relational/sql/ast/InsertTablet.java | 30 +- .../statement/crud/InsertTabletStatement.java | 21 ++ .../dataregion/memtable/TsFileProcessor.java | 4 + .../InsertTabletStatementGeneratorTest.java | 315 ++++++++++++++++++ .../relational/sql/ast/InsertTabletTest.java | 44 +++ 7 files changed, 425 insertions(+), 4 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/InsertTabletStatementGeneratorTest.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableInsertTabletStatementGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableInsertTabletStatementGenerator.java index 08a15eaddb5bd..a789219afd28b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableInsertTabletStatementGenerator.java +++ b/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; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java index 5c91d0af4eaaf..d747031856c5d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java +++ b/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; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertTablet.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertTablet.java index 639e35187a955..b0a112410531d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertTablet.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertTablet.java @@ -98,7 +98,9 @@ public List getAttributeValueList() { 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,6 +122,24 @@ public List getAttributeValueList() { 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) { @@ -127,9 +147,11 @@ private void prepareDeviceID2LastIdxMap() { } 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(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertTabletStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertTabletStatement.java index 7f9d7ce5551a1..4536283b7bff8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertTabletStatement.java +++ b/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; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index 4f8fbffecf1a8..f26949105ebb6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/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( 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..44a434b9c76d8 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/InsertTabletStatementGeneratorTest.java @@ -0,0 +1,315 @@ +/* + * 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 { + TableInsertTabletStatementGenerator generator = createGenerator(); + + TsBlockBuilder builder = createBlockBuilder(); + 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]); + Assert.assertArrayEquals(new int[] {1, 0}, statement.getOriginalRowIndexes()); + } + + @Test + public void testNullValueMovesWithItsRow() throws IllegalPathException { + TableInsertTabletStatementGenerator generator = createGenerator(); + + TsBlockBuilder builder = createBlockBuilder(); + append(builder, 100, 100); + builder.getTimeColumnBuilder().writeLong(15); + builder.getColumnBuilder(0).writeLong(15); + builder.getColumnBuilder(1).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 { + TableInsertTabletStatementGenerator generator = createGenerator(); + + TsBlockBuilder builder = createBlockBuilder(); + 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]); + Assert.assertNull(statement.getOriginalRowIndexes()); + } + + /** + * 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 inputLocations = new LinkedHashMap<>(); + inputLocations.put("time", new InputLocation(0, 0)); + for (int i = 0; i < measurements.length; i++) { + inputLocations.put(measurements[i], new InputLocation(0, i + 1)); + } + Map dataTypeMap = new LinkedHashMap<>(); + for (int i = 0; i < measurements.length; i++) { + dataTypeMap.put(measurements[i], dataTypes[i]); + } + TableInsertTabletStatementGenerator generator = + new TableInsertTabletStatementGenerator( + "test", + new PartialPath(new String[] {"table1"}), + inputLocations, + dataTypeMap, + Arrays.asList(dataTypes), + Collections.nCopies(measurements.length, TsTableColumnCategory.FIELD), + true, + 3, + "time"); + + final TSDataType[] inputTypes = new TSDataType[dataTypes.length + 1]; + inputTypes[0] = TSDataType.INT64; // the time column of the target table + System.arraycopy(dataTypes, 0, inputTypes, 1, dataTypes.length); + TsBlockBuilder builder = new TsBlockBuilder(Arrays.asList(inputTypes)); + // 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).writeLong(time); + builder.getColumnBuilder(1).writeInt((int) time); + builder.getColumnBuilder(2).writeLong(time); + builder.getColumnBuilder(3).writeFloat(time); + builder.getColumnBuilder(4).writeDouble(time); + builder.getColumnBuilder(5).writeBoolean(time % 20 != 10); + if (textIsNull) { + builder.getColumnBuilder(6).appendNull(); + } else { + builder + .getColumnBuilder(6) + .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 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 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); + // the rows follow time order, and each tag keeps its row; the null tag travels in the bitmap + Assert.assertArrayEquals(new long[] {15, 50, 100}, statement.getTimes()); + Binary[] tagColumn = (Binary[]) statement.getColumns()[0]; + Assert.assertFalse(statement.getBitMaps()[0].isMarked(0)); + Assert.assertTrue(statement.getBitMaps()[0].isMarked(1)); + Assert.assertEquals("a", tagColumn[0].toString()); + 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 { + TableInsertTabletStatementGenerator generator = createGenerator(); + + TsBlockBuilder first = createBlockBuilder(); + 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 = createBlockBuilder(); + 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 { + TableInsertTabletStatementGenerator generator = createGenerator(); + + TsBlockBuilder builder = createBlockBuilder(); + 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]); + } + + private TableInsertTabletStatementGenerator createGenerator() throws IllegalPathException { + Map inputLocations = new LinkedHashMap<>(); + inputLocations.put("time", new InputLocation(0, 0)); + inputLocations.put("s1", new InputLocation(0, 1)); + return new TableInsertTabletStatementGenerator( + "test", + new PartialPath(new String[] {"table1"}), + inputLocations, + Collections.singletonMap("s1", TSDataType.INT32), + Collections.singletonList(TSDataType.INT32), + Collections.singletonList(TsTableColumnCategory.FIELD), + false, + 10, + "time"); + } + + private static TsBlockBuilder createBlockBuilder() { + return new TsBlockBuilder(Arrays.asList(TSDataType.INT64, TSDataType.INT32)); + } + + private static void append(TsBlockBuilder builder, long time, int value) { + builder.getTimeColumnBuilder().writeLong(time); + builder.getColumnBuilder(0).writeLong(time); + builder.getColumnBuilder(1).writeInt(value); + builder.declarePosition(); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertTabletTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertTabletTest.java index f16e680502a24..08d0fe81ac8cc 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertTabletTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertTabletTest.java @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; +import org.apache.iotdb.db.utils.sorter.TableModelTabletEventSorter; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.utils.Binary; @@ -196,4 +197,47 @@ public void testDuplicateDeviceUsesLastNonNullAttributeValue() { assertEquals(1, deviceIdList.size()); assertArrayEquals(new Object[] {"d1"}, deviceIdList.get(0)); } + + /** + * The coordinator sorts the rows before planning, but an attribute is device metadata: it is the + * last non-null value of the rows as the client submitted them, so the sort must not change it. + */ + @Test + public void testAttributeValueFollowsSubmittedRowOrderAfterSort() { + final InsertTabletStatement innerStmt = new InsertTabletStatement(); + innerStmt.setDevicePath(new PartialPath("table1", false)); + innerStmt.setMeasurements(new String[] {"deviceId", "attr1", "measurement"}); + innerStmt.setColumnCategories( + new TsTableColumnCategory[] { + TsTableColumnCategory.TAG, TsTableColumnCategory.ATTRIBUTE, TsTableColumnCategory.FIELD + }); + innerStmt.setDataTypes( + new TSDataType[] {TSDataType.STRING, TSDataType.STRING, TSDataType.STRING}); + // the client sends the newer row first: its attribute "old" must not win over the last one + innerStmt.setTimes(new long[] {100, 15}); + innerStmt.setRowCount(2); + innerStmt.setColumns( + new Object[] { + new Binary[] { + new Binary("d1", StandardCharsets.UTF_8), new Binary("d1", StandardCharsets.UTF_8) + }, + new Binary[] { + new Binary("old", StandardCharsets.UTF_8), new Binary("new", StandardCharsets.UTF_8) + }, + new Binary[] { + new Binary("m100", StandardCharsets.UTF_8), new Binary("m15", StandardCharsets.UTF_8) + } + }); + + final TableModelTabletEventSorter sorter = new TableModelTabletEventSorter(innerStmt); + sorter.sortByTimestampIfNecessary(); + innerStmt.setOriginalRowIndexes(sorter.getRowPermutation()); + assertArrayEquals(new long[] {15, 100}, Arrays.copyOf(innerStmt.getTimes(), 2)); + + final InsertTablet insertTablet = new InsertTablet(innerStmt, null); + final List attributeValueList = insertTablet.getAttributeValueList(); + assertEquals(1, attributeValueList.size()); + assertArrayEquals( + new Object[] {new Binary("new", StandardCharsets.UTF_8)}, attributeValueList.get(0)); + } } From 0c035395f503a4bf92f4d545533822e2a003821e Mon Sep 17 00:00:00 2001 From: ColinLee Date: Wed, 4 Nov 2026 15:57:02 +0800 Subject: [PATCH 3/3] Add regression coverage for out-of-order tablets Reproduce the overlap end to end through SQL: build a device whose sequence file is closed, insert the rows [100, 15] through INSERT INTO ... SELECT, and read them back. Without the sort the older row is written into a new sequence TsFile that overlaps the previous one, and the query returns a duplicated, unordered result; the test asserts the exact row sequence in both scan directions. --- .../it/insertquery/IoTDBInsertQueryIT.java | 77 +++++++++++++++++++ 1 file changed, 77 insertions(+) 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..44f14b4129bc0 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 @@ -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). + * + *

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). + * + *

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 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 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 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 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);