This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_flat_device in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 59cd8969857fb29eaaf3c379999eee0becb91f1d Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 2 17:13:18 2026 +0800 ver2 --- .../it/insertquery/IoTDBFlatDeviceIT.java | 341 ++++++++++++++ .../plan/relational/metadata/FlatDeviceEntry.java | 85 ++++ .../dataregion/memtable/AbstractMemTable.java | 49 +- .../memtable/AlignedWritableMemChunk.java | 188 +++++--- .../memtable/AlignedWritableMemChunkGroup.java | 47 +- .../db/utils/datastructure/AlignedTVList.java | 2 +- .../utils/datastructure/AlignedTVListFactory.java | 6 + .../datastructure/ColumnSortedAlignedTVList.java | 454 ++++++++++++++++++ .../planner/node/DeviceTableScanNodeTest.java | 65 +++ .../memtable/FlatDeviceMemTableTest.java | 515 +++++++++++++++++++++ .../ColumnSortedAlignedTVListTest.java | 490 ++++++++++++++++++++ .../commons/schema/table/TsTablePropertyTest.java | 52 +++ .../schema/table/column/TagColumnSchemaTest.java | 44 ++ 13 files changed, 2241 insertions(+), 97 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java new file mode 100644 index 00000000000..9df4d54c8aa --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java @@ -0,0 +1,341 @@ +/* + * 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.relational.it.insertquery; + +import org.apache.iotdb.isession.ITableSession; +import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.TableClusterIT; +import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; +import org.apache.iotdb.itbase.env.BaseEnv; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.common.RowRecord; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +@RunWith(IoTDBTestRunner.class) +@Category({TableLocalStandaloneIT.class, TableClusterIT.class}) +public class IoTDBFlatDeviceIT { + + private static final String DATABASE = "flat_device_it"; + private static final String TABLE = "sensor"; + + @BeforeClass + public static void setUpClass() throws Exception { + EnvFactory.getEnv().initClusterEnvironment(); + } + + @Before + public void setUp() throws SQLException { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE " + DATABASE); + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE " + + TABLE + + " (tag_a STRING TAG, tag_b STRING TAG, attr STRING ATTRIBUTE, value DOUBLE FIELD)" + + " WITH (flat_device=true)"); + } + } + + @After + public void tearDown() throws SQLException { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("DROP DATABASE IF EXISTS " + DATABASE); + } + } + + @AfterClass + public static void tearDownClass() { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + // Covers row writes before and after flush, including equal-time compound TAG keys, current + // ATTRIBUTE lookup, hidden TAG reconstruction, ATTRIBUTE filtering, and aggregation fallback. + @Test + public void testFlatDeviceRowsKeepCompoundKeysAfterFlush() throws SQLException { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + + try (ResultSet resultSet = statement.executeQuery("SHOW CREATE TABLE " + TABLE)) { + assertTrue(resultSet.next()); + assertTrue(resultSet.getString(2).contains("flat_device=true")); + } + + // Equal timestamps with different TAG values are separate rows in the flat group. The + // explicit order also verifies that both TAG columns participate in the row key. + statement.execute( + "INSERT INTO " + + TABLE + + " (time, tag_a, tag_b, attr, value) VALUES (10, 'b', '2', 'attr-b2', 20.0)"); + statement.execute( + "INSERT INTO " + + TABLE + + " (time, tag_a, tag_b, attr, value) VALUES (10, 'a', '2', 'attr-a2', 12.0)"); + statement.execute( + "INSERT INTO " + + TABLE + + " (time, tag_a, tag_b, attr, value) VALUES (10, 'a', '1', 'attr-a1', 11.0)"); + statement.execute( + "INSERT INTO " + + TABLE + + " (time, tag_a, tag_b, attr, value) VALUES (11, 'a', '1', 'attr-a1-later', 21.0)"); + + List<String> expectedRows = + Arrays.asList( + "10,a,1,attr-a1-later,11.0", + "10,a,2,attr-a2,12.0", + "10,b,2,attr-b2,20.0", + "11,a,1,attr-a1-later,21.0"); + assertRows(statement.executeQuery(query()), expectedRows); + assertRows( + statement.executeQuery(queryForAttribute("attr-a2")), + Arrays.asList("10,a,2,attr-a2,12.0")); + assertRowsWithoutTags( + statement.executeQuery(queryWithoutTagsForAttribute("attr-a2")), + Arrays.asList("10,attr-a2,12.0")); + assertCount(statement, 4); + + statement.execute("FLUSH"); + assertRows(statement.executeQuery(query()), expectedRows); + assertRows( + statement.executeQuery(queryForAttribute("attr-a2")), + Arrays.asList("10,a,2,attr-a2,12.0")); + assertRowsWithoutTags( + statement.executeQuery(queryWithoutTagsForAttribute("attr-a2")), + Arrays.asList("10,attr-a2,12.0")); + assertCount(statement, 4); + + // Write another logical device at an already flushed timestamp. This produces unsequence + // data and verifies that the cross-source merge keeps different time-plus-TAG row keys. + statement.execute( + "INSERT INTO " + + TABLE + + " (time, tag_a, tag_b, attr, value) VALUES (10, 'c', '1', 'attr-c1', 31.0)"); + statement.execute( + "INSERT INTO " + + TABLE + + " (time, tag_a, tag_b, attr, value) VALUES (10, 'b', '2', 'attr-b2-later', 25.0)"); + expectedRows = + Arrays.asList( + "10,a,1,attr-a1-later,11.0", + "10,a,2,attr-a2,12.0", + "10,b,2,attr-b2-later,25.0", + "10,c,1,attr-c1,31.0", + "11,a,1,attr-a1-later,21.0"); + assertRows(statement.executeQuery(query()), expectedRows); + assertCount(statement, 5); + + statement.execute("FLUSH"); + assertRows(statement.executeQuery(query()), expectedRows); + assertCount(statement, 5); + } + } + + @Test + public void testFlatDevicePropertyCannotBeAltered() throws SQLException { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + + assertFlatDeviceAlterRejected( + statement, "ALTER TABLE " + TABLE + " SET PROPERTIES flat_device=false"); + assertFlatDeviceAlterRejected( + statement, "ALTER TABLE " + TABLE + " SET PROPERTIES flat_device=true"); + assertFlatDeviceAlterRejected( + statement, "ALTER TABLE " + TABLE + " SET PROPERTIES flat_device=DEFAULT"); + + try (ResultSet resultSet = statement.executeQuery("SHOW CREATE TABLE " + TABLE)) { + assertTrue(resultSet.next()); + assertTrue(resultSet.getString(2).contains("flat_device=true")); + } + + statement.execute( + "CREATE TABLE normal_sensor (tag_a STRING TAG, value DOUBLE FIELD) WITH (ttl='INF')"); + assertFlatDeviceAlterRejected( + statement, "ALTER TABLE normal_sensor SET PROPERTIES flat_device=true"); + } + } + + // Covers Tablet writes whose rows span multiple logical devices and verifies the shared physical + // group preserves time-plus-TAG ordering across flush. + @Test + public void testFlatDeviceTabletWriteUsesOneGroup() throws Exception { + try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) { + session.executeNonQueryStatement("USE " + DATABASE); + + List<IMeasurementSchema> schemas = + Arrays.asList( + new MeasurementSchema("tag_a", TSDataType.STRING), + new MeasurementSchema("tag_b", TSDataType.STRING), + new MeasurementSchema("attr", TSDataType.STRING), + new MeasurementSchema("value", TSDataType.DOUBLE)); + Tablet tablet = + new Tablet( + TABLE, + IMeasurementSchema.getMeasurementNameList(schemas), + IMeasurementSchema.getDataTypeList(schemas), + Arrays.asList( + ColumnCategory.TAG, + ColumnCategory.TAG, + ColumnCategory.ATTRIBUTE, + ColumnCategory.FIELD), + 4); + + addRow(tablet, 12, "c", "2", "attr-c2", 32.0); + addRow(tablet, 10, "b", "1", "attr-b1", 11.0); + addRow(tablet, 10, "b", "2", "attr-b2", 12.0); + addRow(tablet, 10, "a", "2", "attr-a2", 10.0); + session.insert(tablet); + + List<String> expectedRows = + Arrays.asList( + "10,a,2,attr-a2,10.0", + "10,b,1,attr-b1,11.0", + "10,b,2,attr-b2,12.0", + "12,c,2,attr-c2,32.0"); + assertRows(session.executeQueryStatement(query()), expectedRows); + session.executeNonQueryStatement("FLUSH"); + assertRows(session.executeQueryStatement(query()), expectedRows); + } + } + + private static String query() { + return "SELECT time, tag_a, tag_b, attr, value FROM " + TABLE + " ORDER BY time, tag_a, tag_b"; + } + + private static String queryForAttribute(String attribute) { + return "SELECT time, tag_a, tag_b, attr, value FROM " + + TABLE + + " WHERE attr = '" + + attribute + + "' ORDER BY time, tag_a, tag_b"; + } + + private static String queryWithoutTagsForAttribute(String attribute) { + return "SELECT time, attr, value FROM " + + TABLE + + " WHERE attr = '" + + attribute + + "' ORDER BY time"; + } + + private static void assertCount(Statement statement, long expected) throws SQLException { + try (ResultSet resultSet = statement.executeQuery("SELECT count(value) FROM " + TABLE)) { + assertTrue(resultSet.next()); + assertEquals(expected, resultSet.getLong(1)); + assertFalse(resultSet.next()); + } + } + + private static void assertFlatDeviceAlterRejected(Statement statement, String sql) + throws SQLException { + try { + statement.execute(sql); + fail("Expected flat_device alteration to be rejected: " + sql); + } catch (SQLException e) { + assertEquals("701: Table property 'flat_device' cannot be altered.", e.getMessage()); + } + } + + private static void addRow( + Tablet tablet, long time, String tagA, String tagB, String attribute, double value) { + int row = tablet.getRowSize(); + tablet.addTimestamp(row, time); + tablet.addValue("tag_a", row, tagA); + tablet.addValue("tag_b", row, tagB); + tablet.addValue("attr", row, attribute); + tablet.addValue("value", row, value); + } + + private static void assertRows(ResultSet resultSet, List<String> expected) throws SQLException { + List<String> actual = new ArrayList<>(); + while (resultSet.next()) { + actual.add( + resultSet.getLong(1) + + "," + + resultSet.getString(2) + + "," + + resultSet.getString(3) + + "," + + resultSet.getString(4) + + "," + + resultSet.getDouble(5)); + } + assertEquals(expected, actual); + assertFalse(actual.isEmpty()); + } + + private static void assertRowsWithoutTags(ResultSet resultSet, List<String> expected) + throws SQLException { + List<String> actual = new ArrayList<>(); + while (resultSet.next()) { + actual.add( + resultSet.getLong(1) + "," + resultSet.getString(2) + "," + resultSet.getDouble(3)); + } + assertEquals(expected, actual); + assertFalse(actual.isEmpty()); + } + + private static void assertRows(SessionDataSet dataSet, List<String> expected) throws Exception { + List<String> actual = new ArrayList<>(); + while (dataSet.hasNext()) { + RowRecord row = dataSet.next(); + actual.add( + row.getFields().get(0).getLongV() + + "," + + row.getFields().get(1).getBinaryV() + + "," + + row.getFields().get(2).getBinaryV() + + "," + + row.getFields().get(3).getBinaryV() + + "," + + row.getFields().get(4).getDoubleV()); + } + assertEquals(expected, actual); + assertFalse(actual.isEmpty()); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/FlatDeviceEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/FlatDeviceEntry.java new file mode 100644 index 00000000000..66777fc47d0 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/FlatDeviceEntry.java @@ -0,0 +1,85 @@ +/* + * 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.plan.relational.metadata; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.utils.Binary; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Physical table-name entry that resolves row-level logical device metadata for flat-device scans. + */ +public class FlatDeviceEntry extends AlignedDeviceEntry { + + private final Map<IDeviceID, Binary[]> logicalDeviceAttributes; + private final int[] tagMeasurementColumnIndices; + + public FlatDeviceEntry( + IDeviceID physicalDeviceID, + List<DeviceEntry> logicalDeviceEntries, + int[] tagMeasurementColumnIndices) { + super(physicalDeviceID, new Binary[0]); + this.logicalDeviceAttributes = new HashMap<>(); + for (DeviceEntry logicalDeviceEntry : logicalDeviceEntries) { + logicalDeviceAttributes.put( + logicalDeviceEntry.getDeviceID(), logicalDeviceEntry.getAttributeColumnValues()); + } + this.tagMeasurementColumnIndices = tagMeasurementColumnIndices.clone(); + } + + public Binary getAttributeValue(TsBlock measurementDataBlock, int rowIndex, int attributeIndex) { + IDeviceID logicalDeviceID = getLogicalDeviceID(measurementDataBlock, rowIndex); + Binary[] attributes = logicalDeviceAttributes.get(logicalDeviceID); + return attributes == null || attributeIndex >= attributes.length + ? null + : attributes[attributeIndex]; + } + + public boolean containsLogicalDevice(TsBlock measurementDataBlock, int rowIndex) { + return logicalDeviceAttributes.containsKey(getLogicalDeviceID(measurementDataBlock, rowIndex)); + } + + public Column getTagColumn(TsBlock measurementDataBlock, int tagIndex) { + return measurementDataBlock.getColumn(tagMeasurementColumnIndices[tagIndex]); + } + + public int[] getTagMeasurementColumnIndices() { + return tagMeasurementColumnIndices.clone(); + } + + private IDeviceID getLogicalDeviceID(TsBlock measurementDataBlock, int rowIndex) { + String[] segments = new String[tagMeasurementColumnIndices.length + 1]; + segments[0] = String.valueOf(deviceID.segment(0)); + for (int i = 0; i < tagMeasurementColumnIndices.length; i++) { + int columnIndex = tagMeasurementColumnIndices[i]; + if (columnIndex < 0 || measurementDataBlock.getColumn(columnIndex).isNull(rowIndex)) { + segments[i + 1] = null; + } else { + segments[i + 1] = + String.valueOf(measurementDataBlock.getColumn(columnIndex).getObject(rowIndex)); + } + } + return IDeviceID.Factory.DEFAULT_FACTORY.create(segments); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java index a07f9f724e4..b219a46ea00 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java @@ -77,6 +77,7 @@ import java.util.Map.Entry; import java.util.Objects; import java.util.Optional; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Supplier; import java.util.stream.Collectors; public abstract class AbstractMemTable implements IMemTable { @@ -183,7 +184,7 @@ public abstract class AbstractMemTable implements IMemTable { private IWritableMemChunkGroup createAlignedMemChunkGroupIfNotExistAndGet( IDeviceID deviceId, List<IMeasurementSchema> schemaList, - int sortColumnCount, + Supplier<List<Integer>> sortColumnIndicesSupplier, boolean useColumnSortedList) { List<IMeasurementSchema> filteredSchemaList = schemaList.stream().filter(Objects::nonNull).collect(Collectors.toList()); @@ -192,8 +193,12 @@ public abstract class AbstractMemTable implements IMemTable { deviceId, k -> { seriesNumber += filteredSchemaList.size(); + // Evaluate the supplier only for a newly created group. return new AlignedWritableMemChunkGroup( - filteredSchemaList, k.isTableModel(), sortColumnCount, useColumnSortedList); + filteredSchemaList, + k.isTableModel(), + sortColumnIndicesSupplier.get(), + useColumnSortedList); }); for (IMeasurementSchema schema : schemaList) { if (schema != null && !memChunkGroup.contains(schema.getMeasurementName())) { @@ -297,20 +302,19 @@ public abstract class AbstractMemTable implements IMemTable { writableValues, insertRowNode.getColumnCategories(), insertRowNode.isFlatDevice()); - int sortColumnCount = 0; + Supplier<List<Integer>> sortColumnIndicesSupplier = Collections::emptyList; if (insertRowNode.isFlatDevice()) { - sortColumnCount = getTagColumnCount(insertRowNode, schemaList); + sortColumnIndicesSupplier = () -> getTagColumnIndices(insertRowNode, schemaList); } IWritableMemChunkGroup memChunkGroup = createAlignedMemChunkGroupIfNotExistAndGet( insertRowNode.getMemTableDeviceID(insertRowNode.getDeviceID()), schemaList, - sortColumnCount, + sortColumnIndicesSupplier, insertRowNode.isFlatDevice()); if (memChunkGroup instanceof AlignedWritableMemChunkGroup alignedMemChunkGroup) { alignedMemChunkGroup.writeRow( insertRowNode.getTime(), writableValues, schemaList, insertRowNode.getColumnCategories()); - alignedMemChunkGroup.addSortColumnCount(sortColumnCount); } else { memChunkGroup.writeRow(insertRowNode.getTime(), writableValues, schemaList); } @@ -453,7 +457,8 @@ public abstract class AbstractMemTable implements IMemTable { long insertTime, Object[] objectValue) { IWritableMemChunkGroup memChunkGroup = - createAlignedMemChunkGroupIfNotExistAndGet(deviceId, schemaList, 0, false); + createAlignedMemChunkGroupIfNotExistAndGet( + deviceId, schemaList, Collections::emptyList, false); memChunkGroup.writeRow(insertTime, objectValue, schemaList); } @@ -520,9 +525,9 @@ public abstract class AbstractMemTable implements IMemTable { if (schemaList.isEmpty()) { return; } - int sortColumnCount = 0; + Supplier<List<Integer>> sortColumnIndicesSupplier = Collections::emptyList; if (insertTabletNode.isFlatDevice()) { - sortColumnCount = getTagColumnCount(insertTabletNode, schemaList); + sortColumnIndicesSupplier = () -> getTagColumnIndices(insertTabletNode, schemaList); } final List<Pair<IDeviceID, Integer>> deviceEndOffsetPair = insertTabletNode.splitByDevice(start, end); @@ -532,7 +537,7 @@ public abstract class AbstractMemTable implements IMemTable { int splitEnd = pair.right; IWritableMemChunkGroup memChunkGroup = createAlignedMemChunkGroupIfNotExistAndGet( - deviceID, schemaList, sortColumnCount, insertTabletNode.isFlatDevice()); + deviceID, schemaList, sortColumnIndicesSupplier, insertTabletNode.isFlatDevice()); if (memChunkGroup instanceof AlignedWritableMemChunkGroup alignedMemChunkGroup) { alignedMemChunkGroup.writeTablet( insertTabletNode.getTimes(), @@ -543,7 +548,6 @@ public abstract class AbstractMemTable implements IMemTable { splitStart, splitEnd, results); - alignedMemChunkGroup.addSortColumnCount(sortColumnCount); } else { memChunkGroup.writeTablet( insertTabletNode.getTimes(), @@ -558,19 +562,22 @@ public abstract class AbstractMemTable implements IMemTable { } } - private static int getTagColumnCount(InsertNode insertNode, List<IMeasurementSchema> schemaList) { + private static List<Integer> getTagColumnIndices( + InsertNode insertNode, List<IMeasurementSchema> schemaList) { TsTableColumnCategory[] categories = insertNode.getColumnCategories(); - int tagColumnCount = 0; - for (int i = 0; i < schemaList.size(); i++) { - if (schemaList.get(i) != null - && categories != null - && i < categories.length - && categories[i] == TsTableColumnCategory.TAG) { - tagColumnCount++; + List<Integer> tagColumnIndices = new ArrayList<>(); + // InsertNode keeps TAG columns in a prefix; filtering null schemas only compacts that prefix. + for (int i = 0; + categories != null + && i < categories.length + && i < schemaList.size() + && categories[i] == TsTableColumnCategory.TAG; + i++) { + if (schemaList.get(i) != null) { + tagColumnIndices.add(tagColumnIndices.size()); } } - // TAG columns are the leading value columns, so their aligned indices form a contiguous prefix. - return tagColumnCount; + return tagColumnIndices; } private static IMeasurementSchema getMemTableMeasurementSchema(InsertNode insertNode, int index) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java index 6172c2c8163..5e6b76b2f5a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java @@ -77,7 +77,8 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { private List<AlignedTVList> sortedList; private long sortedRowCount = 0; private final boolean ignoreAllNullRows; - private int sortColumnCount; + // TAG columns are appended physically, so their logical key order is tracked separately. + private final List<Integer> sortColumnIndices; private final boolean useColumnSortedList; private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig(); @@ -87,29 +88,34 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { private EncryptParameter encryptParameter; + @TestOnly public AlignedWritableMemChunk(List<IMeasurementSchema> schemaList, boolean isTableModel) { this(schemaList, isTableModel, 0, false, EncryptUtils.getEncryptParameter()); } + @TestOnly public AlignedWritableMemChunk( - List<IMeasurementSchema> schemaList, boolean isTableModel, int sortColumnCount) { + List<IMeasurementSchema> schemaList, + boolean isTableModel, + int sortColumnCount, + boolean useColumnSortedList) { this( schemaList, isTableModel, - sortColumnCount, - sortColumnCount > 0, + createContiguousSortColumnIndices(sortColumnCount), + useColumnSortedList, EncryptUtils.getEncryptParameter()); } public AlignedWritableMemChunk( List<IMeasurementSchema> schemaList, boolean isTableModel, - int sortColumnCount, + List<Integer> sortColumnIndices, boolean useColumnSortedList) { this( schemaList, isTableModel, - sortColumnCount, + sortColumnIndices, useColumnSortedList, EncryptUtils.getEncryptParameter()); } @@ -127,16 +133,31 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { int sortColumnCount, boolean useColumnSortedList, EncryptParameter encryptParameter) { + this( + schemaList, + isTableModel, + createContiguousSortColumnIndices(sortColumnCount), + useColumnSortedList, + encryptParameter); + } + + private AlignedWritableMemChunk( + List<IMeasurementSchema> schemaList, + boolean isTableModel, + List<Integer> sortColumnIndices, + boolean useColumnSortedList, + EncryptParameter encryptParameter) { this.measurementIndexMap = new LinkedHashMap<>(); this.measurementCategoryMap = new LinkedHashMap<>(); this.dataTypes = new ArrayList<>(); - // Keep the chunk's schema order mutable because late schema discovery may insert a TAG. + this.sortColumnIndices = + useColumnSortedList ? new ArrayList<>(sortColumnIndices) : new ArrayList<>(); + // Keep the chunk's schema order mutable because late schema discovery may append a TAG. this.schemaList = new ArrayList<>(schemaList); - this.sortColumnCount = useColumnSortedList ? sortColumnCount : 0; this.useColumnSortedList = useColumnSortedList; for (int i = 0; i < schemaList.size(); i++) { measurementIndexMap.put(schemaList.get(i).getMeasurementName(), i); - if (i < this.sortColumnCount) { + if (this.sortColumnIndices.contains(i)) { measurementCategoryMap.put( schemaList.get(i).getMeasurementName(), TsTableColumnCategory.TAG); } @@ -145,44 +166,45 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { this.list = !useColumnSortedList ? AlignedTVListFactory.createAlignedList(dataTypes) - : AlignedTVListFactory.createColumnSortedAlignedList(dataTypes, this.sortColumnCount); + : AlignedTVListFactory.createColumnSortedAlignedList(dataTypes, this.sortColumnIndices); this.sortedList = new ArrayList<>(); this.ignoreAllNullRows = !isTableModel; this.encryptParameter = encryptParameter; } - private AlignedWritableMemChunk( - List<IMeasurementSchema> schemaList, AlignedTVList list, boolean isTableModel) { - this(schemaList, list, isTableModel, 0, EncryptUtils.getEncryptParameter()); + private static List<Integer> createContiguousSortColumnIndices(int sortColumnCount) { + List<Integer> indices = new ArrayList<>(sortColumnCount); + for (int i = 0; i < sortColumnCount; i++) { + indices.add(i); + } + return indices; } private AlignedWritableMemChunk( - List<IMeasurementSchema> schemaList, - AlignedTVList list, - boolean isTableModel, - EncryptParameter encryptParameter) { - this(schemaList, list, isTableModel, 0, encryptParameter); + List<IMeasurementSchema> schemaList, AlignedTVList list, boolean isTableModel) { + this(schemaList, list, isTableModel, EncryptUtils.getEncryptParameter()); } private AlignedWritableMemChunk( List<IMeasurementSchema> schemaList, AlignedTVList list, boolean isTableModel, - int sortColumnCount, EncryptParameter encryptParameter) { this.measurementIndexMap = new LinkedHashMap<>(); this.measurementCategoryMap = new LinkedHashMap<>(); this.schemaList = new ArrayList<>(schemaList); - this.sortColumnCount = - list instanceof ColumnSortedAlignedTVList columnSortedList - ? columnSortedList.getSortColumnCount() - : sortColumnCount; this.useColumnSortedList = list instanceof ColumnSortedAlignedTVList; + this.sortColumnIndices = + this.useColumnSortedList + ? new ArrayList<>(((ColumnSortedAlignedTVList) list).getSortColumnIndices()) + : new ArrayList<>(); for (int i = 0; i < schemaList.size(); i++) { measurementIndexMap.put(schemaList.get(i).getMeasurementName(), i); - if (i < this.sortColumnCount) { + } + for (int columnIndex : sortColumnIndices) { + if (columnIndex < schemaList.size()) { measurementCategoryMap.put( - schemaList.get(i).getMeasurementName(), TsTableColumnCategory.TAG); + schemaList.get(columnIndex).getMeasurementName(), TsTableColumnCategory.TAG); } } this.list = list; @@ -296,7 +318,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { !this.useColumnSortedList ? AlignedTVListFactory.createAlignedList(new ArrayList<>(dataTypes)) : AlignedTVListFactory.createColumnSortedAlignedList( - new ArrayList<>(dataTypes), this.sortColumnCount); + new ArrayList<>(dataTypes), this.sortColumnIndices); this.dataTypes = this.list.getTsDataTypes(); } @@ -320,21 +342,6 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { } } - public void addSortColumnCount(int sortColumnCount) { - if (sortColumnCount <= this.sortColumnCount - || !(list instanceof ColumnSortedAlignedTVList columnSortedList)) { - return; - } - // Historical lists keep the prefix they were sorted with; the comparator treats later key - // columns as null instead of invalidating and resorting those lists. - columnSortedList.addSortColumnCount(sortColumnCount); - for (int columnIndex = this.sortColumnCount; columnIndex < sortColumnCount; columnIndex++) { - measurementCategoryMap.put( - schemaList.get(columnIndex).getMeasurementName(), TsTableColumnCategory.TAG); - } - this.sortColumnCount = sortColumnCount; - } - @Override public void writeAlignedTablet( long[] times, @@ -372,12 +379,20 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { * * @param schemaListInInsertPlan Contains all existed schema in InsertPlan. If some timeseries * have been deleted, there will be null in its slot. - * @return columnIndexArray: schemaList[i] is schema of columns[columnIndexArray[i]] + * @return value and bitmap arrays ordered by the chunk's physical columns */ private Pair<Object[], BitMap[]> checkAndReorderColumnValuesInInsertPlan( List<IMeasurementSchema> schemaListInInsertPlan, Object[] columnValues, BitMap[] bitMaps) { - Object[] reorderedColumnValues = new Object[schemaList.size()]; - BitMap[] reorderedBitMaps = bitMaps == null ? null : new BitMap[schemaList.size()]; + return checkAndReorderColumnValuesInInsertPlan( + schemaListInInsertPlan, columnValues, bitMaps, null); + } + + private Pair<Object[], BitMap[]> checkAndReorderColumnValuesInInsertPlan( + List<IMeasurementSchema> schemaListInInsertPlan, + Object[] columnValues, + BitMap[] bitMaps, + TsTableColumnCategory[] columnCategories) { + boolean columnOrderMatchesChunk = true; for (int i = 0; i < schemaListInInsertPlan.size(); i++) { IMeasurementSchema measurementSchema = schemaListInInsertPlan.get(i); if (measurementSchema != null) { @@ -386,25 +401,62 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { // We need to extend a new column in AlignedMemChunk and AlignedTVList. // And the reorderedColumnValues should extend one more column for the new measurement if (index == null) { + TsTableColumnCategory columnCategory = + columnCategories != null && i < columnCategories.length ? columnCategories[i] : null; + TsTableColumnCategory measurementCategory = + measurementCategoryMap.computeIfAbsent( + measurementSchema.getMeasurementName(), key -> columnCategory); + boolean isTagColumn = + useColumnSortedList && measurementCategory == TsTableColumnCategory.TAG; index = this.list.getTsDataTypes().size(); - this.measurementIndexMap.put(schemaListInInsertPlan.get(i).getMeasurementName(), index); - this.schemaList.add(schemaListInInsertPlan.get(i)); - this.list.extendColumn(schemaListInInsertPlan.get(i).getType()); - reorderedColumnValues = - Arrays.copyOf(reorderedColumnValues, reorderedColumnValues.length + 1); - if (reorderedBitMaps != null) { - reorderedBitMaps = Arrays.copyOf(reorderedBitMaps, reorderedBitMaps.length + 1); + insertColumn(index, measurementSchema); + if (isTagColumn) { + sortColumnIndices.add(index); + ((ColumnSortedAlignedTVList) this.list).addSortColumn(index); } } - reorderedColumnValues[index] = columnValues[i]; - if (bitMaps != null) { - reorderedBitMaps[index] = bitMaps[i]; + if (index != i) { + columnOrderMatchesChunk = false; } } } + // The aligned TVList expects one value/bitmap array per physical column. Reuse the input + // arrays when the plan already uses the chunk's physical order and has no missing columns. + if (columnOrderMatchesChunk + && schemaListInInsertPlan.size() == schemaList.size() + && columnValues.length == schemaList.size() + && (bitMaps == null || bitMaps.length == schemaList.size())) { + return new Pair<>(columnValues, bitMaps); + } + Object[] reorderedColumnValues = new Object[schemaList.size()]; + BitMap[] reorderedBitMaps = bitMaps == null ? null : new BitMap[schemaList.size()]; + for (int i = 0; i < schemaListInInsertPlan.size(); i++) { + IMeasurementSchema measurementSchema = schemaListInInsertPlan.get(i); + if (measurementSchema == null) { + continue; + } + Integer index = this.measurementIndexMap.get(measurementSchema.getMeasurementName()); + reorderedColumnValues[index] = columnValues[i]; + if (bitMaps != null) { + reorderedBitMaps[index] = bitMaps[i]; + } + } return new Pair<>(reorderedColumnValues, reorderedBitMaps); } + /** + * Appends a newly discovered measurement to the working list and updates the physical index map. + * + * <p>Flat-device TAG columns are appended physically and their indices are recorded separately in + * {@code sortColumnIndices}. Historical TVLists remain unchanged because queries may still hold + * their original column-index mapping. + */ + private void insertColumn(int columnIndex, IMeasurementSchema measurementSchema) { + this.list.insertColumn(columnIndex, measurementSchema.getType()); + this.schemaList.add(measurementSchema); + this.measurementIndexMap.put(measurementSchema.getMeasurementName(), columnIndex); + } + public long[] getAnySatisfiedTimestamp( List<List<TimeRange>> deletionList, List<BitMap> bitMaps, @@ -503,7 +555,9 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { // non-null value in multiple timestamps for the same column. BitMap currentRowNullValueBitmap = null; for (int column = 0; column < schemaList.size(); column++) { - if (alignedTVList.isNullValue(indices == null ? row : indices[j], column)) { + // Late-added columns do not exist in historical lists and are therefore null there. + if (column >= alignedTVList.getTsDataTypes().size() + || alignedTVList.isNullValue(indices == null ? row : indices[j], column)) { continue; } @@ -566,7 +620,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { @TestOnly int getSortColumnCount() { - return sortColumnCount; + return sortColumnIndices.size(); } @Override @@ -651,10 +705,14 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { int columnIndex = measurementIndexMap.get(measurementId); list.deleteColumn(columnIndex); for (AlignedTVList alignedTvList : sortedList) { - alignedTvList.deleteColumn(columnIndex); + // Historical lists may predate a late-added column and therefore be narrower. + if (columnIndex < alignedTvList.getTsDataTypes().size()) { + alignedTvList.deleteColumn(columnIndex); + } } IMeasurementSchema schemaToBeRemoved = schemaList.get(columnIndex); measurementIndexMap.remove(schemaToBeRemoved.getMeasurementName()); + measurementCategoryMap.remove(schemaToBeRemoved.getMeasurementName()); } @Override @@ -1287,13 +1345,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { AlignedTVList list = AlignedTVListFactory.deserialize(stream); AlignedWritableMemChunk chunk = new AlignedWritableMemChunk( - schemaList, - list, - isTableModel, - list instanceof ColumnSortedAlignedTVList - ? ((ColumnSortedAlignedTVList) list).getSortColumnCount() - : 0, - EncryptUtils.getEncryptParameter()); + schemaList, list, isTableModel, EncryptUtils.getEncryptParameter()); chunk.sortedList = sortedList; return chunk; } @@ -1309,13 +1361,7 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { AlignedTVList list = AlignedTVListFactory.deserialize(stream); return new AlignedWritableMemChunk( - schemaList, - list, - isTableModel, - list instanceof ColumnSortedAlignedTVList - ? ((ColumnSortedAlignedTVList) list).getSortColumnCount() - : 0, - EncryptUtils.getEncryptParameter()); + schemaList, list, isTableModel, EncryptUtils.getEncryptParameter()); } public List<IMeasurementSchema> getSchemaList() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java index d50d26ab09b..80641a33cee 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java @@ -21,6 +21,7 @@ package org.apache.iotdb.db.storageengine.dataregion.memtable; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.path.AlignedPath; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.exception.DataTypeInconsistentException; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; @@ -58,10 +59,31 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { public AlignedWritableMemChunkGroup( List<IMeasurementSchema> schemaList, boolean isTableModel, - int sortColumnCount, + List<Integer> sortColumnIndices, boolean useColumnSortedList) { memChunk = - new AlignedWritableMemChunk(schemaList, isTableModel, sortColumnCount, useColumnSortedList); + new AlignedWritableMemChunk( + schemaList, isTableModel, sortColumnIndices, useColumnSortedList); + } + + public AlignedWritableMemChunkGroup( + List<IMeasurementSchema> schemaList, + boolean isTableModel, + int sortColumnCount, + boolean useColumnSortedList) { + this( + schemaList, + isTableModel, + createContiguousSortColumnIndices(sortColumnCount), + useColumnSortedList); + } + + private static List<Integer> createContiguousSortColumnIndices(int sortColumnCount) { + List<Integer> indices = new ArrayList<>(sortColumnCount); + for (int i = 0; i < sortColumnCount; i++) { + indices.add(i); + } + return indices; } @TestOnly @@ -86,6 +108,19 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { memChunk.writeAlignedTablet(times, columns, bitMaps, schemaList, start, end, results); } + public void writeTablet( + long[] times, + Object[] columns, + BitMap[] bitMaps, + List<IMeasurementSchema> schemaList, + TsTableColumnCategory[] columnCategories, + int start, + int end, + TSStatus[] results) { + memChunk.writeAlignedTablet( + times, columns, bitMaps, schemaList, columnCategories, start, end, results); + } + @Override public void release() { memChunk.release(); @@ -114,8 +149,12 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { memChunk.writeAlignedPoints(insertTime, objectValue, schemaList); } - public void addSortColumnCount(int sortColumnCount) { - memChunk.addSortColumnCount(sortColumnCount); + public void writeRow( + long insertTime, + Object[] objectValue, + List<IMeasurementSchema> schemaList, + TsTableColumnCategory[] columnCategories) { + memChunk.writeAlignedPoints(insertTime, objectValue, schemaList, columnCategories); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java index a45d4783196..1bf9bcf1050 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java @@ -651,7 +651,7 @@ public abstract class AlignedTVList extends TVList { * * <p>Aligned TVLists keep all value-column metadata in parallel arrays. Keeping the insertion * here (instead of modifying only {@link #values}) ensures historical rows, null bitmaps, and - * memory accounting remain aligned when a flat-device TAG is inserted before existing fields. + * memory accounting remain aligned whenever a caller changes the physical column layout. */ public synchronized void insertColumn(int columnIndex, TSDataType dataType) { if (columnIndex < 0 || columnIndex > values.size()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java index aca0d3a0c87..a146c867fa0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java @@ -53,6 +53,12 @@ public final class AlignedTVListFactory { return new ColumnSortedAlignedTVList(dataTypes, keyColumnCount); } + /** Creates a column-sorted aligned TVList using explicit physical key-column indices. */ + public static AlignedTVList createColumnSortedAlignedList( + List<TSDataType> dataTypes, List<Integer> keyColumnIndices) { + return new ColumnSortedAlignedTVList(dataTypes, keyColumnIndices, true); + } + public static AlignedTVList deserialize(DataInputStream stream) throws IOException { byte encodedDataType = stream.readByte(); int dataTypeCount = stream.readInt(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java new file mode 100644 index 00000000000..1206eed5476 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java @@ -0,0 +1,454 @@ +/* + * 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.utils.datastructure; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.db.i18n.DataNodeMiscMessages; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.utils.Pair; +import org.apache.tsfile.utils.ReadWriteForEncodingUtils; +import org.apache.tsfile.utils.ReadWriteIOUtils; + +import java.io.DataInputStream; +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; + +/** + * An aligned TVList that orders rows by timestamp followed by selected value columns. Selected + * columns are compared in the configured order, with null values before non-null values. + */ +public class ColumnSortedAlignedTVList extends AlignedTVList { + + static final byte WAL_DATA_TYPE_MARKER = -1; + + /** Physical value-column indices used by the compound row key. */ + private final List<Integer> sortColumnIndices; + + private final TimSort policy; + // Keep value-column ordering separate so timestamp-only binary searches retain their prefix + // guarantees even when equal-time rows need a compound-key sort. + private boolean valueColumnsSorted = true; + + /** + * Creates a regular working column-sorted list. Value-column containers must be initialized for + * writes; clone paths use the constructor overload to defer allocation for columns moved later. + */ + ColumnSortedAlignedTVList(List<TSDataType> types, int sortColumnCount) { + this(types, sortColumnCount, true); + } + + ColumnSortedAlignedTVList( + List<TSDataType> types, int sortColumnCount, boolean initializeValueColumns) { + this(types, createContiguousColumnIndices(sortColumnCount), initializeValueColumns); + } + + ColumnSortedAlignedTVList( + List<TSDataType> types, List<Integer> sortColumnIndices, boolean initializeValueColumns) { + super(validateSortColumns(types, sortColumnIndices), initializeValueColumns); + this.sortColumnIndices = new ArrayList<>(sortColumnIndices); + this.policy = + new TimSort(this) { + @Override + public int compare(int left, int right) { + return compareRows(left, right); + } + }; + } + + /** Creates the physical-index list used by the legacy contiguous sort-column constructor. */ + private static List<Integer> createContiguousColumnIndices(int sortColumnCount) { + if (sortColumnCount < 0) { + throw new IndexOutOfBoundsException(sortColumnCount); + } + List<Integer> sortColumnIndices = new ArrayList<>(sortColumnCount); + for (int columnIndex = 0; columnIndex < sortColumnCount; columnIndex++) { + sortColumnIndices.add(columnIndex); + } + return sortColumnIndices; + } + + /** + * Validates physical key-column indices before the superclass allocates matching value columns. + * The type list is returned unchanged so validation can be composed in the {@code super(...)} + * invocation while preserving the constructor's original list object. + */ + private static List<TSDataType> validateSortColumns( + List<TSDataType> types, List<Integer> sortColumnIndices) { + Objects.requireNonNull(types); + Objects.requireNonNull(sortColumnIndices); + Set<Integer> uniqueColumnIndices = new HashSet<>(); + for (int columnIndex : sortColumnIndices) { + if (columnIndex < 0 || columnIndex >= types.size() || !uniqueColumnIndices.add(columnIndex)) { + throw new IndexOutOfBoundsException(columnIndex); + } + TSDataType dataType = Objects.requireNonNull(types.get(columnIndex)); + if (dataType == TSDataType.VECTOR || dataType == TSDataType.UNKNOWN) { + throw new IllegalArgumentException( + String.format( + DataNodeMiscMessages.MISC_EXCEPTION_DATA_TYPE_S_IS_NOT_SUPPORTED_5D5C02E4, + dataType)); + } + } + return types; + } + + public int getSortColumnCount() { + return sortColumnIndices.size(); + } + + @Override + protected int getRowKeyColumnCount() { + return sortColumnIndices.size(); + } + + /** Adds a physical value-column index to the compound row key. */ + public synchronized void addSortColumn(int columnIndex) { + if (sortColumnIndices.contains(columnIndex)) { + return; + } + validateSortColumns(dataTypes, List.of(columnIndex)); + sortColumnIndices.add(columnIndex); + valueColumnsSorted = false; + seqRowCount = 0; + } + + public synchronized List<Integer> getSortColumnIndices() { + return new ArrayList<>(sortColumnIndices); + } + + @Override + public synchronized boolean isSorted() { + return sorted && valueColumnsSorted; + } + + @Override + public synchronized void putAlignedValue(long timestamp, Object[] value) { + int previousSeqRowCount = seqRowCount; + super.putAlignedValue(timestamp, value); + if (!valueColumnsSorted) { + seqRowCount = previousSeqRowCount; + } else if (rowCount > 1 && compareRows(rowCount - 2, rowCount - 1) > 0) { + valueColumnsSorted = false; + seqRowCount = Math.min(seqRowCount, rowCount - 1); + } + } + + @Override + public synchronized void putAlignedValues( + long[] time, Object[] value, BitMap[] bitMaps, int start, int end, TSStatus[] results) { + int previousRowCount = rowCount; + int previousSeqRowCount = seqRowCount; + super.putAlignedValues(time, value, bitMaps, start, end, results); + if (!valueColumnsSorted) { + seqRowCount = previousSeqRowCount; + return; + } + for (int i = Math.max(1, previousRowCount); i < rowCount; i++) { + if (compareRows(i - 1, i) > 0) { + valueColumnsSorted = false; + seqRowCount = Math.min(seqRowCount, i); + return; + } + } + } + + @Override + public synchronized Pair<Integer, Boolean> delete( + long lowerBound, long upperBound, int columnIndex) { + Pair<Integer, Boolean> result = super.delete(lowerBound, upperBound, columnIndex); + if (result.left > 0 && sortColumnIndices.contains(columnIndex)) { + valueColumnsSorted = false; + seqRowCount = 0; + } + return result; + } + + @Override + public synchronized void deleteColumn(int columnIndex) { + super.deleteColumn(columnIndex); + if (rowCount > 1 && sortColumnIndices.contains(columnIndex)) { + valueColumnsSorted = false; + seqRowCount = 0; + } + } + + @Override + protected void onColumnsReleased(Set<Integer> retainedColumns) { + if (rowCount > 1) { + for (int columnIndex : sortColumnIndices) { + if (!retainedColumns.contains(columnIndex)) { + valueColumnsSorted = false; + seqRowCount = 0; + return; + } + } + } + } + + @Override + protected AlignedTVList createNewList(List<TSDataType> types, boolean initializeValueColumns) { + return new ColumnSortedAlignedTVList(types, sortColumnIndices, initializeValueColumns); + } + + @Override + protected void copyAdditionalStateTo(AlignedTVList cloneList) { + ((ColumnSortedAlignedTVList) cloneList).valueColumnsSorted = valueColumnsSorted; + } + + @Override + public int serializedSize() { + // serializeToWAL writes the compound row-key index count and every physical column index. + return super.serializedSize(Integer.BYTES * (sortColumnIndices.size() + 1)); + } + + @Override + public void serializeToWAL(IWALByteBufferView buffer) { + buffer.put(WAL_DATA_TYPE_MARKER); + serializeWALDataTypes(buffer); + buffer.putInt(sortColumnIndices.size()); + for (int columnIndex : sortColumnIndices) { + buffer.putInt(columnIndex); + } + buffer.putInt(rowCount); + serializeWALRows(buffer); + } + + @SuppressWarnings("java:S6541") + static ColumnSortedAlignedTVList deserialize(DataInputStream stream, int dataTypeCount) + throws IOException { + List<TSDataType> dataTypes = new ArrayList<>(dataTypeCount); + for (int columnIndex = 0; columnIndex < dataTypeCount; columnIndex++) { + dataTypes.add(ReadWriteIOUtils.readDataType(stream)); + } + int sortColumnCount = stream.readInt(); + List<Integer> sortColumnIndices = new ArrayList<>(sortColumnCount); + for (int i = 0; i < sortColumnCount; i++) { + sortColumnIndices.add(stream.readInt()); + } + ColumnSortedAlignedTVList tvList = + new ColumnSortedAlignedTVList(dataTypes, sortColumnIndices, true); + + int rowCount = stream.readInt(); + long[] times = new long[rowCount]; + for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { + times[rowIndex] = stream.readLong(); + } + + Object[] values = new Object[dataTypeCount]; + BitMap[] bitMaps = new BitMap[dataTypeCount]; + for (int columnIndex = 0; columnIndex < dataTypeCount; columnIndex++) { + BitMap bitMap = BitMap.createBitMapDynamically(rowCount); + Object valuesOfOneColumn = + switch (dataTypes.get(columnIndex)) { + case TEXT, BLOB, STRING, OBJECT -> { + Binary[] binaryValues = new Binary[rowCount]; + for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { + binaryValues[rowIndex] = ReadWriteIOUtils.readBinary(stream); + if (ReadWriteIOUtils.readBool(stream)) { + bitMap.mark(rowIndex); + } + } + yield binaryValues; + } + case FLOAT -> { + float[] floatValues = new float[rowCount]; + for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { + floatValues[rowIndex] = stream.readFloat(); + if (ReadWriteIOUtils.readBool(stream)) { + bitMap.mark(rowIndex); + } + } + yield floatValues; + } + case INT32, DATE -> { + int[] intValues = new int[rowCount]; + for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { + intValues[rowIndex] = stream.readInt(); + if (ReadWriteIOUtils.readBool(stream)) { + bitMap.mark(rowIndex); + } + } + yield intValues; + } + case INT64, TIMESTAMP -> { + long[] longValues = new long[rowCount]; + for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { + longValues[rowIndex] = stream.readLong(); + if (ReadWriteIOUtils.readBool(stream)) { + bitMap.mark(rowIndex); + } + } + yield longValues; + } + case DOUBLE -> { + double[] doubleValues = new double[rowCount]; + for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { + doubleValues[rowIndex] = stream.readDouble(); + if (ReadWriteIOUtils.readBool(stream)) { + bitMap.mark(rowIndex); + } + } + yield doubleValues; + } + case BOOLEAN -> { + boolean[] booleanValues = new boolean[rowCount]; + for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { + booleanValues[rowIndex] = ReadWriteIOUtils.readBool(stream); + if (ReadWriteIOUtils.readBool(stream)) { + bitMap.mark(rowIndex); + } + } + yield booleanValues; + } + case VECTOR, UNKNOWN -> + throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT); + }; + values[columnIndex] = valuesOfOneColumn; + bitMaps[columnIndex] = bitMap; + } + + tvList.putAlignedValues(times, values, bitMaps, 0, rowCount, null); + if (stream.read() == 1) { + int length = ReadWriteForEncodingUtils.readVarInt(stream); + byte[] bytes = new byte[length]; + stream.readFully(bytes); + tvList.timeColDeletedMap = new BitMap(rowCount, bytes); + } + return tvList; + } + + @Override + public synchronized int sort() { + if (!isSorted()) { + policy.checkSortedTimestampsAndIndices(); + policy.sort(0, rowCount); + } + policy.clearSortedValue(); + policy.clearSortedTime(); + sorted = true; + valueColumnsSorted = true; + seqRowCount = rowCount; + return rowCount; + } + + @Override + public synchronized void clear() { + super.clear(); + policy.clearSortedTime(); + policy.clearSortedValue(); + valueColumnsSorted = true; + } + + private int compareRows(int left, int right) { + return compareRowKey(left, this, right); + } + + @Override + public int compareRowKey(int leftRowIndex, AlignedTVList rightList, int rightRowIndex) { + int result = Long.compare(getTime(leftRowIndex), rightList.getTime(rightRowIndex)); + if (result != 0) { + return result; + } + + List<Integer> rightSortColumnIndices = + rightList instanceof ColumnSortedAlignedTVList rightColumnSortedList + ? rightColumnSortedList.sortColumnIndices + : List.of(); + int leftValueIndex = getValueIndex(leftRowIndex); + int rightValueIndex = rightList.getValueIndex(rightRowIndex); + int compareColumnCount = Math.max(sortColumnIndices.size(), rightSortColumnIndices.size()); + for (int keyIndex = 0; keyIndex < compareColumnCount; keyIndex++) { + int leftColumnIndex = + keyIndex < sortColumnIndices.size() ? sortColumnIndices.get(keyIndex) : -1; + int rightColumnIndex = + keyIndex < rightSortColumnIndices.size() ? rightSortColumnIndices.get(keyIndex) : -1; + result = + compareValues( + leftValueIndex, rightList, rightValueIndex, leftColumnIndex, rightColumnIndex); + if (result != 0) { + return result; + } + } + return 0; + } + + private int compareValues( + int leftValueIndex, + AlignedTVList rightList, + int rightValueIndex, + int leftColumnIndex, + int rightColumnIndex) { + // Historical lists keep the key definition they were sorted with. A key column introduced + // later is outside that definition and therefore participates in cross-list comparison as null. + boolean leftNull = + leftColumnIndex < 0 + || leftColumnIndex >= dataTypes.size() + || isNullValue(leftValueIndex, leftColumnIndex); + boolean rightNull = + rightColumnIndex < 0 + || rightColumnIndex >= rightList.dataTypes.size() + || rightList.isNullValue(rightValueIndex, rightColumnIndex); + if (leftNull || rightNull) { + return leftNull == rightNull ? 0 : (leftNull ? -1 : 1); + } + + TSDataType dataType = dataTypes.get(leftColumnIndex); + return switch (dataType) { + case BOOLEAN -> + Boolean.compare( + getBooleanByValueIndex(leftValueIndex, leftColumnIndex), + rightList.getBooleanByValueIndex(rightValueIndex, rightColumnIndex)); + case INT32, DATE -> + Integer.compare( + getIntByValueIndex(leftValueIndex, leftColumnIndex), + rightList.getIntByValueIndex(rightValueIndex, rightColumnIndex)); + case INT64, TIMESTAMP -> + Long.compare( + getLongByValueIndex(leftValueIndex, leftColumnIndex), + rightList.getLongByValueIndex(rightValueIndex, rightColumnIndex)); + case FLOAT -> + Float.compare( + getFloatByValueIndex(leftValueIndex, leftColumnIndex), + rightList.getFloatByValueIndex(rightValueIndex, rightColumnIndex)); + case DOUBLE -> + Double.compare( + getDoubleByValueIndex(leftValueIndex, leftColumnIndex), + rightList.getDoubleByValueIndex(rightValueIndex, rightColumnIndex)); + case TEXT, BLOB, STRING, OBJECT -> + // TimSort identifies descending runs with an exact -1 comparison. + Integer.signum( + getBinaryByValueIndex(leftValueIndex, leftColumnIndex) + .compareTo(rightList.getBinaryByValueIndex(rightValueIndex, rightColumnIndex))); + case VECTOR, UNKNOWN -> + throw new IllegalStateException( + String.format( + DataNodeMiscMessages.MISC_EXCEPTION_DATA_TYPE_S_IS_NOT_SUPPORTED_5D5C02E4, + dataType)); + }; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNodeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNodeTest.java new file mode 100644 index 00000000000..a60360875a3 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNodeTest.java @@ -0,0 +1,65 @@ +/* + * 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.plan.relational.planner.node; + +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.QualifiedObjectName; +import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; + +import org.junit.Test; + +import java.nio.ByteBuffer; +import java.util.Collections; + +import static org.junit.Assert.assertEquals; + +public class DeviceTableScanNodeTest { + + @Test + public void testFlatDeviceFlagSurvivesCloneAndSerialization() { + for (boolean flatDevice : new boolean[] {false, true}) { + DeviceTableScanNode node = + new DeviceTableScanNode( + new PlanNodeId("scan"), + new QualifiedObjectName("database", "table"), + Collections.emptyList(), + Collections.emptyMap(), + Collections.emptyList(), + Collections.emptyMap(), + Ordering.ASC, + null, + null, + 0, + 0, + false, + false, + flatDevice); + + assertEquals(flatDevice, node.clone().isFlatDevice()); + + ByteBuffer buffer = ByteBuffer.allocate(1024); + node.serialize(buffer); + buffer.flip(); + DeviceTableScanNode deserialized = (DeviceTableScanNode) PlanNodeType.deserialize(buffer); + assertEquals(flatDevice, deserialized.isFlatDevice()); + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableTest.java new file mode 100644 index 00000000000..008da09f2a6 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableTest.java @@ -0,0 +1,515 @@ +/* + * 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.storageengine.dataregion.memtable; + +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; +import org.apache.iotdb.db.utils.datastructure.AlignedTVList; +import org.apache.iotdb.db.utils.datastructure.ColumnSortedAlignedTVList; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.Assert; +import org.junit.Test; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.stream.Collectors; + +public class FlatDeviceMemTableTest { + + private static final String TABLE_NAME = "table1"; + private static final String[] MEASUREMENTS = {"tag1", "tag2", "attr1", "field1"}; + private static final TSDataType[] DATA_TYPES = { + TSDataType.STRING, TSDataType.STRING, TSDataType.STRING, TSDataType.INT64 + }; + private static final MeasurementSchema[] SCHEMAS = { + new MeasurementSchema("tag1", TSDataType.STRING), + new MeasurementSchema("tag2", TSDataType.STRING), + new MeasurementSchema("attr1", TSDataType.STRING), + new MeasurementSchema("field1", TSDataType.INT64) + }; + private static final TsTableColumnCategory[] CATEGORIES = { + TsTableColumnCategory.TAG, + TsTableColumnCategory.TAG, + TsTableColumnCategory.ATTRIBUTE, + TsTableColumnCategory.FIELD + }; + + // Covers flat row writes with equal timestamps and different tags. All rows must share the + // table-name group, ATTRIBUTE must be excluded, and sorting must use time then both TAG columns. + @Test + public void testFlatRowsUseOneColumnSortedGroup() { + PrimitiveMemTable memTable = new PrimitiveMemTable("db", "0"); + memTable.insertAlignedRow(newRow(binary("b"), binary("x"), 30L)); + memTable.insertAlignedRow(newRow(binary("a"), binary("z"), 20L)); + memTable.insertAlignedRow(newRow(binary("a"), binary("y"), 15L)); + memTable.insertAlignedRow(newRow(null, binary("z"), 10L)); + + Assert.assertEquals(1, memTable.getMemTableMap().size()); + IDeviceID deviceID = memTable.getMemTableMap().keySet().iterator().next(); + Assert.assertArrayEquals(new Object[] {TABLE_NAME}, deviceID.getSegments()); + + AlignedWritableMemChunkGroup group = + (AlignedWritableMemChunkGroup) memTable.getWritableMemChunkGroup(deviceID); + Assert.assertTrue(group.contains("tag1")); + Assert.assertTrue(group.contains("tag2")); + Assert.assertTrue(group.contains("field1")); + Assert.assertFalse(group.contains("attr1")); + Assert.assertEquals(3, memTable.getSeriesNumber()); + Assert.assertEquals(12, memTable.getTotalValueCount()); + Assert.assertEquals(1, memTable.getNullValueCount()); + + AlignedWritableMemChunk chunk = group.getAlignedMemChunk(); + Assert.assertEquals(TSEncoding.DICTIONARY, chunk.getSchemaList().get(0).getEncodingType()); + Assert.assertEquals(TSEncoding.DICTIONARY, chunk.getSchemaList().get(1).getEncodingType()); + Assert.assertNotEquals(TSEncoding.DICTIONARY, chunk.getSchemaList().get(2).getEncodingType()); + AlignedTVList tvList = chunk.getWorkingTVList(); + Assert.assertTrue(tvList instanceof ColumnSortedAlignedTVList); + Assert.assertEquals(2, ((ColumnSortedAlignedTVList) tvList).getSortColumnCount()); + + tvList.sort(); + Assert.assertArrayEquals(new long[] {10, 15, 20, 30}, payloads(tvList)); + Assert.assertTrue(tvList.isNullValue(tvList.getValueIndex(0), 0)); + + chunk.handoverAlignedTvList(); + Assert.assertTrue(chunk.getSortedList().get(0) instanceof ColumnSortedAlignedTVList); + Assert.assertTrue(chunk.getWorkingTVList() instanceof ColumnSortedAlignedTVList); + } + + // The tablet keeps its complete logical device IDs for partition routing, while the MemTable + // still merges all rows into one table-name group containing TAG and FIELD columns. + @Test + public void testFlatTabletKeepsLogicalDevicesAndUsesOneMemTableGroup() throws Exception { + RelationalInsertTabletNode node = newTablet(); + Assert.assertEquals(2, node.splitByDevice(0, node.getRowCount()).size()); + Assert.assertArrayEquals( + new Object[] {TABLE_NAME, "b", "x"}, + node.splitByDevice(0, node.getRowCount()).get(0).left.getSegments()); + Assert.assertArrayEquals( + new Object[] {TABLE_NAME, "a", "z"}, + node.splitByDevice(0, node.getRowCount()).get(1).left.getSegments()); + + PrimitiveMemTable memTable = new PrimitiveMemTable("db", "0"); + Assert.assertEquals(9, memTable.insertAlignedTablet(node, 0, node.getRowCount(), null)); + Assert.assertEquals(1, memTable.getMemTableMap().size()); + + AlignedWritableMemChunkGroup group = + (AlignedWritableMemChunkGroup) memTable.getMemTableMap().values().iterator().next(); + Assert.assertTrue(group.contains("tag1")); + Assert.assertTrue(group.contains("tag2")); + Assert.assertTrue(group.contains("field1")); + Assert.assertFalse(group.contains("attr1")); + Assert.assertEquals( + TSEncoding.DICTIONARY, group.getAlignedMemChunk().getSchemaList().get(0).getEncodingType()); + Assert.assertEquals( + TSEncoding.DICTIONARY, group.getAlignedMemChunk().getSchemaList().get(1).getEncodingType()); + Assert.assertNotEquals( + TSEncoding.DICTIONARY, group.getAlignedMemChunk().getSchemaList().get(2).getEncodingType()); + + AlignedTVList tvList = group.getAlignedMemChunk().getWorkingTVList(); + Assert.assertTrue(tvList instanceof ColumnSortedAlignedTVList); + tvList.sort(); + Assert.assertArrayEquals(new long[] {10, 11, 20}, payloads(tvList)); + } + + // Covers the compatibility path for ordinary aligned Row and Tablet writes. Neither path should + // allocate sort-column metadata or switch away from the timestamp-only AlignedTVList. + @Test + public void testNonFlatWritesKeepSortColumnIndicesNull() throws Exception { + RelationalInsertRowNode rowNode = newRow(binary("a"), binary("x"), 10L); + rowNode.setFlatDevice(false); + PrimitiveMemTable rowMemTable = new PrimitiveMemTable("db", "0"); + rowMemTable.insertAlignedRow(rowNode); + assertNoSortColumns(rowMemTable); + Assert.assertNotEquals( + TSEncoding.DICTIONARY, + ((AlignedWritableMemChunkGroup) rowMemTable.getMemTableMap().values().iterator().next()) + .getAlignedMemChunk() + .getSchemaList() + .get(0) + .getEncodingType()); + + RelationalInsertTabletNode tabletNode = newTablet(); + tabletNode.setFlatDevice(false); + PrimitiveMemTable tabletMemTable = new PrimitiveMemTable("db", "0"); + tabletMemTable.insertAlignedTablet(tabletNode, 0, tabletNode.getRowCount(), null); + for (IWritableMemChunkGroup group : tabletMemTable.getMemTableMap().values()) { + assertNoSortColumns((AlignedWritableMemChunkGroup) group); + Assert.assertNotEquals( + TSEncoding.DICTIONARY, + ((AlignedWritableMemChunkGroup) group) + .getAlignedMemChunk() + .getSchemaList() + .get(0) + .getEncodingType()); + } + } + + // A later flat write may introduce several TAG columns. They are appended physically while the + // separate key-index list preserves their logical TAG order, and scans treat those missing from + // the handed-over list as null. + @Test + public void testNewTagColumnsUpdateSortIndicesWhenWritten() { + MeasurementSchema tag3 = new MeasurementSchema("tag3", TSDataType.STRING); + List<IMeasurementSchema> initialSchemas = new ArrayList<>(List.of(SCHEMAS[0], SCHEMAS[3])); + List<IMeasurementSchema> expandedSchemas = + new ArrayList<>(List.of(SCHEMAS[1], tag3, SCHEMAS[3])); + AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(initialSchemas, true, 1, true); + + chunk.writeAlignedPoints(1, new Object[] {binary("a"), 10L}, new ArrayList<>(initialSchemas)); + chunk.handoverAlignedTvList(); + chunk.writeAlignedPoints( + 2, + new Object[] {binary("b"), binary("c"), 20L}, + expandedSchemas, + new TsTableColumnCategory[] { + TsTableColumnCategory.TAG, TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD + }); + Assert.assertEquals(2, chunk.rowCount()); + Assert.assertEquals(3, chunk.getSortColumnCount()); + Assert.assertTrue(chunk.getWorkingTVList() instanceof ColumnSortedAlignedTVList); + Assert.assertEquals( + List.of("tag1", "field1", "tag2", "tag3"), + chunk.getSchemaList().stream() + .map(IMeasurementSchema::getMeasurementName) + .collect(Collectors.toList())); + AlignedTVList tvList = chunk.getWorkingTVList(); + Assert.assertEquals(20L, tvList.getLongByValueIndex(tvList.getValueIndex(0), 1)); + Assert.assertEquals(binary("b"), tvList.getBinaryByValueIndex(tvList.getValueIndex(0), 2)); + Assert.assertEquals(binary("c"), tvList.getBinaryByValueIndex(tvList.getValueIndex(0), 3)); + + List<BitMap> bitMaps = new ArrayList<>(); + Assert.assertArrayEquals( + new long[] {1, 2}, chunk.getAnySatisfiedTimestamp(null, bitMaps, true, null)); + Assert.assertTrue(bitMaps.get(0).isMarked(2)); + Assert.assertTrue(bitMaps.get(0).isMarked(3)); + Assert.assertTrue(bitMaps.get(1).isMarked(0)); + } + + // After handover, a new TAG must only extend the working list. The historical list retains its + // original column layout and compares the later TAG as null. + @Test + public void testNewTagColumnIsAppendedWithoutChangingHistoricalLists() { + MeasurementSchema tag3 = new MeasurementSchema("tag3", TSDataType.STRING); + List<IMeasurementSchema> initialSchemas = new ArrayList<>(List.of(SCHEMAS[0], SCHEMAS[3])); + List<IMeasurementSchema> expandedSchemas = + new ArrayList<>(List.of(SCHEMAS[0], SCHEMAS[1], tag3, SCHEMAS[3])); + TsTableColumnCategory[] expandedCategories = { + TsTableColumnCategory.TAG, + TsTableColumnCategory.TAG, + TsTableColumnCategory.TAG, + TsTableColumnCategory.FIELD + }; + AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(initialSchemas, true, 1, true); + + chunk.writeAlignedPoints( + 1, + new Object[] {binary("a"), 10L}, + new ArrayList<>(initialSchemas), + new TsTableColumnCategory[] {TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD}); + chunk.handoverAlignedTvList(); + chunk.writeAlignedPoints( + 1, + new Object[] {binary("a"), binary("b"), binary("c"), 20L}, + expandedSchemas, + expandedCategories); + chunk.writeAlignedPoints( + 1, new Object[] {binary("a"), null, null, 30L}, expandedSchemas, expandedCategories); + Assert.assertEquals(1, chunk.getSortedList().size()); + AlignedTVList historicalList = chunk.getSortedList().get(0); + AlignedTVList currentList = chunk.getWorkingTVList(); + Assert.assertEquals(2, historicalList.getTsDataTypes().size()); + Assert.assertEquals(binary("a"), historicalList.getBinaryByValueIndex(0, 0)); + Assert.assertEquals(10L, historicalList.getLongByValueIndex(0, 1)); + Assert.assertEquals(1, ((ColumnSortedAlignedTVList) historicalList).getSortColumnCount()); + + Assert.assertEquals(4, currentList.getTsDataTypes().size()); + Assert.assertEquals( + List.of(0, 2, 3), ((ColumnSortedAlignedTVList) currentList).getSortColumnIndices()); + Assert.assertEquals(binary("b"), currentList.getBinaryByValueIndex(0, 2)); + Assert.assertEquals(binary("c"), currentList.getBinaryByValueIndex(0, 3)); + Assert.assertEquals(20L, currentList.getLongByValueIndex(0, 1)); + currentList.sort(); + Assert.assertEquals(0, historicalList.compareRowKey(0, currentList, 0)); + Assert.assertEquals(0, currentList.compareRowKey(0, historicalList, 0)); + Assert.assertTrue(historicalList.compareRowKey(0, currentList, 1) < 0); + Assert.assertTrue(currentList.compareRowKey(1, historicalList, 0) > 0); + } + + // The next insert intentionally omits the existing TAG. It must still append the new TAG after + // the existing FIELD and register the actual physical position as a row-key column. + @Test + public void testFlatMemTableAppendsNewTagAfterExistingField() { + RelationalInsertRowNode initialNode = + new RelationalInsertRowNode( + new PlanNodeId("initial"), + new PartialPath(TABLE_NAME, false), + true, + new String[] {"tag1", "field1"}, + new TSDataType[] {TSDataType.STRING, TSDataType.INT64}, + new MeasurementSchema[] {SCHEMAS[0], SCHEMAS[3]}, + 1, + new Object[] {binary("a"), 10L}, + false, + new TsTableColumnCategory[] {TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD}); + initialNode.setFlatDevice(true); + + MeasurementSchema tag3 = new MeasurementSchema("tag3", TSDataType.STRING); + RelationalInsertRowNode expandedNode = + new RelationalInsertRowNode( + new PlanNodeId("expanded"), + new PartialPath(TABLE_NAME, false), + true, + new String[] {"tag3", "field1"}, + new TSDataType[] {TSDataType.STRING, TSDataType.INT64}, + new MeasurementSchema[] {tag3, SCHEMAS[3]}, + 2, + new Object[] {binary("c"), 20L}, + false, + new TsTableColumnCategory[] {TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD}); + expandedNode.setFlatDevice(true); + + PrimitiveMemTable memTable = new PrimitiveMemTable("db", "0"); + memTable.insertAlignedRow(initialNode); + memTable.insertAlignedRow(expandedNode); + + AlignedWritableMemChunk chunk = + ((AlignedWritableMemChunkGroup) memTable.getMemTableMap().values().iterator().next()) + .getAlignedMemChunk(); + Assert.assertEquals(2, chunk.getSortColumnCount()); + AlignedTVList tvList = chunk.getWorkingTVList(); + Assert.assertEquals(List.of(0, 2), ((ColumnSortedAlignedTVList) tvList).getSortColumnIndices()); + Assert.assertEquals( + List.of("tag1", "field1", "tag3"), + chunk.getSchemaList().stream() + .map(IMeasurementSchema::getMeasurementName) + .collect(Collectors.toList())); + Assert.assertEquals(20L, tvList.getLongByValueIndex(tvList.getValueIndex(1), 1)); + Assert.assertEquals(binary("c"), tvList.getBinaryByValueIndex(tvList.getValueIndex(1), 2)); + } + + // A failed TAG leaves a null schema inside the logical TAG prefix. The remaining TAG is compacted + // to physical column zero and must still be the initial key column. + @Test + public void testFirstFlatRowCompactsFailedLeadingTag() { + RelationalInsertRowNode first = newRow(binary("failed"), binary("z"), 1L); + RelationalInsertRowNode second = newRow(binary("failed"), binary("a"), 100L); + first.markFailedMeasurement(0); + second.markFailedMeasurement(0); + PrimitiveMemTable memTable = new PrimitiveMemTable("db", "0"); + memTable.insertAlignedRow(first); + memTable.insertAlignedRow(second); + + AlignedWritableMemChunk chunk = getOnlyAlignedMemChunk(memTable); + Assert.assertEquals( + List.of("tag2", "field1"), + chunk.getSchemaList().stream() + .map(IMeasurementSchema::getMeasurementName) + .collect(Collectors.toList())); + Assert.assertEquals( + List.of(0), ((ColumnSortedAlignedTVList) chunk.getWorkingTVList()).getSortColumnIndices()); + chunk.getWorkingTVList().sort(); + Assert.assertEquals( + binary("a"), + chunk + .getWorkingTVList() + .getBinaryByValueIndex(chunk.getWorkingTVList().getValueIndex(0), 0)); + Assert.assertEquals( + 100L, + chunk.getWorkingTVList().getLongByValueIndex(chunk.getWorkingTVList().getValueIndex(0), 1)); + Assert.assertEquals( + binary("z"), + chunk + .getWorkingTVList() + .getBinaryByValueIndex(chunk.getWorkingTVList().getValueIndex(1), 0)); + Assert.assertEquals( + 1L, + chunk.getWorkingTVList().getLongByValueIndex(chunk.getWorkingTVList().getValueIndex(1), 1)); + } + + // The tablet path must compact the same failed leading TAG to the remaining physical prefix. + @Test + public void testFirstFlatTabletCompactsFailedLeadingTag() throws Exception { + RelationalInsertTabletNode node = newTablet(); + node.markFailedMeasurement(0); + PrimitiveMemTable memTable = new PrimitiveMemTable("db", "0"); + memTable.insertAlignedTablet(node, 0, node.getRowCount(), null); + + AlignedWritableMemChunk chunk = getOnlyAlignedMemChunk(memTable); + Assert.assertEquals( + List.of("tag2", "field1"), + chunk.getSchemaList().stream() + .map(IMeasurementSchema::getMeasurementName) + .collect(Collectors.toList())); + Assert.assertEquals( + List.of(0), ((ColumnSortedAlignedTVList) chunk.getWorkingTVList()).getSortColumnIndices()); + chunk.getWorkingTVList().sort(); + Assert.assertEquals( + binary("x"), + chunk + .getWorkingTVList() + .getBinaryByValueIndex(chunk.getWorkingTVList().getValueIndex(0), 0)); + Assert.assertEquals( + 11L, + chunk.getWorkingTVList().getLongByValueIndex(chunk.getWorkingTVList().getValueIndex(0), 1)); + Assert.assertEquals( + binary("z"), + chunk + .getWorkingTVList() + .getBinaryByValueIndex(chunk.getWorkingTVList().getValueIndex(1), 0)); + Assert.assertEquals( + 10L, + chunk.getWorkingTVList().getLongByValueIndex(chunk.getWorkingTVList().getValueIndex(1), 1)); + Assert.assertEquals( + 20L, + chunk.getWorkingTVList().getLongByValueIndex(chunk.getWorkingTVList().getValueIndex(2), 1)); + } + + // Covers WAL persistence of the TVList subtype and configured sort columns. Recovery must keep + // using compound ordering, including when a fresh working list is created after handover. + @Test + public void testColumnSortedMemChunkWalRoundTrip() throws IOException { + List<IMeasurementSchema> schemas = new ArrayList<>(List.of(SCHEMAS[0], SCHEMAS[1], SCHEMAS[3])); + AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(schemas, true, 2, true); + chunk.writeAlignedPoints( + 1, new Object[] {binary("b"), binary("x"), 30L}, new ArrayList<>(schemas)); + + ByteBuffer buffer = ByteBuffer.allocate(chunk.serializedSize()); + chunk.serializeToWAL(new WALByteBufferForTest(buffer)); + Assert.assertFalse(buffer.hasRemaining()); + + AlignedWritableMemChunk recovered = + AlignedWritableMemChunk.deserialize( + new DataInputStream(new ByteArrayInputStream(buffer.array())), true); + Assert.assertTrue(recovered.getWorkingTVList() instanceof ColumnSortedAlignedTVList); + Assert.assertEquals( + 2, ((ColumnSortedAlignedTVList) recovered.getWorkingTVList()).getSortColumnCount()); + + recovered.handoverAlignedTvList(); + Assert.assertTrue(recovered.getWorkingTVList() instanceof ColumnSortedAlignedTVList); + } + + // Covers the direct MemChunk flush path used when TVList handover is disabled. Equal timestamps + // with different tags must both reach the time chunk, while an exact time-plus-tag duplicate is + // merged using the latest non-null field value. + @Test + public void testColumnSortedMemChunkFlushUsesCompoundRowKey() { + List<IMeasurementSchema> schemas = new ArrayList<>(List.of(SCHEMAS[0], SCHEMAS[3])); + AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(schemas, true, 1, true); + chunk.putAlignedRow(1, new Object[] {binary("b"), 20L}); + chunk.putAlignedRow(1, new Object[] {binary("a"), 10L}); + chunk.putAlignedRow(1, new Object[] {binary("b"), null}); + chunk.sortTvListForFlush(); + BlockingQueue<Object> ioTaskQueue = new LinkedBlockingQueue<>(); + + chunk.encodeWorkingAlignedTVList(ioTaskQueue, 100, 100); + + Assert.assertEquals(1, ioTaskQueue.size()); + AlignedChunkWriterImpl writer = (AlignedChunkWriterImpl) ioTaskQueue.remove(); + Assert.assertEquals(2, writer.getTimeChunkWriter().getStatistics().getCount()); + } + + private static RelationalInsertRowNode newRow(Binary tag1, Binary tag2, long fieldValue) { + RelationalInsertRowNode node = + new RelationalInsertRowNode( + new PlanNodeId("row"), + new PartialPath(TABLE_NAME, false), + true, + MEASUREMENTS.clone(), + DATA_TYPES.clone(), + SCHEMAS.clone(), + 1, + new Object[] {tag1, tag2, binary("attribute"), fieldValue}, + false, + CATEGORIES.clone()); + node.setFlatDevice(true); + return node; + } + + private static RelationalInsertTabletNode newTablet() { + RelationalInsertTabletNode node = + new RelationalInsertTabletNode( + new PlanNodeId("tablet"), + new PartialPath(TABLE_NAME, false), + true, + MEASUREMENTS.clone(), + DATA_TYPES.clone(), + SCHEMAS.clone(), + new long[] {2, 1, 1}, + new BitMap[MEASUREMENTS.length], + new Object[] { + new Binary[] {binary("b"), binary("b"), binary("a")}, + new Binary[] {binary("x"), binary("x"), binary("z")}, + new Binary[] {binary("v1"), binary("v2"), binary("v3")}, + new long[] {20, 11, 10} + }, + 3, + CATEGORIES.clone()); + node.setFlatDevice(true); + return node; + } + + private static AlignedWritableMemChunk getOnlyAlignedMemChunk(PrimitiveMemTable memTable) { + return ((AlignedWritableMemChunkGroup) memTable.getMemTableMap().values().iterator().next()) + .getAlignedMemChunk(); + } + + private static long[] payloads(AlignedTVList tvList) { + long[] result = new long[tvList.rowCount()]; + for (int i = 0; i < result.length; i++) { + result[i] = tvList.getLongByValueIndex(tvList.getValueIndex(i), 2); + } + return result; + } + + private static void assertNoSortColumns(PrimitiveMemTable memTable) { + Assert.assertFalse(memTable.getMemTableMap().isEmpty()); + assertNoSortColumns( + (AlignedWritableMemChunkGroup) memTable.getMemTableMap().values().iterator().next()); + } + + private static void assertNoSortColumns(AlignedWritableMemChunkGroup group) { + AlignedWritableMemChunk chunk = group.getAlignedMemChunk(); + Assert.assertEquals(0, chunk.getSortColumnCount()); + Assert.assertFalse(chunk.getWorkingTVList() instanceof ColumnSortedAlignedTVList); + } + + private static Binary binary(String value) { + return value == null ? null : new Binary(value, TSFileConfig.STRING_CHARSET); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVListTest.java new file mode 100644 index 00000000000..4e553186a08 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVListTest.java @@ -0,0 +1,490 @@ +/* + * 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.utils.datastructure; + +import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; +import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.Assert; +import org.junit.Test; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; + +public class ColumnSortedAlignedTVListTest { + + private static final List<TSDataType> COMPOUND_ROW_TYPES = + List.of(TSDataType.STRING, TSDataType.INT64); + private static final List<TSEncoding> COMPOUND_ROW_ENCODINGS = + List.of(TSEncoding.PLAIN, TSEncoding.PLAIN); + + // Covers timestamp-first, multi-column, null-first ordering with payload-bearing rows; equal + // timestamps must follow both configured keys while different timestamps remain time ordered. + @Test + public void testSortByTimeAndSelectedColumns() { + AlignedTVList tvList = + AlignedTVListFactory.createColumnSortedAlignedList( + List.of(TSDataType.INT32, TSDataType.TEXT, TSDataType.INT64), 2); + tvList.putAlignedValue(2, new Object[] {0, binary("z"), 20L}); + tvList.putAlignedValue(1, new Object[] {2, binary("b"), 12L}); + tvList.putAlignedValue(1, new Object[] {1, binary("z"), 11L}); + tvList.putAlignedValue(1, new Object[] {1, binary("a"), 10L}); + tvList.putAlignedValue(1, new Object[] {null, binary("z"), 9L}); + + tvList.sort(); + + Assert.assertArrayEquals(new long[] {1, 1, 1, 1, 2}, times(tvList)); + Assert.assertArrayEquals(new long[] {9, 10, 11, 12, 20}, payloads(tvList)); + } + + // Covers every scalar storage representation accepted as a sort key; each larger value is + // inserted first at the same timestamp and must move behind the smaller value after sorting. + @Test + public void testSupportedSortColumnTypes() { + assertTypeOrdering(TSDataType.BOOLEAN, true, false); + assertTypeOrdering(TSDataType.INT32, 2, 1); + assertTypeOrdering(TSDataType.DATE, 2, 1); + assertTypeOrdering(TSDataType.INT64, 2L, 1L); + assertTypeOrdering(TSDataType.TIMESTAMP, 2L, 1L); + assertTypeOrdering(TSDataType.FLOAT, 2.0F, 1.0F); + assertTypeOrdering(TSDataType.DOUBLE, 2.0D, 1.0D); + assertTypeOrdering(TSDataType.TEXT, binary("z"), binary("a")); + assertTypeOrdering(TSDataType.BLOB, binary("z"), binary("a")); + assertTypeOrdering(TSDataType.STRING, binary("z"), binary("a")); + assertTypeOrdering(TSDataType.OBJECT, binary("z"), binary("a")); + } + + // Covers incremental row insertion at the equal-time boundary; a descending selected value + // must invalidate the compound sorted flag even though timestamps alone remain ordered. + @Test + public void testRowInsertionUpdatesCompoundSortedState() { + AlignedTVList tvList = + AlignedTVListFactory.createColumnSortedAlignedList( + List.of(TSDataType.INT32, TSDataType.INT64), 1); + tvList.putAlignedValue(1, new Object[] {2, 20L}); + tvList.putAlignedValue(1, new Object[] {1, 10L}); + + Assert.assertFalse(tvList.isSorted()); + Assert.assertEquals(1, tvList.seqRowCount()); + tvList.sort(); + Assert.assertTrue(tvList.isSorted()); + Assert.assertArrayEquals(new long[] {10, 20}, payloads(tvList)); + } + + // Covers batch insertion with an internally descending equal-time key range; the batch must be + // reported unsorted and sorting must restore ascending selected-column order. + @Test + public void testBatchInsertionUpdatesCompoundSortedState() { + AlignedTVList tvList = + AlignedTVListFactory.createColumnSortedAlignedList( + List.of(TSDataType.INT32, TSDataType.INT64), 1); + tvList.putAlignedValues( + new long[] {1, 1}, new Object[] {new int[] {2, 1}, new long[] {20, 10}}, null, 0, 2, null); + + Assert.assertFalse(tvList.isSorted()); + Assert.assertEquals(1, tvList.seqRowCount()); + tvList.sort(); + Assert.assertArrayEquals(new long[] {10, 20}, payloads(tvList)); + } + + // Covers sorting beyond TimSort's small-array path so compound comparisons are preserved while + // rows and value indices move through merge buffers. + @Test + public void testSortMoreThanSmallArrayThreshold() { + AlignedTVList tvList = + AlignedTVListFactory.createColumnSortedAlignedList( + List.of(TSDataType.INT32, TSDataType.INT64), 1); + for (int value = TimSort.SMALL_ARRAY_LENGTH + 1; value >= 0; value--) { + tvList.putAlignedValue(1, new Object[] {value, (long) value}); + } + + tvList.sort(); + + long[] expected = new long[TimSort.SMALL_ARRAY_LENGTH + 2]; + for (int i = 0; i < expected.length; i++) { + expected[i] = i; + } + Assert.assertArrayEquals(expected, payloads(tvList)); + } + + // Covers releasing a leading key column after sorting; remaining keys can require a different + // row order, so the compound sorted state must be invalidated and recomputed safely. + @Test + public void testSortAfterSortColumnRelease() { + AlignedTVList tvList = + AlignedTVListFactory.createColumnSortedAlignedList( + List.of(TSDataType.INT32, TSDataType.INT32, TSDataType.INT64), 2); + tvList.putAlignedValue(1, new Object[] {1, 2, 20L}); + tvList.putAlignedValue(1, new Object[] {2, 1, 10L}); + Assert.assertTrue(tvList.isSorted()); + + tvList.releaseNonQueryColumns(Set.of(1, 2)); + + Assert.assertFalse(tvList.isSorted()); + Assert.assertEquals(0, tvList.seqRowCount()); + tvList.sort(); + Assert.assertArrayEquals(new long[] {10, 20}, payloads(tvList)); + } + + // Covers both deep and flush-sort clone paths for an unsorted compound list; each clone must + // retain the subtype, configured key columns, and the need to sort before producing key order. + @Test + public void testClonePreservesColumnSortConfiguration() { + AlignedTVList tvList = + AlignedTVListFactory.createColumnSortedAlignedList( + List.of(TSDataType.INT32, TSDataType.INT64), 1); + tvList.putAlignedValue(1, new Object[] {2, 20L}); + tvList.putAlignedValue(1, new Object[] {1, 10L}); + + AlignedTVList clone = tvList.clone(); + AlignedTVList flushClone = tvList.cloneForFlushSort(); + + assertPreservedClone(clone); + assertPreservedClone(flushClone); + } + + // The subtype marker identifies column-sorted lists without overloading the data-type count. + @Test + public void testWalMarkerAndStandardFormat() throws IOException { + AlignedTVList tvList = newCompoundList(); + tvList.putAlignedValue(1, new Object[] {binary("tag"), 10L}); + + ByteBuffer buffer = ByteBuffer.allocate(tvList.serializedSize()); + tvList.serializeToWAL(new WALByteBufferForTest(buffer)); + Assert.assertEquals(-1, buffer.array()[0]); + + AlignedTVList recovered = deserialize(buffer.array()); + Assert.assertTrue(recovered instanceof ColumnSortedAlignedTVList); + Assert.assertEquals(1, ((ColumnSortedAlignedTVList) recovered).getSortColumnCount()); + + AlignedTVList standardList = AlignedTVListFactory.createAlignedList(COMPOUND_ROW_TYPES); + standardList.putAlignedValue(1, new Object[] {binary("tag"), 10L}); + ByteBuffer standardBuffer = ByteBuffer.allocate(standardList.serializedSize() + Integer.BYTES); + standardList.serializeToWAL(new WALByteBufferForTest(standardBuffer)); + standardBuffer.putInt(123); + DataInputStream standardStream = + new DataInputStream(new ByteArrayInputStream(standardBuffer.array())); + Assert.assertFalse( + AlignedTVListFactory.deserialize(standardStream) instanceof ColumnSortedAlignedTVList); + Assert.assertEquals(123, standardStream.readInt()); + } + + // Covers the ownership-transfer clone used by active queries; the replacement working list must + // keep the compound sorting policy after columns have been copied or moved into it. + @Test + public void testPartialClonePreservesColumnSortConfiguration() { + AlignedTVList tvList = + AlignedTVListFactory.createColumnSortedAlignedList( + List.of(TSDataType.INT32, TSDataType.INT64), 1); + tvList.putAlignedValue(1, new Object[] {2, 20L}); + tvList.putAlignedValue(1, new Object[] {1, 10L}); + + AlignedTVList.PartialClonePlan plan = tvList.preparePartialClone(Set.of(1)); + AlignedTVList replacement = plan.getCloneList(); + plan.commit(); + + assertPreservedClone(replacement); + } + + // Covers constructor validation boundaries; negative/out-of-range indices and non-scalar key + // types must fail immediately instead of surfacing later from the sorting comparator. + @Test + public void testRejectInvalidSortColumns() { + Assert.assertThrows( + IndexOutOfBoundsException.class, + () -> AlignedTVListFactory.createColumnSortedAlignedList(List.of(TSDataType.INT32), -1)); + Assert.assertThrows( + IndexOutOfBoundsException.class, + () -> AlignedTVListFactory.createColumnSortedAlignedList(List.of(TSDataType.INT32), 2)); + Assert.assertThrows( + IllegalArgumentException.class, + () -> AlignedTVListFactory.createColumnSortedAlignedList(List.of(TSDataType.VECTOR), 1)); + } + + // Covers materialization with equal timestamps across different tags and an exact duplicate + // compound key. Different tags must remain separate rows, while the exact duplicate keeps the + // latest non-null field value according to existing aligned overwrite semantics. + @Test + public void testBuildTsBlockUsesCompoundRowKey() { + AlignedTVList tvList = createCompoundKeyDuplicateList(); + + TsBlock tsBlock = tvList.buildTsBlock(0, COMPOUND_ROW_ENCODINGS, null, null, false); + + assertCompoundRows(tsBlock); + } + + // Covers both point and batch iteration over the same compound-key duplicate scenario. The two + // APIs use independent duplicate-resolution paths and must expose identical rows. + @Test + public void testIteratorUsesCompoundRowKey() { + AlignedTVList tvList = createCompoundKeyDuplicateList(); + QueryContext queryContext = new QueryContext(false, false); + + AlignedTVList.AlignedTVListIterator pointIterator = + newIterator(tvList, Ordering.ASC, queryContext); + List<TimeValuePair> points = new ArrayList<>(); + while (pointIterator.hasNextTimeValuePair()) { + points.add(pointIterator.nextTimeValuePair()); + } + Assert.assertEquals(2, points.size()); + assertCompoundPoint(points.get(0), "a", 10L); + assertCompoundPoint(points.get(1), "b", 20L); + + AlignedTVList.AlignedTVListIterator batchIterator = + newIterator(tvList, Ordering.ASC, queryContext); + Assert.assertTrue(batchIterator.hasNextBatch()); + assertCompoundRows(batchIterator.nextBatch()); + Assert.assertFalse(batchIterator.hasNextBatch()); + } + + // Covers the single-TVList flush encoder. Its time page must retain different tags at the same + // timestamp and remove only rows whose complete compound key is duplicated. + @Test + public void testEncodeBatchUsesCompoundRowKey() { + AlignedTVList tvList = createCompoundKeyDuplicateList(); + AlignedTVList.AlignedTVListIterator iterator = + newIterator(tvList, Ordering.ASC, new QueryContext(false, false)); + AlignedChunkWriterImpl writer = + new AlignedChunkWriterImpl( + List.of( + new MeasurementSchema("tag", TSDataType.STRING, TSEncoding.PLAIN), + new MeasurementSchema("field", TSDataType.INT64, TSEncoding.PLAIN))); + BatchEncodeInfo encodeInfo = new BatchEncodeInfo(0, 0, 0, 100, 100, 1024); + long[] times = new long[100]; + + Assert.assertTrue(iterator.hasNextBatch()); + iterator.encodeBatch(writer, encodeInfo, times); + writer.write(times, encodeInfo.pointNumInPage, 0); + writer.sealCurrentPage(); + + Assert.assertEquals(2, encodeInfo.pointNumInPage); + Assert.assertEquals(2, writer.getTimeChunkWriter().getStatistics().getCount()); + } + + // Covers MemChunk handover/query snapshots where compound rows are spread across TVLists. The + // merge heap must order and deduplicate by time plus tag instead of collapsing the whole time. + @Test + public void testMergeIteratorUsesCompoundRowKeyAcrossLists() throws IOException { + AlignedTVList older = newCompoundList(); + older.putAlignedValue(1, new Object[] {binary("a"), 10L}); + older.putAlignedValue(1, new Object[] {binary("c"), 30L}); + older.sort(); + AlignedTVList newer = newCompoundList(); + newer.putAlignedValue(1, new Object[] {binary("a"), 11L}); + newer.putAlignedValue(1, new Object[] {binary("b"), 20L}); + newer.sort(); + + MemPointIterator iterator = + MemPointIteratorFactory.create( + COMPOUND_ROW_TYPES, + null, + List.of(older, newer), + false, + 100, + new QueryContext(false, false)); + List<TimeValuePair> points = new ArrayList<>(); + while (iterator.hasNextTimeValuePair()) { + points.add(iterator.nextTimeValuePair()); + } + + Assert.assertEquals(3, points.size()); + assertCompoundPoint(points.get(0), "a", 11L); + assertCompoundPoint(points.get(1), "b", 20L); + assertCompoundPoint(points.get(2), "c", 30L); + } + + // Historical lists retain their original key prefix. Columns added to the current list must + // compare as null on the historical side, while comparison remains antisymmetric. + @Test + public void testCompareRowKeyTreatsMissingHistoricalSortColumnsAsNull() { + List<TSDataType> types = List.of(TSDataType.STRING, TSDataType.STRING, TSDataType.INT64); + ColumnSortedAlignedTVList historical = + (ColumnSortedAlignedTVList) AlignedTVListFactory.createColumnSortedAlignedList(types, 1); + historical.putAlignedValue(1, new Object[] {binary("a"), null, 10L}); + historical.sort(); + + ColumnSortedAlignedTVList current = + (ColumnSortedAlignedTVList) AlignedTVListFactory.createColumnSortedAlignedList(types, 2); + current.putAlignedValue(1, new Object[] {binary("a"), null, 20L}); + current.putAlignedValue(1, new Object[] {binary("a"), binary("b"), 30L}); + current.sort(); + + Assert.assertEquals(1, historical.getSortColumnCount()); + Assert.assertEquals(2, current.getSortColumnCount()); + Assert.assertEquals(0, historical.compareRowKey(0, current, 0)); + Assert.assertEquals(0, current.compareRowKey(0, historical, 0)); + Assert.assertTrue(historical.compareRowKey(0, current, 1) < 0); + Assert.assertTrue(current.compareRowKey(1, historical, 0) > 0); + } + + // Covers a TAG key physically following a FIELD. Sorting must use positions 0 then 2, never + // treat the FIELD at position 1 as part of the compound key. + @Test + public void testNonContiguousSortColumnIndices() { + List<TSDataType> types = + List.of(TSDataType.STRING, TSDataType.INT64, TSDataType.STRING, TSDataType.INT64); + ColumnSortedAlignedTVList tvList = + (ColumnSortedAlignedTVList) + AlignedTVListFactory.createColumnSortedAlignedList(types, List.of(0, 2)); + tvList.putAlignedValue(1, new Object[] {binary("b"), 20L, binary("z"), 200L}); + tvList.putAlignedValue(1, new Object[] {binary("b"), 10L, binary("a"), 100L}); + tvList.putAlignedValue(1, new Object[] {binary("a"), 30L, binary("z"), 300L}); + + tvList.sort(); + + Assert.assertEquals(List.of(0, 2), tvList.getSortColumnIndices()); + Assert.assertArrayEquals(new long[] {300L, 100L, 200L}, values(tvList, 3)); + } + + // Covers clone and WAL persistence of non-contiguous TAG positions so recovery preserves the + // same physical-key mapping as the original in-memory list. + @Test + public void testNonContiguousSortColumnIndicesSurviveClonesAndWal() throws IOException { + List<TSDataType> types = + List.of(TSDataType.STRING, TSDataType.INT64, TSDataType.STRING, TSDataType.INT64); + ColumnSortedAlignedTVList tvList = + (ColumnSortedAlignedTVList) + AlignedTVListFactory.createColumnSortedAlignedList(types, List.of(0, 2)); + tvList.putAlignedValue(1, new Object[] {binary("a"), 10L, binary("b"), 100L}); + + ColumnSortedAlignedTVList clone = (ColumnSortedAlignedTVList) tvList.clone(); + ColumnSortedAlignedTVList flushClone = (ColumnSortedAlignedTVList) tvList.cloneForFlushSort(); + Assert.assertEquals(List.of(0, 2), clone.getSortColumnIndices()); + Assert.assertEquals(List.of(0, 2), flushClone.getSortColumnIndices()); + + ByteBuffer buffer = ByteBuffer.allocate(tvList.serializedSize()); + tvList.serializeToWAL(new WALByteBufferForTest(buffer)); + ColumnSortedAlignedTVList recovered = (ColumnSortedAlignedTVList) deserialize(buffer.array()); + Assert.assertEquals(List.of(0, 2), recovered.getSortColumnIndices()); + } + + private static void assertTypeOrdering(TSDataType dataType, Object larger, Object smaller) { + AlignedTVList tvList = + AlignedTVListFactory.createColumnSortedAlignedList(List.of(dataType, TSDataType.INT64), 1); + tvList.putAlignedValue(1, new Object[] {larger, 2L}); + tvList.putAlignedValue(1, new Object[] {smaller, 1L}); + + tvList.sort(); + + Assert.assertArrayEquals(new long[] {1, 2}, payloads(tvList)); + } + + private static void assertPreservedClone(AlignedTVList clone) { + Assert.assertTrue(clone instanceof ColumnSortedAlignedTVList); + Assert.assertEquals(1, ((ColumnSortedAlignedTVList) clone).getSortColumnCount()); + Assert.assertFalse(clone.isSorted()); + clone.sort(); + Assert.assertArrayEquals(new long[] {10, 20}, payloads(clone)); + } + + private static AlignedTVList createCompoundKeyDuplicateList() { + AlignedTVList tvList = newCompoundList(); + tvList.putAlignedValue(1, new Object[] {binary("b"), 20L}); + tvList.putAlignedValue(1, new Object[] {binary("a"), 10L}); + tvList.putAlignedValue(1, new Object[] {binary("b"), null}); + tvList.sort(); + return tvList; + } + + private static AlignedTVList newCompoundList() { + return AlignedTVListFactory.createColumnSortedAlignedList(COMPOUND_ROW_TYPES, 1); + } + + private static AlignedTVList.AlignedTVListIterator newIterator( + AlignedTVList tvList, Ordering ordering, QueryContext queryContext) { + return tvList.iterator( + ordering, + tvList.rowCount(), + null, + COMPOUND_ROW_TYPES, + null, + null, + null, + 0, + COMPOUND_ROW_ENCODINGS, + false, + 100, + queryContext); + } + + private static void assertCompoundRows(TsBlock tsBlock) { + Assert.assertEquals(2, tsBlock.getPositionCount()); + Assert.assertEquals(1, tsBlock.getTimeByIndex(0)); + Assert.assertEquals(1, tsBlock.getTimeByIndex(1)); + Assert.assertEquals(binary("a"), tsBlock.getColumn(0).getBinary(0)); + Assert.assertEquals(binary("b"), tsBlock.getColumn(0).getBinary(1)); + Assert.assertEquals(10, tsBlock.getColumn(1).getLong(0)); + Assert.assertEquals(20, tsBlock.getColumn(1).getLong(1)); + } + + private static void assertCompoundPoint( + TimeValuePair point, String expectedTag, long expectedField) { + Assert.assertEquals(1, point.getTimestamp()); + Assert.assertEquals(binary(expectedTag), point.getValues()[0]); + Assert.assertEquals(expectedField, point.getValues()[1]); + } + + private static long[] times(AlignedTVList tvList) { + long[] result = new long[tvList.rowCount()]; + for (int i = 0; i < result.length; i++) { + result[i] = tvList.getTime(i); + } + return result; + } + + private static long[] payloads(AlignedTVList tvList) { + long[] result = new long[tvList.rowCount()]; + int payloadColumnIndex = tvList.getTsDataTypes().size() - 1; + for (int i = 0; i < result.length; i++) { + result[i] = tvList.getLongByValueIndex(tvList.getValueIndex(i), payloadColumnIndex); + } + return result; + } + + private static long[] values(AlignedTVList tvList, int columnIndex) { + long[] result = new long[tvList.rowCount()]; + for (int i = 0; i < result.length; i++) { + result[i] = tvList.getLongByValueIndex(tvList.getValueIndex(i), columnIndex); + } + return result; + } + + private static Binary binary(String value) { + return new Binary(value, TSFileConfig.STRING_CHARSET); + } + + private static AlignedTVList deserialize(byte[] bytes) throws IOException { + return AlignedTVListFactory.deserialize(new DataInputStream(new ByteArrayInputStream(bytes))); + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/TsTablePropertyTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/TsTablePropertyTest.java new file mode 100644 index 00000000000..cbb958c7f0f --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/TsTablePropertyTest.java @@ -0,0 +1,52 @@ +/* + * 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.commons.schema.table; + +import org.junit.Assert; +import org.junit.Test; + +public class TsTablePropertyTest { + + // Covers the compatibility default: tables without an explicit flat_device property must keep + // the existing per-device storage behavior. + @Test + public void testFlatDeviceDefaultsToFalse() { + Assert.assertFalse(new TsTable("table1").isFlatDevice()); + } + + // Covers the enabled property value: the canonical true string must activate flat-device mode. + @Test + public void testFlatDeviceCanBeEnabled() { + TsTable table = new TsTable("table1"); + table.addProp(TsTable.FLAT_DEVICE_PROPERTY, Boolean.TRUE.toString()); + + Assert.assertTrue(table.isFlatDevice()); + } + + // Covers an explicitly disabled property so SHOW/ALTER round trips do not confuse presence with + // activation. + @Test + public void testFlatDeviceCanBeExplicitlyDisabled() { + TsTable table = new TsTable("table1"); + table.addProp(TsTable.FLAT_DEVICE_PROPERTY, Boolean.FALSE.toString()); + + Assert.assertFalse(table.isFlatDevice()); + } +} diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/column/TagColumnSchemaTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/column/TagColumnSchemaTest.java new file mode 100644 index 00000000000..e8f37c6e084 --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/schema/table/column/TagColumnSchemaTest.java @@ -0,0 +1,44 @@ +/* + * 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.commons.schema.table.column; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.Assert; +import org.junit.Test; + +import java.util.Map; + +public class TagColumnSchemaTest { + + @Test + public void testMeasurementSchemaUsesDictionaryEncodingFromCreation() { + Map<String, String> props = Map.of("key", "value"); + + MeasurementSchema schema = + new TagColumnSchema("tag", TSDataType.STRING, props).getMeasurementSchema(); + + Assert.assertEquals("tag", schema.getMeasurementName()); + Assert.assertEquals(TSDataType.STRING, schema.getType()); + Assert.assertEquals(TSEncoding.DICTIONARY, schema.getEncodingType()); + Assert.assertEquals(props, schema.getProps()); + } +}
