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 56c20e3840f4057ea370e4311e6f667b10790804 Author: Tian Jiang <[email protected]> AuthorDate: Thu Sep 10 19:03:03 2026 +0800 tmp --- .../db/queryengine/plan/analyze/AnalyzeUtils.java | 1 + .../dataregion/memtable/AbstractMemTable.java | 46 +++++++++++++++++++++- .../memtable/AlignedWritableMemChunk.java | 1 + .../db/utils/datastructure/AlignedTVList.java | 41 +++++++++++++++++++ .../db/utils/datastructure/AlignedTVListTest.java | 45 +++++++++++++++++++++ 5 files changed, 132 insertions(+), 2 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java index 5da73066567..c4c16d5af7e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java @@ -60,6 +60,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Delete; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.TimeRange; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertMultiTabletsStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertNumericRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; 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 e24308cf505..bd2b7d57702 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 @@ -32,9 +32,11 @@ import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeDevicePathCache; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertNumericRowNode; import org.apache.iotdb.db.schemaengine.schemaregion.utils.ResourceByPathUtils; import org.apache.iotdb.db.storageengine.dataregion.flush.FlushStatus; import org.apache.iotdb.db.storageengine.dataregion.flush.NotifyFlushMemTable; @@ -241,10 +243,13 @@ public abstract class AbstractMemTable implements IMemTable { } @Override - public int insertAlignedRow(InsertRowNode insertRowNode) { + public int insertAlignedRow(AbstractInsertRowNode insertRowNode) { + if (insertRowNode instanceof RelationalInsertNumericRowNode numeric) { + return insertAlignedNumericRow(numeric); + } String[] measurements = insertRowNode.getMeasurements(); - Object[] values = insertRowNode.getValues(); + Object[] values = ((InsertRowNode) insertRowNode).getValues(); List<IMeasurementSchema> schemaList = new ArrayList<>(); List<TSDataType> dataTypes = new ArrayList<>(); Object[] writableValues = new Object[measurements == null ? 0 : measurements.length]; @@ -286,6 +291,43 @@ public abstract class AbstractMemTable implements IMemTable { return pointsInserted; } + private int insertAlignedNumericRow(RelationalInsertNumericRowNode row) { + List<IMeasurementSchema> schemas = new ArrayList<>(); + 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)) { + nullPoints++; + } + schemas.add(null); + } else { + schemas.add(row.getMeasurementSchemas()[i]); + rowSize += row.getDataType(i).getDataTypeSize(); + } + } + if (schemas.isEmpty()) { + return 0; + } + AlignedWritableMemChunkGroup group = + (AlignedWritableMemChunkGroup) + createAlignedMemChunkGroupIfNotExistAndGet(row.getDeviceID(), schemas); + group + .getAlignedMemChunk() + .writeAlignedNumericPoints(row.getTime(), row.getNumericValues(), schemas); + memSize += rowSize; + int validValues = row.getValidMeasurementNumber(true); + int inserted = + validValues + - (IoTDBDescriptor.getInstance().getConfig().isIncludeNullValueInWriteThroughputMetric() + ? 0 + : nullPoints); + totalPointsNum += inserted; + totalValueCount += validValues; + nullValueCount += nullPoints; + return inserted; + } + @Override public int insertTablet(InsertTabletNode insertTabletNode, int start, int end) throws WriteProcessException { 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 d1c15384173..37bd7d3c870 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 @@ -27,6 +27,7 @@ import org.apache.iotdb.db.i18n.DataNodeMiscMessages; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; +import org.apache.iotdb.db.utils.NumericRowValues; import org.apache.iotdb.db.utils.datastructure.AlignedTVList; import org.apache.iotdb.db.utils.datastructure.BatchEncodeInfo; import org.apache.iotdb.db.utils.datastructure.MemPointIterator; 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 d2176c6891e..88c8409cf7a 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 @@ -31,6 +31,7 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferVie import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; import org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager; import org.apache.iotdb.db.utils.MathUtils; +import org.apache.iotdb.db.utils.NumericRowValues; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.block.column.Column; @@ -523,6 +524,46 @@ public abstract class AlignedTVList extends TVList { } } + /** Writes row bits into native column arrays without a boxed intermediate row. */ + public synchronized void putAlignedNumericValue( + long timestamp, NumericRowValues row, int[] columnMapping) { + checkExpansion(); + int arrayIndex = rowCount / ARRAY_SIZE; + int elementIndex = rowCount % ARRAY_SIZE; + maxTime = Math.max(maxTime, timestamp); + minTime = Math.min(minTime, timestamp); + timestamps.get(arrayIndex)[elementIndex] = timestamp; + for (int i = 0; i < values.size(); i++) { + int source = columnMapping[i]; + if (source < 0 || row.isNull(source)) { + markNullValue(i, arrayIndex, elementIndex); + continue; + } + Object valueArray = getOrCreateValueArray(i, arrayIndex, elementIndex); + long value = row.get(source); + switch (dataTypes.get(i)) { + case BOOLEAN -> ((boolean[]) valueArray)[elementIndex] = value != 0; + case INT32, DATE -> ((int[]) valueArray)[elementIndex] = (int) value; + case INT64, TIMESTAMP -> ((long[]) valueArray)[elementIndex] = value; + case FLOAT -> ((float[]) valueArray)[elementIndex] = Float.intBitsToFloat((int) value); + case DOUBLE -> ((double[]) valueArray)[elementIndex] = Double.longBitsToDouble(value); + default -> throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT); + } + } + if (indices != null) { + indices.get(arrayIndex)[elementIndex] = rowCount; + } + rowCount++; + if (sorted) { + if (rowCount > 1 && timestamp < getTime(rowCount - 2)) { + sorted = false; + segmentMovedMap = null; + } else { + seqRowCount++; + } + } + } + @Override public Object getAlignedValue(int index) { return getAlignedValueForQuery(index, null, null); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java index 708f2113cdf..a39fc45def7 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java @@ -20,6 +20,7 @@ package org.apache.iotdb.db.utils.datastructure; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; +import org.apache.iotdb.db.utils.NumericRowValues; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.common.conf.TSFileConfig; @@ -44,9 +45,53 @@ import java.util.Set; import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE; import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_ARRAY_HEADER; import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF; +import static org.junit.Assert.assertEquals; public class AlignedTVListTest { + @Test + public void testNumericRowsMatchObjectRowsAcrossBlocksAndSorting() { + // Reordered columns, nulls and descending timestamps must match the existing write path + // across a primitive-array block boundary and after sorting. + List<TSDataType> types = + Arrays.asList( + TSDataType.BOOLEAN, + TSDataType.INT32, + TSDataType.FLOAT, + TSDataType.DOUBLE, + TSDataType.INT64, + TSDataType.STRING); + AlignedTVList numeric = AlignedTVList.newAlignedList(types); + AlignedTVList boxed = AlignedTVList.newAlignedList(types); + try { + for (int i = ARRAY_SIZE + 3; i >= 0; i--) { + NumericRowValues row = new NumericRowValues(5); + row.getValues()[0] = i; + row.getValues()[1] = i % 2; + row.getValues()[2] = -i; + row.getValues()[3] = Float.floatToRawIntBits(-0.0f); + row.getValues()[4] = Double.doubleToRawLongBits(-i * 0.25); + if (i % 3 == 0) { + row.setNull(2); + } + // Also exercise a missing STRING column in an existing memtable. + numeric.putAlignedNumericValue(i, row, new int[] {1, 2, 3, 4, 0, -1}); + boxed.putAlignedValue( + i, new Object[] {i % 2 != 0, i % 3 == 0 ? null : -i, -0.0f, -i * 0.25, (long) i, null}); + } + numeric.sort(); + boxed.sort(); + assertEquals(boxed.rowCount(), numeric.rowCount()); + for (int i = 0; i < numeric.rowCount(); i++) { + assertEquals(boxed.getTime(i), numeric.getTime(i)); + assertEquals(boxed.getAlignedValue(i), numeric.getAlignedValue(i)); + } + } finally { + numeric.clear(); + boxed.clear(); + } + } + @Test public void testValueListArrayMemCostExcludesLazyBitmapAndNullPlaceholder() { long expected = (long) ARRAY_SIZE * Long.BYTES + NUM_BYTES_ARRAY_HEADER + NUM_BYTES_OBJECT_REF;
