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 7ad6ab3edd03f81f07ed5eaa9e8e6dbe4e80b5f0 Author: Tian Jiang <[email protected]> AuthorDate: Thu Sep 17 11:07:48 2026 +0800 add null bitmap --- .../it/session/IoTDBTableModelSessionIT.java | 79 ++++++++++++++++++++++ .../org/apache/iotdb/isession/ITableSession.java | 14 ++++ .../java/org/apache/iotdb/isession/NumericRow.java | 29 +++++++- .../isession/NumericRowPreparedStatement.java | 10 +++ .../apache/iotdb/session/i18n/SessionMessages.java | 2 + .../apache/iotdb/session/i18n/SessionMessages.java | 2 + .../java/org/apache/iotdb/session/Session.java | 5 +- .../apache/iotdb/session/SessionConnection.java | 9 +++ .../org/apache/iotdb/session/TableSession.java | 35 +++++++++- .../iotdb/session/pool/TableSessionWrapper.java | 15 +++- .../apache/iotdb/session/util/SessionUtils.java | 53 ++++++++++++++- .../session/PreparedNumericRowBufferTest.java | 56 +++++++++++++++ .../iotdb/session/util/SessionUtilsTest.java | 63 +++++++++++++++++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 6 ++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 6 ++ .../protocol/session/PreparedNumericRowInfo.java | 25 ++++++- .../protocol/thrift/impl/ClientRPCServiceImpl.java | 3 +- .../plan/parser/StatementGenerator.java | 3 +- .../statement/crud/InsertNumericRowStatement.java | 17 ++++- .../apache/iotdb/db/utils/NumericRowValues.java | 30 ++++++++ .../session/PreparedNumericRowInfoTest.java | 75 ++++++++++++++++++++ .../thrift-datanode/src/main/thrift/client.thrift | 5 ++ 22 files changed, 527 insertions(+), 15 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 29ee26b30c0..9ac67ad9653 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 @@ -34,6 +34,7 @@ import org.apache.iotdb.rpc.StatementExecutionException; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.read.common.RowRecord; +import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.write.record.Tablet; import org.junit.After; import org.junit.Assume; @@ -232,6 +233,84 @@ public class IoTDBTableModelSessionIT { } } + @Test + public void testNumericRowNullBitmap() throws Exception { + // Exercise bits 0/7/8, interleaved TAG metadata, prepared fast/fallback paths and pooled + // sessions. + try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) { + session.executeNonQueryStatement("CREATE DATABASE numeric_nulls"); + session.executeNonQueryStatement("USE numeric_nulls"); + StringBuilder ddl = new StringBuilder("CREATE TABLE readings(tag STRING TAG"); + List<String> names = new ArrayList<>(); + List<TSDataType> types = new ArrayList<>(); + List<ColumnCategory> categories = new ArrayList<>(); + long[] values = new long[9]; + for (int i = 0; i < 9; i++) { + ddl.append(", f").append(i).append(" DOUBLE FIELD"); + names.add("f" + i); + types.add(TSDataType.DOUBLE); + categories.add(ColumnCategory.FIELD); + values[i] = Double.doubleToRawLongBits(i + 0.25); + } + session.executeNonQueryStatement(ddl.append(")").toString()); + names.add(1, "tag"); + types.add(1, TSDataType.STRING); + categories.add(1, ColumnCategory.TAG); + BitMap nulls = new BitMap(9); + nulls.mark(0); + nulls.mark(7); + nulls.mark(8); + String[] tags = {"device"}; + session.insert(new NumericRow("readings", 1, names, types, categories, values, tags, nulls)); + try (NumericRowPreparedStatement prepared = + session.prepareInsertNumericRow("readings", names, types, categories)) { + session.executePreparedInsertNumericRow(prepared, 2, values, tags, nulls); + session.executePreparedInsertNumericRow(prepared, 3, values, tags); + // All-null table rows retain their timestamp/TAGs; null slots must never become zeros. + BitMap allNulls = new BitMap(9); + allNulls.markAll(); + session.executePreparedInsertNumericRow(prepared, 3, values, tags, allNulls); + session.insert( + new NumericRow("readings", 10, names, types, categories, values, tags, allNulls)); + session.executePreparedInsertNumericRow(prepared, 11, values, tags, allNulls); + session.executeNonQueryStatement("ALTER TABLE readings ADD COLUMN tag2 STRING TAG"); + session.executePreparedInsertNumericRow(prepared, 4, values, tags, nulls); + } + try (ITableSessionPool pool = EnvFactory.getEnv().getTableSessionPool(1, "numeric_nulls"); + ITableSession pooled = pool.getSession(); + NumericRowPreparedStatement prepared = + pooled.prepareInsertNumericRow("readings", names, types, categories)) { + pooled.insert(new NumericRow("readings", 5, names, types, categories, values, tags, nulls)); + pooled.executePreparedInsertNumericRow(prepared, 6, values, tags, nulls); + } + assertNumericNullRows(session); + session.executeNonQueryStatement("FLUSH"); + assertNumericNullRows(session); + } + } + + private void assertNumericNullRows(ITableSession session) throws Exception { + try (SessionDataSet dataSet = + session.executeQueryStatement( + "SELECT time,tag,tag2,f0,f1,f2,f3,f4,f5,f6,f7,f8 FROM readings ORDER BY time")) { + for (int time : new int[] {1, 2, 3, 4, 5, 6, 10, 11}) { + assertTrue(dataSet.hasNext()); + RowRecord row = dataSet.next(); + assertEquals(time, row.getFields().get(0).getLongV()); + assertEquals("device", row.getFields().get(1).getStringValue()); + assertNull(row.getFields().get(2).getDataType()); + for (int field = 0; field < 9; field++) { + if (time >= 10 || (time != 3 && (field == 0 || field == 7 || field == 8))) { + assertNull(row.getFields().get(field + 3).getDataType()); + } else { + assertEquals(field + 0.25, row.getFields().get(field + 3).getDoubleV(), 0); + } + } + } + assertFalse(dataSet.hasNext()); + } + } + @Test public void testPreparedNumericRowAndTabletWritePerformance() throws Exception { Assume.assumeTrue( diff --git a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/ITableSession.java b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/ITableSession.java index d14abab1238..35a2f71b7dc 100644 --- a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/ITableSession.java +++ b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/ITableSession.java @@ -24,6 +24,7 @@ import org.apache.iotdb.rpc.StatementExecutionException; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.write.record.Tablet; import java.util.List; @@ -73,6 +74,19 @@ public interface ITableSession extends AutoCloseable { NumericRowPreparedStatement prepared, long timestamp, long[] values, String[] tagValues) throws StatementExecutionException, IoTDBConnectionException; + /** + * Executes a prepared row with an optional FIELD-only null bitmap sized to {@code values.length}. + * Marked slots are null and their values are ignored. Do not mutate the bitmap during execution. + * Requires a server supporting numeric row null bitmaps. + */ + void executePreparedInsertNumericRow( + NumericRowPreparedStatement prepared, + long timestamp, + long[] values, + String[] tagValues, + BitMap nullBitmap) + throws StatementExecutionException, IoTDBConnectionException; + /** * Executes a non-query SQL statement, such as a DDL or DML command. * diff --git a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRow.java b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRow.java index 502c75e05e4..040cb7632ba 100644 --- a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRow.java +++ b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRow.java @@ -21,6 +21,7 @@ package org.apache.iotdb.isession; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.BitMap; import java.util.List; @@ -29,8 +30,9 @@ import java.util.List; * * <p>Column names, types and categories describe all supplied columns in the same order. The value * arrays contain only their respective FIELD or TAG values, in occurrence order in the column list. - * TAGs may be null or empty; missing TAG columns are null. ATTRIBUTE columns, nonnumeric FIELDs and - * null FIELD values are not supported. + * TAGs may be null or empty; missing TAG columns are null. An optional null bitmap marks null + * FIELDs by their index in {@code values}; marked slots retain their position but are not sent. + * ATTRIBUTE columns and nonnumeric FIELDs are not supported. * * <p>BOOLEAN uses zero for false and nonzero for true. INT32 and DATE use the low 32 bits (DATE is * encoded as yyyyMMdd); INT64 and TIMESTAMP use the long value. FLOAT uses {@link @@ -48,6 +50,7 @@ public final class NumericRow { private final List<ColumnCategory> columnCategories; private final long[] values; private final String[] tagValues; + private final BitMap nullBitmap; /** Creates a row with numeric FIELDs and optional STRING TAGs in separate arrays. */ public NumericRow( @@ -58,6 +61,23 @@ public final class NumericRow { List<ColumnCategory> columnCategories, long[] values, String[] tagValues) { + this(tableName, timestamp, columnNames, dataTypes, columnCategories, values, tagValues, null); + } + + /** + * Creates a row with an optional FIELD-only null bitmap, sized exactly to {@code values.length}. + * A marked bit denotes null. The bitmap is retained and must not change during insertion. + * Requires a server supporting numeric row null bitmaps when the bitmap is supplied. + */ + public NumericRow( + String tableName, + long timestamp, + List<String> columnNames, + List<TSDataType> dataTypes, + List<ColumnCategory> columnCategories, + long[] values, + String[] tagValues, + BitMap nullBitmap) { this.tableName = tableName; this.timestamp = timestamp; this.columnNames = columnNames; @@ -65,6 +85,7 @@ public final class NumericRow { this.columnCategories = columnCategories; this.values = values; this.tagValues = tagValues; + this.nullBitmap = nullBitmap; } public String getTableName() { @@ -94,4 +115,8 @@ public final class NumericRow { public String[] getTagValues() { return tagValues; } + + public BitMap getNullBitmap() { + return nullBitmap; + } } diff --git a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRowPreparedStatement.java b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRowPreparedStatement.java index 8852d05a8a8..ee0b0f21804 100644 --- a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRowPreparedStatement.java +++ b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRowPreparedStatement.java @@ -29,6 +29,7 @@ public final class NumericRowPreparedStatement implements AutoCloseable { private final int valueBytes; private ByteBuffer tagBuffer; private ByteBuffer valueBuffer; + private ByteBuffer nullBitmapBuffer; private long preparedId = -1; private Object owner; private Runnable closeAction = () -> {}; @@ -108,6 +109,15 @@ public final class NumericRowPreparedStatement implements AutoCloseable { return preparedId; } + /** Returns a reusable FIELD null bitmap buffer, allocated only for nullable executions. */ + public ByteBuffer prepareNullBitmapBuffer() { + if (nullBitmapBuffer == null) { + nullBitmapBuffer = ByteBuffer.allocate((dataTypes.size() - tagCount + 7) / 8); + } + nullBitmapBuffer.clear(); + return nullBitmapBuffer; + } + public Object getOwner() { return owner; } diff --git a/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java b/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java index 173260e4a07..a1bdf45262e 100644 --- a/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java +++ b/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java @@ -20,6 +20,8 @@ package org.apache.iotdb.session.i18n; public final class SessionMessages { + public static final String EXCEPTION_NUMERIC_ROW_NULL_BITMAP_SIZE_MISMATCH_EXPECTED_ARG_FIELDS_BUT_GOT_ARG_8D30F6A0 = + "Numeric row null bitmap size mismatch: expected %d FIELDs but got %d"; public static final String EXCEPTION_NUMERIC_ROW_ARGUMENTS_MUST_BE_NON_NULL_COLUMN_METADATA_LENGTHS_AND_PER_CATEGORY_VALUE_COUNTS_MUST_MATCH_EB12B233 = "Numeric row arguments must be non-null; column metadata lengths and per-category value counts must match."; public static final String EXCEPTION_NUMERIC_ROWS_SUPPORT_NUMERIC_FIELD_COLUMNS_AND_STRING_TAG_COLUMNS_ONLY_87159FFA = "Numeric rows support numeric FIELD columns and STRING TAG columns only."; diff --git a/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java b/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java index b45704d8738..260b000e61a 100644 --- a/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java +++ b/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java @@ -20,6 +20,8 @@ package org.apache.iotdb.session.i18n; public final class SessionMessages { + public static final String EXCEPTION_NUMERIC_ROW_NULL_BITMAP_SIZE_MISMATCH_EXPECTED_ARG_FIELDS_BUT_GOT_ARG_8D30F6A0 = + "Numeric row 的 null bitmap 大小不匹配：预期 %d 个 FIELD，实际为 %d"; public static final String EXCEPTION_NUMERIC_ROW_ARGUMENTS_MUST_BE_NON_NULL_COLUMN_METADATA_LENGTHS_AND_PER_CATEGORY_VALUE_COUNTS_MUST_MATCH_EB12B233 = "数值行参数不能为 null；列元数据长度必须一致，FIELD 和 TAG 的数量必须分别与对应的值数组长度一致。"; public static final String EXCEPTION_NUMERIC_ROWS_SUPPORT_NUMERIC_FIELD_COLUMNS_AND_STRING_TAG_COLUMNS_ONLY_87159FFA = "数值行仅支持数值类型的 FIELD 列和 STRING 类型的 TAG 列。"; diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java index 835cf924777..f5c6ebd1b80 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java @@ -2922,7 +2922,7 @@ public class Session implements ISession { } ByteBuffer valueBuffer = SessionUtils.getValueBuffer( - rpcTypes, values, rpcMeasurements, rpcColumnCategories, tagValues); + rpcTypes, values, rpcMeasurements, rpcColumnCategories, tagValues, row.getNullBitmap()); TSInsertRecordReq request = new TSInsertRecordReq(); request.setPrefixPath(tableName); request.setTimestamp(timestamp); @@ -2930,6 +2930,9 @@ public class Session implements ISession { request.setValues(valueBuffer); request.setIsWriteToTable(true); request.setIsNumericRow(true); + if (row.getNullBitmap() != null) { + request.setNullBitmap(SessionUtils.encodeNumericRowNullBitmap(row.getNullBitmap(), null)); + } request.setIsAligned(true); request.setColumnCategoryies( rpcColumnCategories.stream() diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java index e01c5dd8a20..f98e58965ae 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/SessionConnection.java @@ -782,8 +782,17 @@ public class SessionConnection { protected void executePreparedNumericRow( long id, long timestamp, ByteBuffer values, ByteBuffer tags) throws IoTDBConnectionException, StatementExecutionException { + executePreparedNumericRow(id, timestamp, values, tags, null); + } + + protected void executePreparedNumericRow( + long id, long timestamp, ByteBuffer values, ByteBuffer tags, ByteBuffer nullBitmap) + throws IoTDBConnectionException, StatementExecutionException { TSExecutePreparedNumericRowReq request = new TSExecutePreparedNumericRowReq(sessionId, id, timestamp, values, tags); + if (nullBitmap != null) { + request.setNullBitmap(nullBitmap); + } try { RpcUtils.verifySuccess(client.executePreparedNumericRow(request)); } catch (TException e) { diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSession.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSession.java index 7833557e10b..da93c37586f 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSession.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSession.java @@ -27,9 +27,11 @@ import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; import org.apache.iotdb.service.rpc.thrift.TSPrepareNumericRowReq; import org.apache.iotdb.session.i18n.SessionMessages; +import org.apache.iotdb.session.util.SessionUtils; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.utils.ReadWriteIOUtils; import org.apache.tsfile.write.record.Tablet; @@ -101,6 +103,17 @@ public class TableSession implements ITableSession { public void executePreparedInsertNumericRow( NumericRowPreparedStatement prepared, long timestamp, long[] values, String[] tagValues) throws StatementExecutionException, IoTDBConnectionException { + executePreparedInsertNumericRow(prepared, timestamp, values, tagValues, null); + } + + @Override + public void executePreparedInsertNumericRow( + NumericRowPreparedStatement prepared, + long timestamp, + long[] values, + String[] tagValues, + BitMap nullBitmap) + throws StatementExecutionException, IoTDBConnectionException { SessionConnection connection = session.getDefaultSessionConnection(); if (prepared == null || prepared.isClosed() || prepared.getOwner() != connection) { throw new IoTDBConnectionException( @@ -109,10 +122,14 @@ public class TableSession implements ITableSession { } int tagCount = prepared.getTagCount(); int fieldCount = prepared.getDataTypes().size() - tagCount; - if (values.length != fieldCount || tagValues.length != tagCount) { + if (values == null + || tagValues == null + || values.length != fieldCount + || tagValues.length != tagCount) { throw new IllegalArgumentException( SessionMessages.EXCEPTION_PREPARED_NUMERIC_ROW_VALUE_COUNT_MISMATCH_9ACD3A0C); } + SessionUtils.validateNumericRowNullBitmap(nullBitmap, fieldCount); int tagBytes = 0; for (String tag : tagValues) { tagBytes += ReadWriteIOUtils.sizeToWrite(tag); @@ -127,6 +144,9 @@ public class TableSession implements ITableSession { for (int i = 0; i < prepared.getDataTypes().size(); i++) { if (prepared.getColumnCategories().get(i) == ColumnCategory.FIELD) { long value = values[valueIndex++]; + if (nullBitmap != null && nullBitmap.isMarked(valueIndex - 1)) { + continue; + } switch (prepared.getDataTypes().get(i)) { case BOOLEAN -> valueBuffer.put((byte) (value == 0 ? 0 : 1)); case INT32, DATE, FLOAT -> valueBuffer.putInt((int) value); @@ -138,8 +158,17 @@ public class TableSession implements ITableSession { } } valueBuffer.flip(); - connection.executePreparedNumericRow( - prepared.getPreparedId(), timestamp, valueBuffer, tagBuffer); + if (nullBitmap == null) { + connection.executePreparedNumericRow( + prepared.getPreparedId(), timestamp, valueBuffer, tagBuffer); + } else { + connection.executePreparedNumericRow( + prepared.getPreparedId(), + timestamp, + valueBuffer, + tagBuffer, + SessionUtils.encodeNumericRowNullBitmap(nullBitmap, prepared.prepareNullBitmapBuffer())); + } } @Override diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionWrapper.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionWrapper.java index 627501a3200..7c950949d0d 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionWrapper.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionWrapper.java @@ -31,6 +31,7 @@ import org.apache.iotdb.session.i18n.SessionMessages; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.write.record.Tablet; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -104,8 +105,20 @@ public class TableSessionWrapper implements ITableSession { public void executePreparedInsertNumericRow( NumericRowPreparedStatement prepared, long timestamp, long[] values, String[] tagValues) throws StatementExecutionException, IoTDBConnectionException { + executePreparedInsertNumericRow(prepared, timestamp, values, tagValues, null); + } + + @Override + public void executePreparedInsertNumericRow( + NumericRowPreparedStatement prepared, + long timestamp, + long[] values, + String[] tagValues, + BitMap nullBitmap) + throws StatementExecutionException, IoTDBConnectionException { try { - tableSession.executePreparedInsertNumericRow(prepared, timestamp, values, tagValues); + tableSession.executePreparedInsertNumericRow( + prepared, timestamp, values, tagValues, nullBitmap); } catch (IoTDBConnectionException e) { sessionPool.cleanSessionAndMayThrowConnectionException(session); closed.set(true); diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java index 40fa0fae41f..fd1dd00cc38 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java @@ -171,6 +171,18 @@ public class SessionUtils { List<ColumnCategory> categories, String[] tagValues) throws IoTDBConnectionException { + return getValueBuffer(types, values, measurements, categories, tagValues, null); + } + + /** Null FIELD slots retain their indices but consume no payload bytes. */ + public static ByteBuffer getValueBuffer( + List<TSDataType> types, + long[] values, + List<String> measurements, + List<ColumnCategory> categories, + String[] tagValues, + BitMap nullBitmap) + throws IoTDBConnectionException { if (types == null || values == null || measurements == null @@ -180,6 +192,7 @@ public class SessionUtils { || types.size() != categories.size()) { throw invalidNumericRowArguments(); } + validateNumericRowNullBitmap(nullBitmap, values.length); int fieldCount = 0; int tagCount = 0; int length = types.size(); @@ -191,14 +204,20 @@ public class SessionUtils { } length += ReadWriteIOUtils.sizeToWrite(tagValues[tagCount++]); } else if (categories.get(i) == ColumnCategory.FIELD && type != null) { - fieldCount++; - length += + if (fieldCount >= values.length) { + throw invalidNumericRowArguments(); + } + int valueSize = switch (type) { case BOOLEAN -> Byte.BYTES; case INT32, DATE, FLOAT -> Integer.BYTES; case INT64, TIMESTAMP, DOUBLE -> Long.BYTES; default -> throw new IoTDBConnectionException(MSG_UNSUPPORTED_DATA_TYPE + type); }; + if (nullBitmap == null || !nullBitmap.isMarked(fieldCount)) { + length += valueSize; + } + fieldCount++; } else { throw new IllegalArgumentException( SessionMessages @@ -226,6 +245,9 @@ public class SessionUtils { TSDataType type = types.get(i); if (categories.get(i) == ColumnCategory.FIELD) { long value = values[fieldIndex++]; + if (nullBitmap != null && nullBitmap.isMarked(fieldIndex - 1)) { + continue; + } switch (type) { case BOOLEAN -> buffer.put((byte) (value == 0 ? 0 : 1)); case INT32, DATE, FLOAT -> buffer.putInt((int) value); @@ -244,6 +266,33 @@ public class SessionUtils { .EXCEPTION_NUMERIC_ROW_ARGUMENTS_MUST_BE_NON_NULL_COLUMN_METADATA_LENGTHS_AND_PER_CATEGORY_VALUE_COUNTS_MUST_MATCH_EB12B233); } + public static void validateNumericRowNullBitmap(BitMap nullBitmap, int fieldCount) { + if (nullBitmap != null && nullBitmap.getSize() != fieldCount) { + throw new IllegalArgumentException( + String.format( + SessionMessages + .EXCEPTION_NUMERIC_ROW_NULL_BITMAP_SIZE_MISMATCH_EXPECTED_ARG_FIELDS_BUT_GOT_ARG_8D30F6A0, + fieldCount, + nullBitmap.getSize())); + } + } + + /** Encodes FIELD bits LSB first with zero padding, including when BitMap.markAll() was used. */ + public static ByteBuffer encodeNumericRowNullBitmap(BitMap nullBitmap, ByteBuffer buffer) { + int fieldCount = nullBitmap.getSize(); + int length = (fieldCount + 7) / 8; + if (buffer == null) { + buffer = ByteBuffer.allocate(length); + } + buffer.clear(); + buffer.put(nullBitmap.getByteArray(), 0, length); + if (fieldCount % 8 != 0) { + buffer.put(length - 1, (byte) (buffer.get(length - 1) & ((1 << (fieldCount % 8)) - 1))); + } + buffer.flip(); + return buffer; + } + public static int calculateLength(List<TSDataType> types, List<? extends Object> values) throws IoTDBConnectionException { int res = 0; diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/PreparedNumericRowBufferTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/PreparedNumericRowBufferTest.java index a409d6844f0..500e4a7b088 100644 --- a/iotdb-client/session/src/test/java/org/apache/iotdb/session/PreparedNumericRowBufferTest.java +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/PreparedNumericRowBufferTest.java @@ -24,6 +24,7 @@ import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.utils.ReadWriteIOUtils; import org.junit.Test; @@ -44,6 +45,61 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; public class PreparedNumericRowBufferTest { + // Reusing the statement after sparse/all-null rows must reset both payload and bitmap cursors. + @Test + public void testNullBitmapBufferReuse() throws Exception { + Session session = mock(Session.class); + SessionConnection connection = mock(SessionConnection.class); + when(session.getDefaultSessionConnection()).thenReturn(connection); + TableSession tableSession = new TableSession(session); + NumericRowPreparedStatement prepared = + new NumericRowPreparedStatement( + "t", + List.of("a", "b", "c"), + List.of(TSDataType.INT64, TSDataType.INT64, TSDataType.INT64), + List.of(ColumnCategory.FIELD, ColumnCategory.FIELD, ColumnCategory.FIELD)); + prepared.bind(1, connection, () -> {}); + List<ByteBuffer> bitmaps = new ArrayList<>(); + List<ByteBuffer> payloads = new ArrayList<>(); + int[] masks = {2, 7, 0}; + doAnswer( + invocation -> { + int index = bitmaps.size(); + ByteBuffer values = invocation.getArgument(2); + ByteBuffer bitmap = invocation.getArgument(4); + bitmaps.add(bitmap); + payloads.add(values); + assertEquals(masks[index], bitmap.get()); + assertFalse(bitmap.hasRemaining()); + for (int field = 0; field < 3; field++) { + if ((masks[index] & (1 << field)) == 0) { + assertEquals(field + 10, values.getLong()); + } + } + assertFalse(values.hasRemaining()); + return null; + }) + .when(connection) + .executePreparedNumericRow( + eq(1L), anyLong(), any(ByteBuffer.class), any(ByteBuffer.class), any(ByteBuffer.class)); + BitMap nulls = new BitMap(3); + for (int mask : masks) { + nulls.reset(); + for (int i = 0; i < 3; i++) { + if ((mask & (1 << i)) != 0) { + nulls.mark(i); + } + } + tableSession.executePreparedInsertNumericRow( + prepared, 1, new long[] {10, 11, 12}, new String[0], nulls); + } + assertEquals(3, bitmaps.size()); + for (int i = 1; i < 3; i++) { + assertSame(bitmaps.get(0), bitmaps.get(i)); + assertSame(payloads.get(0), payloads.get(i)); + } + } + @Test public void testBuffersAcrossTagGrowthShrinkAndRpcFailure() throws Exception { Session session = mock(Session.class); diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java index dfb1ef4a2bf..632fbbda8fb 100644 --- a/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java @@ -46,6 +46,69 @@ import static org.junit.Assert.assertThrows; public class SessionUtilsTest { + // Null FIELDs are omitted without shifting indices around interleaved TAG metadata or NaN bits. + @Test + public void testNumericRowNullBitmapEncoding() throws Exception { + BitMap nulls = new BitMap(9); + nulls.mark(0); + nulls.mark(7); + nulls.mark(8); + List<TSDataType> types = new ArrayList<>(Collections.nCopies(9, TSDataType.DOUBLE)); + List<ColumnCategory> categories = new ArrayList<>(Collections.nCopies(9, ColumnCategory.FIELD)); + types.add(3, TSDataType.STRING); + categories.add(3, ColumnCategory.TAG); + long[] values = new long[9]; + Arrays.fill(values, 0x7ff8000000001234L); + ByteBuffer payload = + SessionUtils.getValueBuffer( + types, values, Collections.nCopies(10, "v"), categories, new String[] {"tag"}, nulls); + for (TSDataType type : types) { + Assert.assertEquals(type.serialize(), payload.get()); + } + Assert.assertEquals("tag", ReadWriteIOUtils.readString(payload)); + for (int i = 0; i < 6; i++) { + Assert.assertEquals(0x7ff8000000001234L, payload.getLong()); + } + Assert.assertFalse(payload.hasRemaining()); + Assert.assertEquals( + ByteBuffer.wrap(new byte[] {(byte) 0x81, 1}), + SessionUtils.encodeNumericRowNullBitmap(nulls, null)); + } + + // markAll may set padding in BitMap; the wire representation uses exactly ceil(FIELDs/8) bytes. + @Test + public void testNumericRowNullBitmapBoundaries() throws Exception { + for (int size : new int[] {1, 7, 8, 9, 16}) { + BitMap nulls = new BitMap(size); + nulls.markAll(); + byte[] original = nulls.getByteArray().clone(); + ByteBuffer bitmap = SessionUtils.encodeNumericRowNullBitmap(nulls, null); + Assert.assertEquals((size + 7) / 8, bitmap.remaining()); + Assert.assertEquals( + size % 8 == 0 ? 255 : (1 << (size % 8)) - 1, bitmap.get(bitmap.limit() - 1) & 255); + Assert.assertArrayEquals(original, nulls.getByteArray()); + ByteBuffer payload = + SessionUtils.getValueBuffer( + Collections.nCopies(size, TSDataType.INT64), + new long[size], + Collections.nCopies(size, "v"), + Collections.nCopies(size, ColumnCategory.FIELD), + new String[0], + nulls); + Assert.assertEquals(size, payload.remaining()); + } + assertThrows( + IllegalArgumentException.class, + () -> + SessionUtils.getValueBuffer( + List.of(TSDataType.INT64), + new long[1], + List.of("v"), + List.of(ColumnCategory.FIELD), + new String[0], + new BitMap(2))); + } + @Test public void testNumericRowEncodingPreservesRawBits() throws IoTDBConnectionException { List<TSDataType> types = diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index bf7a3601003..63de7dbb603 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -20,6 +20,12 @@ package org.apache.iotdb.db.i18n; public final class DataNodeQueryMessages { + public static final String EXCEPTION_NUMERIC_ROW_NULL_BITMAP_LENGTH_MISMATCH_EXPECTED_ARG_BYTES_FOR_ARG_FIELDS_BUT_GOT_ARG_31F1DB42 = + "Numeric row null bitmap length mismatch: expected %d bytes for %d FIELDs but got %d"; + public static final String EXCEPTION_NUMERIC_ROW_NULL_BITMAP_HAS_SET_BITS_OUTSIDE_ARG_FIELDS_AD6066D1 = + "Numeric row null bitmap has set bits outside %d FIELDs"; + public static final String EXCEPTION_NUMERIC_ROW_PAYLOAD_HAS_ARG_TRAILING_BYTES_89E46A34 = + "Numeric row payload has %d trailing bytes"; // --- Common --- diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index f26f2306be2..12e61875802 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -20,6 +20,12 @@ package org.apache.iotdb.db.i18n; public final class DataNodeQueryMessages { + public static final String EXCEPTION_NUMERIC_ROW_NULL_BITMAP_LENGTH_MISMATCH_EXPECTED_ARG_BYTES_FOR_ARG_FIELDS_BUT_GOT_ARG_31F1DB42 = + "Numeric row 的 null bitmap 长度不匹配：预期 %d 字节（对应 %d 个 FIELD），实际为 %d"; + public static final String EXCEPTION_NUMERIC_ROW_NULL_BITMAP_HAS_SET_BITS_OUTSIDE_ARG_FIELDS_AD6066D1 = + "Numeric row 的 null bitmap 在 %d 个 FIELD 范围外存在置位"; + public static final String EXCEPTION_NUMERIC_ROW_PAYLOAD_HAS_ARG_TRAILING_BYTES_89E46A34 = + "Numeric row 数据存在 %d 个多余的尾随字节"; // --- Common --- diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfo.java index 6d5dc1fce4b..938afbdc852 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfo.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfo.java @@ -205,6 +205,11 @@ public final class PreparedNumericRowInfo { } public InsertNumericRowStatement bind(long timestamp, ByteBuffer tags, ByteBuffer values) { + return bind(timestamp, tags, values, null); + } + + public InsertNumericRowStatement bind( + long timestamp, ByteBuffer tags, ByteBuffer values, ByteBuffer nullBitmap) { if (tags == null) { throw invalid( DataNodeMiscMessages.EXCEPTION_PREPARED_NUMERIC_ROW_TAG_BUFFER_IS_NULL_3FF27BF7); @@ -213,11 +218,21 @@ public final class PreparedNumericRowInfo { throw invalid( DataNodeMiscMessages.EXCEPTION_PREPARED_NUMERIC_ROW_VALUE_BUFFER_IS_NULL_A928F919); } - if (values.remaining() != valueBytes) { + NumericRowValues.validateNullBitmap(nullBitmap, types.length - tagCount); + // Preserve the cached fixed-width length for the common non-null path. + int expectedBytes = valueBytes; + if (nullBitmap != null) { + for (int i = tagCount; i < types.length; i++) { + if (NumericRowValues.isNullField(nullBitmap, i - tagCount)) { + expectedBytes -= types[i].getDataTypeSize(); + } + } + } + if (values.remaining() != expectedBytes) { throw invalid( DataNodeMiscMessages .EXCEPTION_PREPARED_NUMERIC_ROW_VALUE_BUFFER_SIZE_MISMATCH_EXPECTED_ARG_BYTES_BUT_GOT_ARG_B75D48C8, - valueBytes, + expectedBytes, values.remaining()); } TimestampPrecisionUtils.checkTimestampPrecision(timestamp); @@ -252,7 +267,11 @@ public final class PreparedNumericRowInfo { tags.remaining()); } for (int i = tagCount; i < types.length; i++) { - row.read(i, types[i], values); + if (NumericRowValues.isNullField(nullBitmap, i - tagCount)) { + row.setNull(i); + } else { + row.read(i, types[i], values); + } } } catch (BufferUnderflowException e) { throw invalid( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java index d48aef4cbf6..f74ed0b8a58 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java @@ -2620,7 +2620,8 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { .bind( req.getTimestamp(), ByteBuffer.wrap(req.getTagValues()), - ByteBuffer.wrap(req.getValues())); + ByteBuffer.wrap(req.getValues()), + req.bufferForNullBitmap()); long queryId = SESSION_MANAGER.requestQueryId(); return COORDINATOR.executeForTableModel( statement, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGenerator.java index 3aac7cfaccb..eb8cebad679 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGenerator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGenerator.java @@ -328,7 +328,8 @@ public class StatementGenerator { } insertStatement.setColumnCategories(columnCategories); if (insertRecordReq.isIsNumericRow()) { - ((InsertNumericRowStatement) insertStatement).fillValues(insertRecordReq.values); + ((InsertNumericRowStatement) insertStatement) + .fillValues(insertRecordReq.values, insertRecordReq.bufferForNullBitmap()); for (int i = 0; i < columnCategories.length; i++) { TSDataType type = insertStatement.getDataType(i); if (!(columnCategories[i] == TsTableColumnCategory.TAG && type == TSDataType.STRING) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertNumericRowStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertNumericRowStatement.java index 7093b1615fb..98e255de911 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertNumericRowStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertNumericRowStatement.java @@ -100,6 +100,10 @@ public class InsertNumericRowStatement extends InsertBaseStatement { /** RPC cells carry typed primitive FIELD bits or a length-prefixed STRING TAG. */ public void fillValues(ByteBuffer buffer) throws QueryProcessException { + fillValues(buffer, null); + } + + public void fillValues(ByteBuffer buffer, ByteBuffer nullBitmap) throws QueryProcessException { dataTypes = new TSDataType[measurements.length]; for (int i = 0; i < dataTypes.length; i++) { byte type = buffer.get(); @@ -124,6 +128,7 @@ public class InsertNumericRowStatement extends InsertBaseStatement { } } int tagCount = countTagColumns(columnCategories); + NumericRowValues.validateNullBitmap(nullBitmap, measurements.length - tagCount); numericValues = new NumericRowValues(tagCount, measurements.length - tagCount); ColumnCategory[] categories = new ColumnCategory[measurements.length]; for (int i = 0; i < measurements.length; i++) { @@ -140,9 +145,19 @@ public class InsertNumericRowStatement extends InsertBaseStatement { } for (int i = 0; i < dataTypes.length; i++) { if (columnCategories[i] == TsTableColumnCategory.FIELD) { - numericValues.read(i, dataTypes[i], buffer); + if (NumericRowValues.isNullField(nullBitmap, i - tagCount)) { + numericValues.setNull(i); + } else { + numericValues.read(i, dataTypes[i], buffer); + } } } + if (buffer.hasRemaining()) { + throw new IllegalArgumentException( + String.format( + DataNodeQueryMessages.EXCEPTION_NUMERIC_ROW_PAYLOAD_HAS_ARG_TRAILING_BYTES_89E46A34, + buffer.remaining())); + } } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/NumericRowValues.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/NumericRowValues.java index 8b9a03a44cc..1638f0fd00d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/NumericRowValues.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/NumericRowValues.java @@ -161,6 +161,36 @@ public final class NumericRowValues { }; } + /** Validates the optional FIELD-only RPC bitmap without changing its cursor. */ + public static void validateNullBitmap(ByteBuffer bitmap, int fieldCount) { + if (bitmap == null) { + return; + } + int length = (fieldCount + 7) / 8; + if (bitmap.remaining() != length) { + throw new IllegalArgumentException( + String.format( + DataNodeQueryMessages + .EXCEPTION_NUMERIC_ROW_NULL_BITMAP_LENGTH_MISMATCH_EXPECTED_ARG_BYTES_FOR_ARG_FIELDS_BUT_GOT_ARG_31F1DB42, + length, + fieldCount, + bitmap.remaining())); + } + if (fieldCount % 8 != 0 + && (bitmap.get(bitmap.limit() - 1) & ~((1 << (fieldCount % 8)) - 1)) != 0) { + throw new IllegalArgumentException( + String.format( + DataNodeQueryMessages + .EXCEPTION_NUMERIC_ROW_NULL_BITMAP_HAS_SET_BITS_OUTSIDE_ARG_FIELDS_AD6066D1, + fieldCount)); + } + } + + public static boolean isNullField(ByteBuffer bitmap, int fieldIndex) { + return bitmap != null + && (bitmap.get(bitmap.position() + fieldIndex / 8) & (1 << (fieldIndex % 8))) != 0; + } + public Object getObject(int index, TSDataType type) { if (isNull(index)) { return null; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfoTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfoTest.java index e195d69660c..e8e6ef31153 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfoTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfoTest.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.protocol.session; +import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.queryengine.common.SessionInfo; import org.apache.iotdb.commons.schema.table.TreeViewSchema; import org.apache.iotdb.commons.schema.table.TsTable; @@ -29,12 +30,14 @@ import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager; import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType; import org.apache.iotdb.db.queryengine.plan.analyze.schema.SchemaValidator; +import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator; import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableHeaderSchemaValidator; import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertNumericRow; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertNumericRowStatement; import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq; import org.apache.iotdb.service.rpc.thrift.TSPrepareNumericRowReq; import org.apache.tsfile.enums.ColumnCategory; @@ -45,14 +48,17 @@ import org.junit.Test; import java.nio.ByteBuffer; import java.time.ZoneId; +import java.util.ArrayList; import java.util.Arrays; import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyBoolean; @@ -191,6 +197,75 @@ public class PreparedNumericRowInfoTest { assertNull(statement.getPreparedTableVersion()); } + // The optional bitmap uses FIELD indices, not measurement indices, and accepts sliced buffers. + @Test + public void testNullBitmapDecodingAndValidation() throws Exception { + TSPrepareNumericRowReq req = request("tag", "v"); + req.setMeasurements(new ArrayList<>(req.getMeasurements())); + req.setDataTypes(new ArrayList<>(req.getDataTypes())); + req.setColumnCategories(new ArrayList<>(req.getColumnCategories())); + for (int i = 1; i < 9; i++) { + req.addToMeasurements("v" + i); + req.addToDataTypes(TSDataType.DOUBLE.ordinal()); + req.addToColumnCategories((byte) ColumnCategory.FIELD.ordinal()); + } + PreparedNumericRowInfo info = new PreparedNumericRowInfo(DATABASE, req); + ByteBuffer bitmap = ByteBuffer.wrap(new byte[] {99, (byte) 0x81, 1, 99}); + bitmap.position(1).limit(3); + ByteBuffer values = ByteBuffer.allocate(6 * 8); + for (int i = 1; i <= 6; i++) { + values.putLong(i == 1 ? 0x7ff8000000001234L : Double.doubleToRawLongBits(i)); + } + values.flip(); + InsertNumericRowStatement prepared = info.bind(1, tagBuffer(1), values, bitmap); + assertEquals(1, bitmap.position()); + assertEquals(0, values.position()); + ByteBuffer payload = ByteBuffer.allocate(10 + 5 + values.remaining()); + payload.put(TSDataType.STRING.serialize()); + for (int i = 0; i < 9; i++) { + payload.put(TSDataType.DOUBLE.serialize()); + } + payload.put(tagBuffer(1)).put(values.duplicate()).flip(); + TSInsertRecordReq insertReq = new TSInsertRecordReq(); + insertReq + .setPrefixPath("t") + .setTimestamp(1) + .setMeasurements(req.getMeasurements()) + .setColumnCategoryies(req.getColumnCategories()) + .setIsNumericRow(true) + .setIsWriteToTable(true) + .setValues(payload) + .setNullBitmap(bitmap); + InsertNumericRowStatement direct = + (InsertNumericRowStatement) StatementGenerator.createStatement(insertReq); + assertEquals(prepared.getNumericValues(), direct.getNumericValues()); + assertTrue(prepared.getNumericValues().isNull(1)); + assertTrue(prepared.getNumericValues().isNull(8)); + assertTrue(prepared.getNumericValues().isNull(9)); + assertFalse(prepared.getNumericValues().isNull(0)); + assertEquals(0x7ff8000000001234L, prepared.getNumericValues().get(2)); + for (byte[] bad : new byte[][] {new byte[0], new byte[1], new byte[3], {0, 2}}) { + assertThrows( + IllegalArgumentException.class, + () -> info.bind(1, tagBuffer(1), values, ByteBuffer.wrap(bad))); + insertReq.setValues(payload.duplicate().rewind()).setNullBitmap(bad); + assertThrows( + IllegalArgumentException.class, () -> StatementGenerator.createStatement(insertReq)); + } + assertThrows( + SemanticException.class, + () -> info.bind(1, tagBuffer(1), ByteBuffer.allocate(9 * 8), bitmap)); + ByteBuffer allNulls = ByteBuffer.wrap(new byte[] {(byte) 255, 1}); + insertReq.setValues(payload.duplicate().rewind()).setNullBitmap(allNulls); + assertThrows( + IllegalArgumentException.class, () -> StatementGenerator.createStatement(insertReq)); + InsertNumericRowStatement allNull = + info.bind(1, tagBuffer(1), ByteBuffer.allocate(0), allNulls); + for (int i = 1; i <= 9; i++) { + assertTrue(allNull.getNumericValues().isNull(i)); + } + } + private static void install(TsTable table) { DataNodeTableCache.getInstance().preUpdateTable(DATABASE, table, null); DataNodeTableCache.getInstance().commitUpdateTable(DATABASE, "t", null); diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/client.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/client.thrift index c08959e5fe4..edcf8834804 100644 --- a/iotdb-protocol/thrift-datanode/src/main/thrift/client.thrift +++ b/iotdb-protocol/thrift-datanode/src/main/thrift/client.thrift @@ -250,6 +250,9 @@ struct TSInsertRecordReq { 8: optional list<byte> columnCategoryies // Numeric table rows retain primitive storage throughout the server write path. 9: optional bool isNumericRow + // Numeric FIELDs only, in FIELD occurrence order: bit i (LSB first) means null. + // Exactly ceil(FIELD count / 8) bytes, unused high bits zero; null FIELD payloads are omitted. + 10: optional binary nullBitmap } struct TSInsertStringRecordReq { @@ -735,4 +738,6 @@ struct TSExecutePreparedNumericRowReq { 4: required binary values // Length-prefixed UTF-8 TAGs; length -1 represents a null TAG. 5: required binary tagValues + // Same FIELD-only bitmap and omitted-null-payload encoding as TSInsertRecordReq. + 6: optional binary nullBitmap }
