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 77b68b6ff3f64d22d9cf7fdf18aff1b6cbd51e30 Author: Tian Jiang <[email protected]> AuthorDate: Thu Sep 24 09:31:11 2026 +0800 Add perf test --- .../it/insertquery/IoTDBFlatDeviceIT.java | 99 ++++++++++++++++++++++ 1 file changed, 99 insertions(+) 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 de9d122af4f..170117f2101 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 @@ -18,6 +18,13 @@ package org.apache.iotdb.relational.it.insertquery; +import java.util.Random; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicLong; import org.apache.iotdb.isession.ITableSession; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.it.env.EnvFactory; @@ -39,6 +46,7 @@ import org.junit.After; import org.junit.AfterClass; import org.junit.Before; import org.junit.BeforeClass; +import org.junit.Ignore; import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; @@ -604,6 +612,97 @@ public class IoTDBFlatDeviceIT { } } + @Ignore + @Test + public void testMassiveSeriesWrite() throws SQLException, ExecutionException, InterruptedException { + int concurrency = 10; + int deviceNum = 500000; + int fieldNum = 500; + int devicePerTablet = 100; + int fieldPerTablet = 100; + int loop = 100; + + AtomicLong writtenPointNum = new AtomicLong(); + AtomicLong writtenTabletNum = new AtomicLong(); + int[] writtenTabletNumPerTask = new int[concurrency]; + + List<String> columnNames = new ArrayList<>(1 + fieldNum); + List<TSDataType> dataTypes = new ArrayList<>(1 + fieldNum); + List<ColumnCategory> columnCategories = new ArrayList<>(1 + fieldNum); + columnNames.add("device_id"); + dataTypes.add(TSDataType.STRING); + columnCategories.add(ColumnCategory.TAG); + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + StringBuilder createTableSql = new StringBuilder("CREATE TABLE test ( device_id TAG"); + for (int i = 0; i < fieldNum; i++) { + createTableSql.append(", s").append(i).append(" DOUBLE"); + columnNames.add("s" + i); + dataTypes.add(TSDataType.DOUBLE); + columnCategories.add(ColumnCategory.FIELD); + } + createTableSql.append(") WITH (flat_device=true)"); + statement.execute(createTableSql.toString()); + } + + long startTime = System.currentTimeMillis(); + ExecutorService threadPool = Executors.newCachedThreadPool(); + List<Future<Void>> tasks = new ArrayList<>(concurrency); + for (int i = 0; i < concurrency; i++) { + final int taskNum = i; + Callable<Void> task = () -> { + Random random = new Random(); + try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) { + session.executeNonQueryStatement("USE " + DATABASE); + + for (int l = 0; l < loop; l++) { + for (int deviceStart = taskNum * devicePerTablet; deviceStart < deviceNum; deviceStart += devicePerTablet * concurrency) { + int fieldStart = random.nextInt(1, fieldNum + 1 - fieldPerTablet); + int fieldEnd = Math.min(fieldStart + fieldPerTablet, fieldNum); + + List<String> subColumnNames = new ArrayList<>(columnNames.subList(fieldStart, fieldEnd)); + subColumnNames.add(0, "device_id"); + List<TSDataType> subDataTypes = new ArrayList<>(dataTypes.subList(fieldStart, fieldEnd)); + subDataTypes.add(0, TSDataType.STRING); + ArrayList<ColumnCategory> subColumnCategories = new ArrayList<>( + columnCategories.subList(fieldStart, fieldEnd)); + subColumnCategories.add(0, ColumnCategory.TAG); + + Tablet tablet = new Tablet("test", subColumnNames, subDataTypes, subColumnCategories, devicePerTablet); + + for (int batchDeviceIndex = 0; batchDeviceIndex < devicePerTablet; batchDeviceIndex++) { + String deviceId = "d" + (deviceStart + batchDeviceIndex); + tablet.addTimestamp(batchDeviceIndex, l); + tablet.addValue(batchDeviceIndex, 0, deviceId); + + for (int f = 0; f < fieldEnd - fieldStart; f++) { + tablet.addValue(batchDeviceIndex, f + 1, (long) l * 1.0); + } + } + + session.insert(tablet); + writtenPointNum.addAndGet((long) (fieldEnd - fieldStart) * devicePerTablet); + writtenTabletNum.addAndGet(1); + writtenTabletNumPerTask[taskNum] ++; + } + long elapsedTime = System.currentTimeMillis() - startTime; + System.out.printf("client %d loop %d finished after %dms, speed %fpt/s%n", taskNum, l, elapsedTime, writtenPointNum.get() / (elapsedTime / 1000.0)); + System.out.printf("%d points written%n", writtenPointNum.get()); + System.out.printf("%d tablets written, %s%n", writtenTabletNum.get(), Arrays.toString(writtenTabletNumPerTask)); + } + } + + return null; + }; + tasks.add(threadPool.submit(task)); + } + + for (Future<Void> task : tasks) { + task.get(); + } + } + private static void assertDeletionRows(Statement statement, List<String> expected) throws SQLException { try (ResultSet resultSet = statement.executeQuery(query())) {
