This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch insert_numeric_row in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 408fdc8f70fe06e73cd7b44027035f4aa42516ac Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 23 12:21:32 2026 +0800 optimize --- .../it/session/IoTDBTableModelSessionIT.java | 72 +++++--- .../dataregion/memtable/AbstractMemTable.java | 29 ++- .../memtable/AbstractMemTableNumericRowTest.java | 195 +++++++++++++++++++++ 3 files changed, 268 insertions(+), 28 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBTableModelSessionIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBTableModelSessionIT.java index df4669db155..3b0313ec006 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBTableModelSessionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBTableModelSessionIT.java @@ -435,7 +435,8 @@ public class IoTDBTableModelSessionIT { "Performance benchmark disabled; set iotdb.numeric.row.insert.perf.enabled=true to run", Boolean.getBoolean("iotdb.numeric.row.insert.perf.enabled")); final int concurrency = 16; - // Compare sustained single-row writes after warming both paths; verify all acknowledged rows. + // Compare sustained single-row writes after warming all three paths; verify all acknowledged + // rows. final int warmupRounds = 2; final int measuredRounds = 6; final int fieldCount = 99; @@ -452,6 +453,7 @@ public class IoTDBTableModelSessionIT { session.executeNonQueryStatement("USE " + databasePrefix + i); session.executeNonQueryStatement(ddl.toString()); session.executeNonQueryStatement(ddl.toString().replace("prepared_perf", "tablet_perf")); + session.executeNonQueryStatement(ddl.toString().replace("prepared_perf", "row_perf")); } } @@ -469,10 +471,10 @@ public class IoTDBTableModelSessionIT { ITableSession[] sessions = new ITableSession[concurrency]; NumericRowPreparedStatement[] prepared = new NumericRowPreparedStatement[concurrency]; Tablet[] tablets = new Tablet[concurrency]; + NumericRow[] rows = new NumericRow[concurrency]; String[][] tags = new String[concurrency][]; long[][] numericValues = new long[concurrency][fieldCount]; - long[] preparedRows = new long[concurrency]; - long[] tabletRows = new long[concurrency]; + long[][] pathRows = new long[3][concurrency]; ExecutorService executor = Executors.newFixedThreadPool(concurrency); try { // Connections, USE, prepare and reusable input allocation are outside every timed round. @@ -484,16 +486,18 @@ public class IoTDBTableModelSessionIT { tablets[i] = new Tablet("tablet_perf", names, types, categories, 1); tags[i] = new String[] {"device-" + i}; Arrays.fill(numericValues[i], Double.doubleToRawLongBits(1.0)); + rows[i] = + new NumericRow("row_perf", 0, names, types, categories, numericValues[i], tags[i]); } IntConsumer preparedWrite = threadId -> { try { sessions[threadId].executePreparedInsertNumericRow( prepared[threadId], - preparedRows[threadId], + pathRows[0][threadId], numericValues[threadId], tags[threadId]); - preparedRows[threadId]++; + pathRows[0][threadId]++; } catch (Exception e) { throw new RuntimeException(e); } @@ -502,36 +506,56 @@ public class IoTDBTableModelSessionIT { threadId -> { try { Tablet tablet = tablets[threadId]; - tablet.addTimestamp(0, tabletRows[threadId]); + tablet.addTimestamp(0, pathRows[2][threadId]); tablet.addValue("tag", 0, tags[threadId][0]); for (int i = 0; i < fieldCount; i++) { tablet.addValue(names.get(i + 1), 0, 1.0D); } sessions[threadId].insert(tablet); tablet.reset(); - tabletRows[threadId]++; + pathRows[2][threadId]++; } catch (Exception e) { throw new RuntimeException(e); } }; - double[][] rates = new double[2][measuredRounds]; - long[] totalRows = new long[2]; - long[] totalNanos = new long[2]; + IntConsumer rowWrite = + threadId -> { + try { + rows[threadId] = + new NumericRow( + "row_perf", + pathRows[1][threadId], + names, + types, + categories, + numericValues[threadId], + tags[threadId]); + sessions[threadId].insert(rows[threadId]); + pathRows[1][threadId]++; + } catch (Exception e) { + throw new RuntimeException(e); + } + }; + IntConsumer[] paths = {preparedWrite, rowWrite, tabletWrite}; + String[] pathNames = {"prepared numeric row", "insert NumericRow", "insertTablet batch=1"}; + System.out.printf("configuration: concurrency=%d, tabletBatch=1%n", concurrency); + double[][] rates = new double[paths.length][measuredRounds]; + long[] totalRows = new long[paths.length]; + long[] totalNanos = new long[paths.length]; for (int round = 0; round < warmupRounds + measuredRounds; round++) { boolean warmup = round < warmupRounds; int seconds = warmup ? 3 : 5; - // Alternate AB/BA to distribute shared JVM warmup and background storage work. - for (int position = 0; position < 2; position++) { - int path = (round + position) % 2; + // Rotate all three paths so each occupies every position during measured rounds. + for (int position = 0; position < paths.length; position++) { + int path = (round + position) % paths.length; WritePerformance result = - runConcurrentWrites( - executor, concurrency, seconds, path == 0 ? preparedWrite : tabletWrite); + runConcurrentWrites(executor, concurrency, seconds, paths[path]); System.out.printf( "%s round %d position %d %s: %d rows, %.3f s, %.1f rows/s%n", warmup ? "warmup" : "measured", warmup ? round + 1 : round - warmupRounds + 1, position + 1, - path == 0 ? "prepared numeric row" : "insertTablet batch=1", + pathNames[path], result.rows(), result.nanos() / 1_000_000_000.0, result.rowsPerSecond()); @@ -542,25 +566,31 @@ public class IoTDBTableModelSessionIT { } } } - for (int path = 0; path < 2; path++) { + for (int path = 0; path < paths.length; path++) { Arrays.sort(rates[path]); System.out.printf( "summary %s: median %.1f rows/s, aggregate %.1f rows/s, min %.1f, max %.1f%n", - path == 0 ? "prepared numeric row" : "insertTablet batch=1", + pathNames[path], (rates[path][measuredRounds / 2 - 1] + rates[path][measuredRounds / 2]) / 2, totalRows[path] * 1_000_000_000.0 / totalNanos[path], rates[path][0], rates[path][measuredRounds - 1]); } // Per-device monotonic timestamps span warmup and measurement, so no rows overwrite others. - for (int path = 0; path < 2; path++) { - String table = path == 0 ? "prepared_perf" : "tablet_perf"; - long[] rowsByThread = path == 0 ? preparedRows : tabletRows; + for (int path = 0; path < paths.length; path++) { + String table = + switch (path) { + case 0 -> "prepared_perf"; + case 1 -> "row_perf"; + default -> "tablet_perf"; + }; + long[] rowsByThread = pathRows[path]; long expectedRows = Arrays.stream(rowsByThread).sum(); for (int i = 0; i < concurrency; i++) { try (SessionDataSet dataSet = sessions[i].executeQueryStatement("SELECT count(*) FROM " + table)) { assertTrue(dataSet.hasNext()); + // Each worker writes to its own database, including warm-up rows. assertEquals(rowsByThread[i], dataSet.next().getFields().get(0).getLongV()); assertFalse(dataSet.hasNext()); } 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 bd2b7d57702..fa8d28d67fd 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 @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.path.AlignedFullPath; import org.apache.iotdb.commons.path.IFullPath; import org.apache.iotdb.commons.path.NonAlignedFullPath; import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.DataTypeInconsistentException; import org.apache.iotdb.db.exception.WriteProcessException; @@ -182,12 +183,13 @@ public abstract class AbstractMemTable implements IMemTable { private IWritableMemChunkGroup createAlignedMemChunkGroupIfNotExistAndGet( IDeviceID deviceId, List<IMeasurementSchema> schemaList) { - List<IMeasurementSchema> filteredSchemaList = - schemaList.stream().filter(Objects::nonNull).collect(Collectors.toList()); IWritableMemChunkGroup memChunkGroup = memTableMap.computeIfAbsent( deviceId, k -> { + // Existing devices need no filtered copy; new chunks own this mutable schema list. + List<IMeasurementSchema> filteredSchemaList = + schemaList.stream().filter(Objects::nonNull).collect(Collectors.toList()); seriesNumber += filteredSchemaList.size(); return new AlignedWritableMemChunkGroup(filteredSchemaList, k.isTableModel()); }); @@ -292,12 +294,26 @@ public abstract class AbstractMemTable implements IMemTable { } private int insertAlignedNumericRow(RelationalInsertNumericRowNode row) { - List<IMeasurementSchema> schemas = new ArrayList<>(); + int measurementCount = row.getMeasurements().length; + int valueCount = row.getValueCount(); + TsTableColumnCategory[] categories = row.getColumnCategories(); + List<IMeasurementSchema> schemas = new ArrayList<>(measurementCount); + int validValues = 0; int nullPoints = 0; long rowSize = Long.BYTES + Integer.BYTES; - for (int i = 0; i < row.getMeasurements().length; i++) { - if (!row.isValidMeasurement(i, true) || row.isNull(i)) { - if (row.isValidMeasurement(i, true) && i < row.getValueCount() && row.isNull(i)) { + for (int i = 0; i < measurementCount; i++) { + boolean valid = row.isValidMeasurement(i, true); + boolean hasValue = i < valueCount; + // Match numeric-row presence checks, counting valid NULL fields but not missing slots. + if (valid + && hasValue + && categories != null + && i < categories.length + && categories[i] != null) { + validValues++; + } + if (!valid || !hasValue || row.isNull(i)) { + if (valid && hasValue) { nullPoints++; } schemas.add(null); @@ -316,7 +332,6 @@ public abstract class AbstractMemTable implements IMemTable { .getAlignedMemChunk() .writeAlignedNumericPoints(row.getTime(), row.getNumericValues(), schemas); memSize += rowSize; - int validValues = row.getValidMeasurementNumber(true); int inserted = validValues - (IoTDBDescriptor.getInstance().getConfig().isIncludeNullValueInWriteThroughputMetric() diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTableNumericRowTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTableNumericRowTest.java new file mode 100644 index 00000000000..098f6ab72e5 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTableNumericRowTest.java @@ -0,0 +1,195 @@ +/* + * 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.exception.IllegalPathException; +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.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertNumericRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; +import org.apache.iotdb.db.utils.NumericRowValues; +import org.apache.iotdb.db.utils.datastructure.AlignedTVList; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.util.Arrays; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +public class AbstractMemTableNumericRowTest { + private PrimitiveMemTable memTable; + private boolean previousIncludeNull; + + @Before + public void setUp() { + memTable = new PrimitiveMemTable("numeric_test", "0"); + previousIncludeNull = + IoTDBDescriptor.getInstance().getConfig().isIncludeNullValueInWriteThroughputMetric(); + IoTDBDescriptor.getInstance().getConfig().setIncludeNullValueInWriteThroughputMetric(false); + } + + @After + public void tearDown() { + memTable.release(); + IoTDBDescriptor.getInstance() + .getConfig() + .setIncludeNullValueInWriteThroughputMetric(previousIncludeNull); + } + + @Test + public void testNullAndFailedFieldAccountingWithBothMetricSettings() throws Exception { + // TAGs and failed FIELDs never count; valid NULL FIELDs count only in the enabled metric. + NumericRowValues values = new NumericRowValues(1, 3); + values.setTag(0, "device"); + values.setRawValue(1, Double.doubleToRawLongBits(10)); + values.setNull(2); + RelationalInsertNumericRowNode row = + numericRow(new String[] {"tag", "a", "b", "failed"}, 1, values, 1); + row.markFailedMeasurement(3); + + assertEquals(1, memTable.insertAlignedRow(row)); + assertEquals(2, memTable.getTotalValueCount()); + assertEquals(1, memTable.getNullValueCount()); + assertEquals(1, memTable.getSeriesNumber()); + assertEquals(20, memTable.memSize()); + + IoTDBDescriptor.getInstance().getConfig().setIncludeNullValueInWriteThroughputMetric(true); + row.setTime(2); + assertEquals(2, memTable.insertAlignedRow(row)); + values.setNull(1); + row.setTime(3); + assertEquals(2, memTable.insertAlignedRow(row)); + assertEquals(5, memTable.getTotalPointsNum()); + assertEquals(6, memTable.getTotalValueCount()); + assertEquals(4, memTable.getNullValueCount()); + assertEquals(52, memTable.memSize()); + assertEquals(1, memTable.getSeriesNumber()); + AlignedTVList list = chunk(row).getWorkingTVList(); + assertEquals(10, list.getDoubleByValueIndex(0, 0), 0); + assertTrue(list.isNullValue(2, 0)); + } + + @Test + public void testMissingSlotsAndInvalidMetadataAreNotCounted() throws Exception { + // A valid schema beyond the value buffer is absent, not a NULL point or a new series. + NumericRowValues values = new NumericRowValues(1, 1); + values.setTag(0, "device"); + values.setRawValue(1, Double.doubleToRawLongBits(12)); + RelationalInsertNumericRowNode row = + numericRow(new String[] {"tag", "a", "missing", "invalid"}, 1, values, 1); + row.getMeasurementSchemas()[3] = null; + assertEquals(1, memTable.insertAlignedRow(row)); + assertEquals(1, memTable.getTotalValueCount()); + assertEquals(0, memTable.getNullValueCount()); + assertEquals(1, memTable.getSeriesNumber()); + assertEquals(20, memTable.memSize()); + assertEquals(12, chunk(row).getWorkingTVList().getDoubleByValueIndex(0, 0), 0); + } + + @Test + public void testExistingGroupCanGrowAfterNumericAndTabletWrites() throws Exception { + // Creation filters NULL schemas; a later Tablet materializes that field in the same group. + NumericRowValues values = new NumericRowValues(0, 2); + values.setRawValue(0, Double.doubleToRawLongBits(1)); + values.setNull(1); + RelationalInsertNumericRowNode row = numericRow(new String[] {"a", "b"}, 0, values, 1); + assertEquals(1, memTable.insertAlignedRow(row)); + AlignedWritableMemChunk original = chunk(row); + row.setTime(2); + assertEquals(1, memTable.insertAlignedRow(row)); + RelationalInsertTabletNode tablet = + new RelationalInsertTabletNode( + new PlanNodeId("tablet"), + new PartialPath("t"), + true, + new String[] {"b"}, + new TSDataType[] {TSDataType.DOUBLE}, + new MeasurementSchema[] {new MeasurementSchema("b", TSDataType.DOUBLE)}, + new long[] {3}, + null, + new Object[] {new double[] {30}}, + 1, + new TsTableColumnCategory[] {TsTableColumnCategory.FIELD}); + assertEquals(1, memTable.insertAlignedTablet(tablet, 0, 1, null)); + values.setRawValue(0, Double.doubleToRawLongBits(4)); + values.setRawValue(1, Double.doubleToRawLongBits(40)); + row.setTime(4); + assertEquals(2, memTable.insertAlignedRow(row)); + assertSame(original, chunk(row)); + assertEquals(2, memTable.getSeriesNumber()); + assertEquals(5, memTable.getTotalPointsNum()); + assertEquals(7, memTable.getTotalValueCount()); + assertEquals(2, memTable.getNullValueCount()); + AlignedTVList list = original.getWorkingTVList(); + assertTrue(list.isNullValue(0, 1)); + assertTrue(list.isNullValue(2, 0)); + assertEquals(30, list.getDoubleByValueIndex(2, 1), 0); + assertEquals(4, list.getDoubleByValueIndex(3, 0), 0); + assertEquals(40, list.getDoubleByValueIndex(3, 1), 0); + } + + @Test + public void testEmptyRowDoesNotCreateGroup() throws Exception { + // Empty input leaves counters and the device map untouched. + assertEquals( + 0, memTable.insertAlignedRow(numericRow(new String[0], 0, new NumericRowValues(0, 0), 1))); + assertTrue(memTable.getMemTableMap().isEmpty()); + assertEquals(0, memTable.memSize()); + assertEquals(0, memTable.getTotalValueCount()); + } + + private AlignedWritableMemChunk chunk(RelationalInsertNumericRowNode row) { + return ((AlignedWritableMemChunkGroup) memTable.getMemTableMap().get(row.getDeviceID())) + .getAlignedMemChunk(); + } + + private static RelationalInsertNumericRowNode numericRow( + String[] names, int tags, NumericRowValues values, long time) throws IllegalPathException { + TSDataType[] types = new TSDataType[names.length]; + TsTableColumnCategory[] categories = new TsTableColumnCategory[names.length]; + Arrays.fill(types, TSDataType.DOUBLE); + Arrays.fill(categories, TsTableColumnCategory.FIELD); + MeasurementSchema[] schemas = new MeasurementSchema[names.length]; + for (int i = 0; i < names.length; i++) { + if (i < tags) { + types[i] = TSDataType.STRING; + categories[i] = TsTableColumnCategory.TAG; + } + schemas[i] = new MeasurementSchema(names[i], types[i]); + } + return new RelationalInsertNumericRowNode( + new PlanNodeId("numeric"), + new PartialPath("t"), + names, + types, + schemas, + time, + values, + categories); + } +}
