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 2a3e6b79a8bd4562b45bd364ec3759dd095d2c17 Author: Tian Jiang <[email protected]> AuthorDate: Tue Sep 8 14:41:17 2026 +0800 add tests --- .../relational/it/db/it/IoTDBLoadTsFileIT.java | 98 ++++++++ .../it/insertquery/IoTDBFlatDeviceIT.java | 247 +++++++++++++++++++++ 2 files changed, 345 insertions(+) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTsFileIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTsFileIT.java index 4f33ff23a41..36e1b062cbb 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTsFileIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBLoadTsFileIT.java @@ -26,14 +26,20 @@ 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.common.conf.TSFileConfig; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.Pair; import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.chunk.AlignedChunkWriterImpl; import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.IMeasurementSchema; import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -634,6 +640,98 @@ public class IoTDBLoadTsFileIT { } } + // The source has physical TAG chunks and a binary footer marker while the target is ordinary. + // With size-based conversion disabled, only the source marker can select the correct parser. + @Test + public void testLoadFlatDeviceFileIntoNormalTable() throws Exception { + assertLoadFlatDeviceFile(false); + } + + // Loading a flat-device source into a flat-device target must preserve TAG values (including + // null and empty values) and fields both before and after a subsequent FLUSH. + @Test + public void testLoadFlatDeviceFileIntoFlatTable() throws Exception { + assertLoadFlatDeviceFile(true); + } + + private void assertLoadFlatDeviceFile(boolean flatTarget) throws Exception { + final String database = "flat_file_load"; + final String table = "sensor"; + final List<IMeasurementSchema> schemas = + Arrays.asList( + new MeasurementSchema("tag_a", TSDataType.STRING, TSEncoding.DICTIONARY), + new MeasurementSchema("tag_b", TSDataType.STRING, TSEncoding.DICTIONARY), + new MeasurementSchema("value_1", TSDataType.INT64, TSEncoding.PLAIN)); + final String[][] tags = {{"a", "1"}, {"b", "2"}, {"a", null}, {"a", ""}}; + final File file = new File(tmpDir, "physical-flat-device.tsfile"); + try (TsFileIOWriter writer = new TsFileIOWriter(file)) { + writer + .getSchema() + .registerTableSchema( + new TableSchema( + table, + schemas, + Arrays.asList(ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.FIELD))); + // Encode the on-disk contract directly so this test detects incompatible writer changes. + writer.addTsFileProperty("flat_device." + table, new byte[] {1}); + writer.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {table})); + final AlignedChunkWriterImpl chunks = new AlignedChunkWriterImpl(schemas); + for (int row = 0; row < tags.length; row++) { + final long time = row + 1; + chunks.getTimeChunkWriter().write(time); + for (int tag = 0; tag < 2; tag++) { + final String value = tags[row][tag]; + chunks + .getValueChunkWriterByIndex(tag) + .write( + time, + value == null + ? Binary.EMPTY_VALUE + : new Binary(value, TSFileConfig.STRING_CHARSET), + value == null); + } + chunks.getValueChunkWriterByIndex(2).write(time, time * 10, false); + } + chunks.writeToFileWriter(writer); + writer.endChunkGroup(); + writer.endFile(); + } + + 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 sensor (tag_a STRING TAG, tag_b STRING TAG, value_1 INT64 FIELD)" + + " WITH (flat_device=" + + flatTarget + + ")"); + statement.execute( + String.format( + "LOAD '%s' WITH ('database'='%s', 'tablet-conversion-threshold'='-1')", + file.getAbsolutePath(), database)); + for (int phase = 0; phase < 2; phase++) { + try (ResultSet resultSet = + statement.executeQuery( + "SELECT time, tag_a, tag_b, value_1 FROM sensor ORDER BY time")) { + int row = 0; + while (resultSet.next()) { + Assert.assertTrue("Unexpected extra row", row < tags.length); + Assert.assertEquals(row + 1, resultSet.getLong(1)); + Assert.assertEquals(tags[row][0], resultSet.getString(2)); + Assert.assertEquals(tags[row][1], resultSet.getString(3)); + Assert.assertEquals((row + 1) * 10, resultSet.getLong(4)); + row++; + } + Assert.assertEquals(tags.length, row); + } + if (phase == 0) { + statement.execute("FLUSH"); + } + } + } + } + private static void addFlatDeviceLoadRow( final Tablet tablet, final long time, 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 index 9df4d54c8aa..544664292a8 100644 --- 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 @@ -21,6 +21,8 @@ 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.env.cluster.env.AbstractEnv; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.TableClusterIT; import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; @@ -28,6 +30,7 @@ import org.apache.iotdb.itbase.env.BaseEnv; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.TsFileSequenceReader; import org.apache.tsfile.read.common.RowRecord; import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.IMeasurementSchema; @@ -40,6 +43,10 @@ import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; @@ -47,7 +54,10 @@ import java.sql.Statement; import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -62,6 +72,11 @@ public class IoTDBFlatDeviceIT { @BeforeClass public static void setUpClass() throws Exception { + EnvFactory.getEnv().getConfig().getCommonConfig().setWalMode("SYNC"); + // Keep sealed files stable while checking the persisted footer. + EnvFactory.getEnv().getConfig().getCommonConfig().setEnableSeqSpaceCompaction(false); + EnvFactory.getEnv().getConfig().getCommonConfig().setEnableUnseqSpaceCompaction(false); + EnvFactory.getEnv().getConfig().getCommonConfig().setEnableCrossSpaceCompaction(false); EnvFactory.getEnv().initClusterEnvironment(); } @@ -243,6 +258,238 @@ public class IoTDBFlatDeviceIT { } } + // Null and empty TAGs are distinct compound keys, even at the same timestamp. Rewriting one + // flushed key must update only that row, leaving the other logical devices intact. + @Test + public void testNullAndEmptyTagsAcrossFlush() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, attr, value) VALUES " + + "(10, 'a', null, 'null-tag', 1.0)," + + "(10, 'a', '', 'empty-tag', 2.0)," + + "(10, null, 'b', 'leading-null', 3.0)"); + for (int phase = 0; phase < 2; phase++) { + assertRows( + statement.executeQuery(queryForAttribute("null-tag")), + Arrays.asList("10,a,null,null-tag,1.0")); + assertRows( + statement.executeQuery(queryForAttribute("empty-tag")), + Arrays.asList("10,a,,empty-tag,2.0")); + assertRows( + statement.executeQuery(queryForAttribute("leading-null")), + Arrays.asList("10,null,b,leading-null,3.0")); + assertCount(statement, 3); + statement.execute("FLUSH"); + } + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, value) VALUES (10, 'a', null, 4.0)"); + statement.execute("FLUSH"); + assertRows( + statement.executeQuery(queryForAttribute("null-tag")), + Arrays.asList("10,a,null,null-tag,4.0")); + assertRows( + statement.executeQuery(queryForAttribute("empty-tag")), + Arrays.asList("10,a,,empty-tag,2.0")); + assertCount(statement, 3); + } + assertFooterProperty(TABLE, true); + } + + // Interleave normal and flat table writes in one database. Each table must retain its own + // single-byte layout marker and its own TAG values after sealing the files. + @Test + public void testMixedTableFooterProperties() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "CREATE TABLE normal_sensor (tag_a STRING TAG, tag_b STRING TAG, " + + "attr STRING ATTRIBUTE, value DOUBLE FIELD) WITH (flat_device=false)"); + for (String table : Arrays.asList(TABLE, "normal_sensor")) { + statement.execute( + "INSERT INTO " + + table + + " (time, tag_a, tag_b, attr, value) VALUES " + + "(10, 'a', '1', 'first', 1.0), (10, 'b', '2', 'second', 2.0)"); + } + statement.execute("FLUSH"); + for (String table : Arrays.asList(TABLE, "normal_sensor")) { + assertRows( + statement.executeQuery( + "SELECT time, tag_a, tag_b, attr, value FROM " + + table + + " ORDER BY time, tag_a, tag_b"), + Arrays.asList("10,a,1,first,1.0", "10,b,2,second,2.0")); + } + } + assertFooterProperty(TABLE, true); + assertFooterProperty("normal_sensor", false); + } + + // A sealed flat-device file must retain equal-time compound keys and its footer marker after + // DataNode restart, when all in-memory writer state has been discarded. + @Test + public void testSealedFlatDeviceFileAfterRestart() throws Exception { + assertRecovery(true); + } + + // Kill DataNodes before FLUSH with synchronous WAL enabled. WAL replay must recover both + // equal-time logical devices and write the flat-device marker when the recovered file is sealed. + @Test + public void testFlatDeviceWalRecovery() throws Exception { + assertRecovery(false); + } + + // DELETE runs before FLUSH: compound TAG/time matching, NULL versus empty TAGs, repeated + // deletion and reinsertion must affect only the selected logical rows, including after flush. + @Test + public void testFlatDeviceMemoryDeletionByTags() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, value) VALUES " + + "(9, 'a', 'x', 9), (10, 'a', 'x', 10), (10, 'a', 'y', 11)," + + "(10, 'b', 'x', 12), (10, 'a', null, 13), (10, 'a', '', 14)," + + "(10, null, 'x', 15), (11, 'a', 'x', 16)"); + assertCount(statement, 8); + for (int i = 0; i < 2; i++) { + statement.execute("DELETE FROM sensor WHERE tag_a = 'a' AND tag_b = 'x' AND time = 10"); + statement.execute("DELETE FROM sensor WHERE tag_a = 'a' AND tag_b IS NULL AND time = 10"); + statement.execute("DELETE FROM sensor WHERE tag_a IS NULL AND time = 10"); + assertRows( + statement.executeQuery(query()), + Arrays.asList( + "9,a,x,null,9.0", + "10,a,,null,14.0", + "10,a,y,null,11.0", + "10,b,x,null,12.0", + "11,a,x,null,16.0")); + assertCount(statement, 5); + } + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, value) VALUES (10, 'a', 'x', 100)"); + for (int phase = 0; phase < 2; phase++) { + assertRows( + statement.executeQuery(query()), + Arrays.asList( + "9,a,x,null,9.0", + "10,a,,null,14.0", + "10,a,x,null,100.0", + "10,a,y,null,11.0", + "10,b,x,null,12.0", + "11,a,x,null,16.0")); + assertCount(statement, 6); + if (phase == 0) { + statement.execute("FLUSH"); + } + } + } + } + + // Time-only and whole-table deletion keep their existing path. A fresh write after deleting + // the entire in-memory group must remain queryable both before and after FLUSH. + @Test + public void testFlatDeviceMemoryTimeAndFullDeletion() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, value) VALUES " + + "(9, 'a', 'x', 9), (10, 'a', 'x', 10), (10, 'b', 'x', 11), (11, 'a', 'x', 12)"); + statement.execute("DELETE FROM sensor WHERE time >= 10 AND time <= 10"); + assertRows( + statement.executeQuery(query()), Arrays.asList("9,a,x,null,9.0", "11,a,x,null,12.0")); + statement.execute("DELETE FROM sensor"); + assertCount(statement, 0); + statement.execute("INSERT INTO sensor (time, tag_a, tag_b, value) VALUES (10, 'b', 'x', 20)"); + assertRows(statement.executeQuery(query()), Arrays.asList("10,b,x,null,20.0")); + statement.execute("FLUSH"); + assertRows(statement.executeQuery(query()), Arrays.asList("10,b,x,null,20.0")); + } + } + + // An unflushed TAG deletion must be replayed from WAL after a forced restart without deleting + // another logical device at the same timestamp or resurrecting the deleted row on FLUSH. + @Test + public void testFlatDeviceMemoryDeletionWalRecovery() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, value) VALUES " + + "(10, 'a', 'x', 10), (10, 'b', 'x', 20)"); + statement.execute("DELETE FROM sensor WHERE tag_a = 'a' AND time = 10"); + assertRows(statement.executeQuery(query()), Arrays.asList("10,b,x,null,20.0")); + } + EnvFactory.getEnv().shutdownForciblyAllDataNodes(); + EnvFactory.getEnv().startAllDataNodes(); + ((AbstractEnv) EnvFactory.getEnv()).checkClusterStatusWithoutUnknown(); + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + assertRows(statement.executeQuery(query()), Arrays.asList("10,b,x,null,20.0")); + statement.execute("FLUSH"); + assertRows(statement.executeQuery(query()), Arrays.asList("10,b,x,null,20.0")); + } + } + + private static void assertRecovery(boolean flushBeforeRestart) throws Exception { + final List<String> expected = Arrays.asList("10,a,1,first,1.0", "10,b,2,second,2.0"); + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, attr, value) VALUES " + + "(10, 'a', '1', 'first', 1.0), (10, 'b', '2', 'second', 2.0)"); + assertRows(statement.executeQuery(query()), expected); + if (flushBeforeRestart) { + statement.execute("FLUSH"); + } + } + EnvFactory.getEnv().shutdownForciblyAllDataNodes(); + EnvFactory.getEnv().startAllDataNodes(); + ((AbstractEnv) EnvFactory.getEnv()).checkClusterStatusWithoutUnknown(); + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + assertRows(statement.executeQuery(query()), expected); + statement.execute("FLUSH"); + assertRows(statement.executeQuery(query()), expected); + } + assertFooterProperty(TABLE, true); + } + + private static void assertFooterProperty(String table, boolean flatDevice) throws IOException { + int checkedFiles = 0; + for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) { + final Path dataDirectory = Paths.get(node.getDataNodeDir(), "data"); + if (!Files.exists(dataDirectory)) { + continue; + } + try (Stream<Path> paths = Files.walk(dataDirectory)) { + for (Path file : + paths + .filter(path -> path.toString().endsWith(".tsfile")) + .filter(path -> path.toString().contains(DATABASE)) + .collect(Collectors.toList())) { + try (TsFileSequenceReader reader = new TsFileSequenceReader(file.toString())) { + if (reader.getTableSchemaMap().containsKey(table)) { + assertArrayEquals( + file.toString(), + new byte[] {(byte) (flatDevice ? 1 : 0)}, + reader.getTsFileProperties().get("flat_device." + table)); + checkedFiles++; + } + } + } + } + } + assertTrue("No sealed TsFile found for " + table, checkedFiles > 0); + } + private static String query() { return "SELECT time, tag_a, tag_b, attr, value FROM " + TABLE + " ORDER BY time, tag_a, tag_b"; }
