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); 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/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/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 69% 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..1d8e7876bad6b 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,64 @@ 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)); + // The index already describes the sorted order, so rearranging the timestamps with it is O(N); + // sorting the timestamps again would be O(N log N). + final long[] timestamps = dataAdapter.getTimestamps(); + final int rowSize = dataAdapter.getRowSize(); + final long[] sortedTimestamps = new long[timestamps.length]; + for (int i = 0; i < rowSize; i++) { + sortedTimestamps[i] = timestamps[index[i]]; + } + dataAdapter.setTimestamps(sortedTimestamps); + } } 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/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)); + } } 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());