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 0d296b082d493d3e5a4a7d6361ffe163333640fe Author: Tian Jiang <[email protected]> AuthorDate: Tue Sep 15 14:54:57 2026 +0800 Add insert numeric row interface --- .../IoTDBNumericRowInsertPerformanceIT.java | 220 +++++--- .../java/org/apache/iotdb/session/Session.java | 37 +- .../apache/iotdb/session/util/SessionUtils.java | 15 +- .../java/org/apache/iotdb/session/SessionTest.java | 36 +- .../iotdb/session/util/SessionUtilsTest.java | 31 +- .../plan/parser/StatementGenerator.java | 4 +- .../node/write/RelationalInsertNumericRowNode.java | 590 +++++++++++++++++++++ .../statement/crud/InsertNumericRowStatement.java | 71 ++- .../iotdb/db/storageengine/StorageEngine.java | 4 +- .../apache/iotdb/db/utils/NumericRowValues.java | 307 +++++++++++ .../plan/parser/NumericTableRowTest.java | 524 ++++++++++++++++++ 11 files changed, 1732 insertions(+), 107 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/performance/IoTDBNumericRowInsertPerformanceIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/performance/IoTDBNumericRowInsertPerformanceIT.java index 84a52da1451..e2d6a481743 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/performance/IoTDBNumericRowInsertPerformanceIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/performance/IoTDBNumericRowInsertPerformanceIT.java @@ -48,14 +48,19 @@ import java.util.Collections; import java.util.List; import java.util.Locale; import java.util.Properties; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; /** * Manual end-to-end benchmark for the numeric table-row API and relational {@link Tablet} writes. * * <p>Enable this test with {@code -Diotdb.numeric.row.insert.perf.enabled=true}. The timed region - * includes the Session call and server write confirmation. Schema creation and client-side input - * preparation are outside the timed region. By default both APIs write one row per request; set - * {@code iotdb.numeric.row.insert.perf.tablet.row.count} to compare against batched Tablet writes. + * includes the concurrent Session calls and server write confirmations. Schema creation and + * client-side input preparation are outside the timed region. By default 16 client sessions write + * one row per request; set {@code iotdb.numeric.row.insert.perf.client.count} or {@code + * iotdb.numeric.row.insert.perf.tablet.row.count} to vary concurrency or Tablet batching. * * <p>Results are written to {@code target/numeric-row-insert-performance.properties}. Percentiles * describe elapsed time per complete round, not individual RPC latency. See the adjacent README for @@ -69,13 +74,15 @@ public class IoTDBNumericRowInsertPerformanceIT { private static final String ENABLED_PROPERTY = PROPERTY_PREFIX + "enabled"; private static final String ROW_COUNT_PROPERTY = PROPERTY_PREFIX + "row.count"; private static final String COLUMN_COUNT_PROPERTY = PROPERTY_PREFIX + "column.count"; + private static final String CLIENT_COUNT_PROPERTY = PROPERTY_PREFIX + "client.count"; private static final String TABLET_ROW_COUNT_PROPERTY = PROPERTY_PREFIX + "tablet.row.count"; private static final String WARMUP_ROUND_COUNT_PROPERTY = PROPERTY_PREFIX + "warmup.round.count"; private static final String ROUND_COUNT_PROPERTY = PROPERTY_PREFIX + "round.count"; private static final String RESULT_PATH_PROPERTY = PROPERTY_PREFIX + "result.path"; private static final int ROW_COUNT = Integer.getInteger(ROW_COUNT_PROPERTY, 10_000); - private static final int COLUMN_COUNT = Integer.getInteger(COLUMN_COUNT_PROPERTY, 8); + private static final int COLUMN_COUNT = Integer.getInteger(COLUMN_COUNT_PROPERTY, 100); + private static final int CLIENT_COUNT = Integer.getInteger(CLIENT_COUNT_PROPERTY, 16); private static final int TABLET_ROW_COUNT = Integer.getInteger(TABLET_ROW_COUNT_PROPERTY, 1); private static final int WARMUP_ROUND_COUNT = Integer.getInteger(WARMUP_ROUND_COUNT_PROPERTY, 1); private static final int ROUND_COUNT = Integer.getInteger(ROUND_COUNT_PROPERTY, 5); @@ -119,27 +126,41 @@ public class IoTDBNumericRowInsertPerformanceIT { final long[] numericRowElapsedNanos = new long[ROUND_COUNT]; final long[] tabletElapsedNanos = new long[ROUND_COUNT]; - try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) { - session.executeNonQueryStatement("CREATE DATABASE " + DATABASE); - session.executeNonQueryStatement("USE " + DATABASE); + final List<ITableSession> sessions = new ArrayList<>(CLIENT_COUNT); + final ExecutorService executor = Executors.newFixedThreadPool(CLIENT_COUNT); + try { + final ITableSession setupSession = EnvFactory.getEnv().getTableSessionConnection(); + setupSession.executeNonQueryStatement("CREATE DATABASE " + DATABASE); + setupSession.executeNonQueryStatement("USE " + DATABASE); + sessions.add(setupSession); + for (int client = 1; client < CLIENT_COUNT; client++) { + final ITableSession session = EnvFactory.getEnv().getTableSessionConnection(); + session.executeNonQueryStatement("USE " + DATABASE); + sessions.add(session); + } final int totalRoundCount = WARMUP_ROUND_COUNT + ROUND_COUNT; for (int round = 0; round < totalRoundCount; round++) { - final NumericRowWorkload numericRowWorkload = prepareNumericRowWorkload(session, round); - final TabletWorkload tabletWorkload = prepareTabletWorkload(session, round); + final List<NumericRowWorkload> numericRowWorkloads = + prepareNumericRowWorkloads(setupSession, round); + final List<TabletWorkload> tabletWorkloads = prepareTabletWorkloads(setupSession, round); final long numericRowNanos; final long tabletNanos; // Alternate execution order so one path does not consistently receive a warmer server. if ((round & 1) == 0) { - numericRowNanos = executeNumericRowWorkload(session, numericRowWorkload); - tabletNanos = executeTabletWorkload(session, tabletWorkload); + numericRowNanos = executeNumericRowWorkload(sessions, numericRowWorkloads, executor); + tabletNanos = executeTabletWorkload(sessions, tabletWorkloads, executor); } else { - tabletNanos = executeTabletWorkload(session, tabletWorkload); - numericRowNanos = executeNumericRowWorkload(session, numericRowWorkload); + tabletNanos = executeTabletWorkload(sessions, tabletWorkloads, executor); + numericRowNanos = executeNumericRowWorkload(sessions, numericRowWorkloads, executor); + } + for (NumericRowWorkload workload : numericRowWorkloads) { + assertRowCount(setupSession, workload.tableName); + } + for (TabletWorkload workload : tabletWorkloads) { + assertRowCount(setupSession, workload.tableName); } - assertRowCount(session, numericRowWorkload.tableName); - assertRowCount(session, tabletWorkload.tableName); if (round >= WARMUP_ROUND_COUNT) { final int measuredRound = round - WARMUP_ROUND_COUNT; @@ -147,79 +168,138 @@ public class IoTDBNumericRowInsertPerformanceIT { tabletElapsedNanos[measuredRound] = tabletNanos; } } + } finally { + executor.shutdownNow(); + for (ITableSession session : sessions) { + session.close(); + } } final BenchmarkResult result = new BenchmarkResult(numericRowElapsedNanos, tabletElapsedNanos); writeResult(result); } - private static NumericRowWorkload prepareNumericRowWorkload(ITableSession session, int round) - throws Exception { - final String tableName = "numeric_row_" + round; - createTable(session, tableName); - final long startTimestamp = (long) round * ROW_COUNT; - final NumericRow[] rows = new NumericRow[ROW_COUNT]; - final String[] tagValues = new String[0]; - for (int row = 0; row < ROW_COUNT; row++) { - rows[row] = - new NumericRow( - tableName, - startTimestamp + row, - columnNames, - dataTypes, - columnCategories, - createRawValues(startTimestamp + row), - tagValues); + private static List<NumericRowWorkload> prepareNumericRowWorkloads( + ITableSession session, int round) throws Exception { + final List<NumericRowWorkload> workloads = new ArrayList<>(CLIENT_COUNT); + for (int client = 0; client < CLIENT_COUNT; client++) { + final String tableName = "numeric_row_" + round + "_" + client; + createTable(session, tableName); + final long startTimestamp = ((long) round * CLIENT_COUNT + client) * ROW_COUNT; + final NumericRow[] rows = new NumericRow[ROW_COUNT]; + final String[] tagValues = new String[0]; + for (int row = 0; row < ROW_COUNT; row++) { + rows[row] = + new NumericRow( + tableName, + startTimestamp + row, + columnNames, + dataTypes, + columnCategories, + createRawValues(startTimestamp + row), + tagValues); + } + workloads.add(new NumericRowWorkload(tableName, rows)); } - return new NumericRowWorkload(tableName, rows); + return workloads; } - private static TabletWorkload prepareTabletWorkload(ITableSession session, int round) + private static List<TabletWorkload> prepareTabletWorkloads(ITableSession session, int round) throws Exception { - final String tableName = "tablet_" + round; - createTable(session, tableName); - final long startTimestamp = (long) round * ROW_COUNT; - final List<Tablet> tablets = new ArrayList<>((ROW_COUNT - 1) / TABLET_ROW_COUNT + 1); - for (int rowStart = 0; rowStart < ROW_COUNT; ) { - final int rowCount = Math.min(TABLET_ROW_COUNT, ROW_COUNT - rowStart); - final Tablet tablet = - new Tablet(tableName, columnNames, dataTypes, columnCategories, rowCount); - for (int row = 0; row < rowCount; row++) { - final int rowIndex = tablet.getRowSize(); - final long timestamp = startTimestamp + rowStart + row; - final long[] values = createRawValues(timestamp); - tablet.addTimestamp(rowIndex, timestamp); - for (int column = 0; column < COLUMN_COUNT; column++) { - tablet.addValue( - columnNames.get(column), - rowIndex, - toTabletValue(dataTypes.get(column), values[column])); + final List<TabletWorkload> workloads = new ArrayList<>(CLIENT_COUNT); + for (int client = 0; client < CLIENT_COUNT; client++) { + final String tableName = "tablet_" + round + "_" + client; + createTable(session, tableName); + final long startTimestamp = ((long) round * CLIENT_COUNT + client) * ROW_COUNT; + final List<Tablet> tablets = new ArrayList<>((ROW_COUNT - 1) / TABLET_ROW_COUNT + 1); + for (int rowStart = 0; rowStart < ROW_COUNT; ) { + final int rowCount = Math.min(TABLET_ROW_COUNT, ROW_COUNT - rowStart); + final Tablet tablet = + new Tablet(tableName, columnNames, dataTypes, columnCategories, rowCount); + for (int row = 0; row < rowCount; row++) { + final int rowIndex = tablet.getRowSize(); + final long timestamp = startTimestamp + rowStart + row; + final long[] values = createRawValues(timestamp); + tablet.addTimestamp(rowIndex, timestamp); + for (int column = 0; column < COLUMN_COUNT; column++) { + tablet.addValue( + columnNames.get(column), + rowIndex, + toTabletValue(dataTypes.get(column), values[column])); + } } + tablets.add(tablet); + rowStart += rowCount; } - tablets.add(tablet); - rowStart += rowCount; + workloads.add(new TabletWorkload(tableName, tablets)); } - return new TabletWorkload(tableName, tablets); + return workloads; } private static long executeNumericRowWorkload( - ITableSession session, NumericRowWorkload numericRowWorkload) throws Exception { - final long startNanos = System.nanoTime(); - for (final NumericRow row : numericRowWorkload.rows) { - session.insert(row); - } - return System.nanoTime() - startNanos; + List<ITableSession> sessions, List<NumericRowWorkload> workloads, ExecutorService executor) + throws Exception { + return executeConcurrently( + sessions, + workloads, + executor, + (session, workload) -> { + for (NumericRow row : workload.rows) { + session.insert(row); + } + }); } - private static long executeTabletWorkload(ITableSession session, TabletWorkload tabletWorkload) + private static long executeTabletWorkload( + List<ITableSession> sessions, List<TabletWorkload> workloads, ExecutorService executor) throws Exception { + return executeConcurrently( + sessions, + workloads, + executor, + (session, workload) -> { + for (Tablet tablet : workload.tablets) { + session.insert(tablet); + } + }); + } + + private static <T> long executeConcurrently( + List<ITableSession> sessions, + List<T> workloads, + ExecutorService executor, + WorkloadWriter<T> writer) + throws Exception { + final CountDownLatch ready = new CountDownLatch(CLIENT_COUNT); + final CountDownLatch start = new CountDownLatch(1); + final List<Future<?>> futures = new ArrayList<>(CLIENT_COUNT); + for (int client = 0; client < CLIENT_COUNT; client++) { + final ITableSession session = sessions.get(client); + final T workload = workloads.get(client); + futures.add( + executor.submit( + () -> { + ready.countDown(); + start.await(); + writer.write(session, workload); + return null; + })); + } + ready.await(); final long startNanos = System.nanoTime(); - for (final Tablet tablet : tabletWorkload.tablets) { - session.insert(tablet); + start.countDown(); + for (Future<?> future : futures) { + future.get(); } return System.nanoTime() - startNanos; } + @FunctionalInterface + private interface WorkloadWriter<T> { + + void write(ITableSession session, T workload) throws Exception; + } + private static void createTable(ITableSession session, String tableName) throws Exception { final StringBuilder statement = new StringBuilder("CREATE TABLE "); statement.append(tableName).append('('); @@ -301,6 +381,7 @@ public class IoTDBNumericRowInsertPerformanceIT { private static void assertValidConfiguration() { Assert.assertTrue(ROW_COUNT > 0); Assert.assertTrue(COLUMN_COUNT > 0); + Assert.assertTrue(CLIENT_COUNT > 0); Assert.assertTrue(TABLET_ROW_COUNT > 0); Assert.assertTrue(WARMUP_ROUND_COUNT >= 0); Assert.assertTrue(ROUND_COUNT > 0); @@ -318,17 +399,18 @@ public class IoTDBNumericRowInsertPerformanceIT { Integer.toString(Runtime.getRuntime().availableProcessors())); properties.setProperty( "client.max.heap.bytes", Long.toString(Runtime.getRuntime().maxMemory())); - properties.setProperty("client.count", "1"); + properties.setProperty("client.count", Integer.toString(CLIENT_COUNT)); properties.setProperty("row.count", Integer.toString(ROW_COUNT)); properties.setProperty("column.count", Integer.toString(COLUMN_COUNT)); properties.setProperty("tablet.row.count", Integer.toString(TABLET_ROW_COUNT)); properties.setProperty("warmup.round.count", Integer.toString(WARMUP_ROUND_COUNT)); properties.setProperty("round.count", Integer.toString(ROUND_COUNT)); properties.setProperty("column.types", join(dataTypes)); - properties.setProperty("numeric.row.requests.per.round", Integer.toString(ROW_COUNT)); + properties.setProperty( + "numeric.row.requests.per.round", Integer.toString(CLIENT_COUNT * ROW_COUNT)); properties.setProperty( "insert.tablet.requests.per.round", - Integer.toString((ROW_COUNT - 1) / TABLET_ROW_COUNT + 1)); + Integer.toString(CLIENT_COUNT * ((ROW_COUNT - 1) / TABLET_ROW_COUNT + 1))); properties.setProperty("numeric.row.round.elapsed.nanos", join(result.numericRowElapsedNanos)); properties.setProperty("insert.tablet.round.elapsed.nanos", join(result.tabletElapsedNanos)); properties.setProperty( @@ -430,7 +512,7 @@ public class IoTDBNumericRowInsertPerformanceIT { } private double numericRowRowsPerSecond() { - return ROW_COUNT * 1_000_000_000.0D / numericRowP50Nanos(); + return CLIENT_COUNT * ROW_COUNT * 1_000_000_000.0D / numericRowP50Nanos(); } private double numericRowValuesPerSecond() { @@ -446,7 +528,7 @@ public class IoTDBNumericRowInsertPerformanceIT { } private double tabletRowsPerSecond() { - return ROW_COUNT * 1_000_000_000.0D / tabletP50Nanos(); + return CLIENT_COUNT * ROW_COUNT * 1_000_000_000.0D / tabletP50Nanos(); } private double tabletValuesPerSecond() { 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 716d14fa802..835cf924777 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 @@ -2890,18 +2890,49 @@ public class Session implements ISession { List<ColumnCategory> columnCategories = row.getColumnCategories(); long[] values = row.getValues(); String[] tagValues = row.getTagValues(); + if (measurements == null + || types == null + || columnCategories == null + || values == null + || tagValues == null + || measurements.size() != types.size() + || measurements.size() != columnCategories.size()) { + throw new IllegalArgumentException( + SessionMessages + .EXCEPTION_NUMERIC_ROW_ARGUMENTS_MUST_BE_NON_NULL_COLUMN_METADATA_LENGTHS_AND_PER_CATEGORY_VALUE_COUNTS_MUST_MATCH_EB12B233); + } + // Keep wire metadata aligned with the compact payload: TAG columns precede FIELD columns. + List<String> rpcMeasurements = new ArrayList<>(measurements.size()); + List<TSDataType> rpcTypes = new ArrayList<>(types.size()); + List<ColumnCategory> rpcColumnCategories = new ArrayList<>(columnCategories.size()); + for (int pass = 0; pass < 2; pass++) { + ColumnCategory category = pass == 0 ? ColumnCategory.TAG : ColumnCategory.FIELD; + for (int i = 0; i < columnCategories.size(); i++) { + if (columnCategories.get(i) == category) { + rpcMeasurements.add(measurements.get(i)); + rpcTypes.add(types.get(i)); + rpcColumnCategories.add(category); + } + } + } + if (rpcMeasurements.size() != measurements.size()) { + throw new IllegalArgumentException( + SessionMessages + .EXCEPTION_NUMERIC_ROWS_SUPPORT_NUMERIC_FIELD_COLUMNS_AND_STRING_TAG_COLUMNS_ONLY_87159FFA); + } ByteBuffer valueBuffer = - SessionUtils.getValueBuffer(types, values, measurements, columnCategories, tagValues); + SessionUtils.getValueBuffer( + rpcTypes, values, rpcMeasurements, rpcColumnCategories, tagValues); TSInsertRecordReq request = new TSInsertRecordReq(); request.setPrefixPath(tableName); request.setTimestamp(timestamp); - request.setMeasurements(measurements); + request.setMeasurements(rpcMeasurements); request.setValues(valueBuffer); request.setIsWriteToTable(true); request.setIsNumericRow(true); request.setIsAligned(true); request.setColumnCategoryies( - columnCategories.stream() + rpcColumnCategories.stream() .map(category -> (byte) category.ordinal()) .collect(Collectors.toList())); String[] deviceSegments = new String[tagValues.length + 1]; 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 0d838b69254..40fa0fae41f 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 @@ -211,13 +211,20 @@ public class SessionUtils { ByteBuffer buffer = ByteBuffer.allocate(length); int fieldIndex = 0; int tagIndex = 0; + for (TSDataType type : types) { + ReadWriteIOUtils.write(type, buffer); + } + tagIndex = 0; for (int i = 0; i < types.size(); i++) { TSDataType type = types.get(i); - ReadWriteIOUtils.write(type, buffer); - if (categories.get(i) == ColumnCategory.TAG) { - // A length of -1 represents a null TAG; an empty TAG has length zero. + if (categories.get(i) == ColumnCategory.TAG && type == TSDataType.STRING) { ReadWriteIOUtils.write(tagValues[tagIndex++], buffer); - } else { + } + } + fieldIndex = 0; + for (int i = 0; i < types.size(); i++) { + TSDataType type = types.get(i); + if (categories.get(i) == ColumnCategory.FIELD) { long value = values[fieldIndex++]; switch (type) { case BOOLEAN -> buffer.put((byte) (value == 0 ? 0 : 1)); diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionTest.java index 88280ac8b4b..9f79ee200d9 100644 --- a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionTest.java +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionTest.java @@ -194,23 +194,47 @@ public class SessionTest { public void testNumericTableRowWithTagsRequest() throws Exception { // The NumericRow entry point must forward the TAG payload and metadata with numeric flags. ITableSession tableSession = new TableSession((Session) session); - List<String> names = Arrays.asList("tag", "enabled"); - List<TSDataType> types = Arrays.asList(TSDataType.STRING, TSDataType.BOOLEAN); - List<ColumnCategory> categories = Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD); + List<String> names = Arrays.asList("enabled", "tag", "value", "region"); + List<TSDataType> types = + Arrays.asList(TSDataType.BOOLEAN, TSDataType.STRING, TSDataType.DOUBLE, TSDataType.STRING); + List<ColumnCategory> categories = + Arrays.asList( + ColumnCategory.FIELD, ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.TAG); + List<String> rpcNames = Arrays.asList("tag", "region", "enabled", "value"); + List<TSDataType> rpcTypes = + Arrays.asList(TSDataType.STRING, TSDataType.STRING, TSDataType.BOOLEAN, TSDataType.DOUBLE); + List<ColumnCategory> rpcCategories = + Arrays.asList( + ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.FIELD); tableSession.insert( new NumericRow( - "readings", 123L, names, types, categories, new long[] {1}, new String[] {"device"})); + "readings", + 123L, + names, + types, + categories, + new long[] {1, Double.doubleToRawLongBits(2.5)}, + new String[] {"device", "cn"})); ArgumentCaptor<TSInsertRecordReq> captor = ArgumentCaptor.forClass(TSInsertRecordReq.class); Mockito.verify(sessionConnection).insertRecord(captor.capture()); TSInsertRecordReq request = captor.getValue(); Assert.assertTrue(request.isIsNumericRow()); Assert.assertTrue(request.isIsWriteToTable()); assertEquals( - Arrays.asList((byte) ColumnCategory.TAG.ordinal(), (byte) ColumnCategory.FIELD.ordinal()), + Arrays.asList( + (byte) ColumnCategory.TAG.ordinal(), + (byte) ColumnCategory.TAG.ordinal(), + (byte) ColumnCategory.FIELD.ordinal(), + (byte) ColumnCategory.FIELD.ordinal()), request.getColumnCategoryies()); + assertEquals(rpcNames, request.getMeasurements()); assertEquals( SessionUtils.getValueBuffer( - types, new long[] {1}, names, categories, new String[] {"device"}), + rpcTypes, + new long[] {1, Double.doubleToRawLongBits(2.5)}, + rpcNames, + rpcCategories, + new String[] {"device", "cn"}), request.bufferForValues()); } 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 d2c24a4da8e..dfb1ef4a2bf 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 @@ -77,14 +77,17 @@ public class SessionUtilsTest { Collections.nCopies(8, ColumnCategory.FIELD), new String[0]); ByteBuffer expected = ByteBuffer.allocate(46); - expected.put(TSDataType.BOOLEAN.serialize()).put((byte) 0); - expected.put(TSDataType.BOOLEAN.serialize()).put((byte) 1); - expected.put(TSDataType.INT32.serialize()).putInt(Integer.MIN_VALUE); - expected.put(TSDataType.INT64.serialize()).putLong(Long.MIN_VALUE); - expected.put(TSDataType.FLOAT.serialize()).putInt(0x7fc01234); - expected.put(TSDataType.DOUBLE.serialize()).putLong(0x7ff8000000001234L); - expected.put(TSDataType.DATE.serialize()).putInt(20260910); - expected.put(TSDataType.TIMESTAMP.serialize()).putLong(Long.MAX_VALUE); + for (TSDataType type : types) { + expected.put(type.serialize()); + } + expected.put((byte) 0); + expected.put((byte) 1); + expected.putInt(Integer.MIN_VALUE); + expected.putLong(Long.MIN_VALUE); + expected.putInt(0x7fc01234); + expected.putLong(0x7ff8000000001234L); + expected.putInt(20260910); + expected.putLong(Long.MAX_VALUE); expected.flip(); Assert.assertEquals(expected, actual); } @@ -154,16 +157,14 @@ public class SessionUtilsTest { Arrays.asList("a", "d", "b", "enabled", "c"), categories, new String[] {"设备😀", null, ""}); - Assert.assertEquals(TSDataType.STRING.serialize(), buffer.get()); + for (TSDataType type : types) { + Assert.assertEquals(type.serialize(), buffer.get()); + } Assert.assertEquals("设备😀", ReadWriteIOUtils.readString(buffer)); - Assert.assertEquals(TSDataType.DOUBLE.serialize(), buffer.get()); + Assert.assertNull(ReadWriteIOUtils.readString(buffer)); + Assert.assertEquals("", ReadWriteIOUtils.readString(buffer)); Assert.assertEquals(0x7ff8000000001234L, buffer.getLong()); - Assert.assertEquals(TSDataType.STRING.serialize(), buffer.get()); - Assert.assertEquals(-1, buffer.getInt()); - Assert.assertEquals(TSDataType.BOOLEAN.serialize(), buffer.get()); Assert.assertEquals(1, buffer.get()); - Assert.assertEquals(TSDataType.STRING.serialize(), buffer.get()); - Assert.assertEquals(0, buffer.getInt()); Assert.assertFalse(buffer.hasRemaining()); } 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 0303eed2777..3aac7cfaccb 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 @@ -305,7 +305,6 @@ public class StatementGenerator { } InsertNumericRowStatement numeric = (InsertNumericRowStatement) insertStatement; numeric.setTime(insertRecordReq.getTimestamp()); - numeric.fillValues(insertRecordReq.values); insertStatement.setAligned(true); } else { InsertRowStatement row = (InsertRowStatement) insertStatement; @@ -327,7 +326,9 @@ public class StatementGenerator { TsTableColumnCategory.fromTsFileColumnCategory( ColumnCategory.values()[insertRecordReq.getColumnCategoryies().get(i).intValue()]); } + insertStatement.setColumnCategories(columnCategories); if (insertRecordReq.isIsNumericRow()) { + ((InsertNumericRowStatement) insertStatement).fillValues(insertRecordReq.values); for (int i = 0; i < columnCategories.length; i++) { TSDataType type = insertStatement.getDataType(i); if (!(columnCategories[i] == TsTableColumnCategory.TAG && type == TSDataType.STRING) @@ -339,7 +340,6 @@ public class StatementGenerator { } } } - insertStatement.setColumnCategories(columnCategories); } PERFORMANCE_OVERVIEW_METRICS.recordParseCost(System.nanoTime() - startTime); return insertStatement; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertNumericRowNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertNumericRowNode.java new file mode 100644 index 00000000000..0f8a78ed914 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertNumericRowNode.java @@ -0,0 +1,590 @@ +/* + * 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.queryengine.plan.planner.plan.node.write; + +import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.IPlanVisitor; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.commons.utils.TimePartitionUtils; +import org.apache.iotdb.db.exception.DataTypeInconsistentException; +import org.apache.iotdb.db.i18n.DataNodeQueryMessages; +import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceSchemaCache; +import org.apache.iotdb.db.storageengine.dataregion.memtable.AbstractMemTable; +import org.apache.iotdb.db.storageengine.dataregion.memtable.IWritableMemChunkGroup; +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.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.utils.ReadWriteIOUtils; +import org.apache.tsfile.utils.TsPrimitiveType; +import org.apache.tsfile.write.UnSupportedDataTypeException; +import org.apache.tsfile.write.schema.MeasurementSchema; + +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +/** + * A numeric table row with its own plan type and wire format. FIELDs retain raw long bits and TAGs + * retain strings through consensus and WAL recovery. + */ +public class RelationalInsertNumericRowNode extends AbstractInsertRowNode { + private long time; + private NumericRowValues numericValues; + + public RelationalInsertNumericRowNode(PlanNodeId id) { + super(id); + isAligned = true; + } + + public RelationalInsertNumericRowNode( + PlanNodeId id, + PartialPath table, + String[] measurements, + TSDataType[] dataTypes, + MeasurementSchema[] schemas, + long time, + NumericRowValues values, + TsTableColumnCategory[] categories) { + super(id, table, true, measurements, dataTypes); + this.measurementSchemas = schemas; + this.time = time; + this.numericValues = values; + setColumnCategories(categories); + } + + @Override + public PlanNodeType getType() { + return PlanNodeType.RELATIONAL_INSERT_NUMERIC_ROW; + } + + @Override + public <R, C> R accept(IPlanVisitor<R, C> visitor, C context) { + return ((PlanVisitor<R, C>) visitor).visitRelationalInsertNumericRow(this, context); + } + + @Override + public InsertNode mergeInsertNode(List<InsertNode> nodes) { + // One numeric row has exactly one device and timestamp, so it cannot be split into fragments. + // InsertNode.merge returns the sole node before reaching this method. + throw new UnsupportedOperationException(); + } + + @Override + public long getTime() { + return time; + } + + public void setTime(long time) { + this.time = time; + } + + @Override + public long getMinTime() { + return time; + } + + public NumericRowValues getNumericValues() { + return numericValues; + } + + public long[] getValues() { + return numericValues.getValues(); + } + + public void setNumericValues(NumericRowValues values) { + numericValues = values; + deviceID = null; + } + + @Override + public Object getValuesForMemoryAccounting() { + return numericValues; + } + + @Override + public int getValueCount() { + return numericValues == null ? 0 : numericValues.size(); + } + + @Override + public boolean isNull(int index) { + return index < 0 || index >= getValueCount() || numericValues.isNull(index); + } + + @Override + public Object getValue(int index) { + return isNull(index) ? null : numericValues.getObject(index, getDataType(index)); + } + + public String getTableName() { + return targetPath.getFullPath(); + } + + @Override + public IDeviceID getDeviceID() { + if (deviceID == null) { + int tagCount = 0; + for (int i = 0; columnCategories != null && i < columnCategories.length; i++) { + if (columnCategories[i] == TsTableColumnCategory.TAG && isPresent(i)) { + tagCount++; + } + } + String[] segments = new String[tagCount + 1]; + segments[0] = getTableName(); + int tagIndex = 1; + for (int i = 0; columnCategories != null && i < columnCategories.length; i++) { + if (columnCategories[i] == TsTableColumnCategory.TAG && isPresent(i)) { + segments[tagIndex++] = numericValues.getTag(i); + } + } + deviceID = IDeviceID.Factory.DEFAULT_FACTORY.create(segments); + } + return deviceID; + } + + private boolean isPresent(int index) { + return measurements != null + && index >= 0 + && index < measurements.length + && measurements[index] != null + && index < getValueCount() + && columnCategories != null + && index < columnCategories.length + && columnCategories[index] != null + && (measurementSchemas == null + || index < measurementSchemas.length && measurementSchemas[index] != null); + } + + @Override + protected int getValidMeasurementNumber() { + int count = 0; + for (int i = 0; i < getValueCount(); i++) { + if (isPresent(i)) { + count++; + } + } + return count; + } + + @Override + public int getValidMeasurementNumber(boolean fieldOnly) { + int count = 0; + for (int i = 0; i < getValueCount(); i++) { + if (isPresent(i) && isValidMeasurement(i, fieldOnly)) { + count++; + } + } + return count; + } + + @Override + public void markFailedMeasurement(int index) { + if (!isPresent(index)) { + return; + } + measurements[index] = null; + if (dataTypes != null && index < dataTypes.length) { + dataTypes[index] = null; + } + numericValues.setNull(index); + deviceID = null; + measurementColumnCnt = -1; + } + + @Override + public List<WritePlanNode> splitByPartition(IAnalysis analysis) { + TTimePartitionSlot slot = TimePartitionUtils.getTimePartitionSlot(time); + dataRegionReplicaSet = + analysis + .getDataPartitionInfo() + .getDataRegionReplicaSetForWriting(getDeviceID(), slot, analysis.getDatabaseName()); + analysis.setRedirectNodeList( + Collections.singletonList( + dataRegionReplicaSet.getDataNodeLocations().get(0).getClientRpcEndPoint())); + return Collections.singletonList(this); + } + + @Override + public void addChild(PlanNode child) { + /* A single-row write has no children. */ + } + + @Override + public int allowedChildCount() { + return NO_CHILD_ALLOWED; + } + + @Override + public List<String> getOutputColumnNames() { + return null; + } + + @Override + public PlanNode clone() { + throw new UnsupportedOperationException(); + } + + @Override + public long getLastCacheTimestamp() { + return time; + } + + @Override + public boolean hasLastCacheValue(int index) { + return isPresent(index) && !isNull(index) && isWritableFieldMeasurement(index); + } + + @Override + public TimeValuePair getLastCacheValue(int index) { + return hasLastCacheValue(index) + ? new TimeValuePair(time, TsPrimitiveType.getByType(getDataType(index), getValue(index))) + : null; + } + + @Override + public void updateLastCache(String database) { + TableDeviceSchemaCache.getInstance() + .updateLastCacheIfExists(database, getDeviceID(), measurements, measurementSchemas, this); + } + + @Override + public void checkDataType(AbstractMemTable memTable) throws DataTypeInconsistentException { + IWritableMemChunkGroup group = memTable.getWritableMemChunkGroup(getDeviceID()); + if (group != null) { + group.checkDataType(this); + } + } + + private int bodySize() { + // Fixed body fields: timestamp, table name, number of valid columns, and schema-present flag. + int size = + Long.BYTES + ReadWriteIOUtils.sizeToWrite(getTableName()) + Integer.BYTES + Byte.BYTES; + for (int i = 0; i < getValueCount(); i++) { + if (!isPresent(i)) { + continue; + } + // Column identity: either the measurement name or the serialized measurement schema. + size += + measurementSchemas == null + ? ReadWriteIOUtils.sizeToWrite(measurements[i]) + : measurementSchemas[i].serializedSize(); + // Column data type, column category, and null marker. + size += 3 * Byte.BYTES; + if (!isNull(i)) { + // Non-null TAGs carry a length-prefixed String; numeric FIELDs carry one long value. + size += + columnCategories[i] == TsTableColumnCategory.TAG + ? ReadWriteIOUtils.sizeToWrite(numericValues.getTag(i)) + : Long.BYTES; + } + } + return size; + } + + @Override + protected int serializedAttributesSize() { + return PlanNodeType.BYTES + bodySize(); + } + + @Override + public int serializedSize() { + return PlanNodeType.BYTES + Long.BYTES + bodySize(); + } + + @Override + protected void serializeAttributes(ByteBuffer buffer) { + getType().serialize(buffer); + serializeBody(buffer); + } + + @Override + protected void serializeAttributes(DataOutputStream stream) throws IOException { + getType().serialize(stream); + serializeBody(stream); + } + + @Override + public void serializeToWAL(IWALByteBufferView buffer) { + serializeToWAL(buffer, getEncodedSearchIndex()); + } + + @Override + public void serializeToWAL(IWALByteBufferView buffer, long encodedSearchIndex) { + buffer.putShort(getType().getNodeType()); + buffer.putLong(encodedSearchIndex); + serializeBody(buffer); + } + + private void serializeBody(ByteBuffer buffer) { + ReadWriteIOUtils.write(time, buffer); + ReadWriteIOUtils.write(getTableName(), buffer); + ReadWriteIOUtils.write(getValidMeasurementNumber(), buffer); + ReadWriteIOUtils.write((byte) (measurementSchemas == null ? 0 : 1), buffer); + for (int i = 0; i < getValueCount(); i++) { + if (!isPresent(i)) { + continue; + } + if (measurementSchemas == null) { + ReadWriteIOUtils.write(measurements[i], buffer); + } else { + measurementSchemas[i].serializeTo(buffer); + } + ReadWriteIOUtils.write( + getDataType(i) == null ? (byte) -1 : getDataType(i).serialize(), buffer); + ReadWriteIOUtils.write(columnCategories[i].getCategory(), buffer); + ReadWriteIOUtils.write((byte) (isNull(i) ? 1 : 0), buffer); + if (!isNull(i)) { + if (columnCategories[i] == TsTableColumnCategory.TAG) { + ReadWriteIOUtils.write(numericValues.getTag(i), buffer); + } else { + ReadWriteIOUtils.write(numericValues.get(i), buffer); + } + } + } + } + + private void serializeBody(DataOutputStream stream) throws IOException { + ReadWriteIOUtils.write(time, stream); + ReadWriteIOUtils.write(getTableName(), stream); + ReadWriteIOUtils.write(getValidMeasurementNumber(), stream); + ReadWriteIOUtils.write((byte) (measurementSchemas == null ? 0 : 1), stream); + for (int i = 0; i < getValueCount(); i++) { + if (!isPresent(i)) { + continue; + } + if (measurementSchemas == null) { + ReadWriteIOUtils.write(measurements[i], stream); + } else { + measurementSchemas[i].serializeTo(stream); + } + ReadWriteIOUtils.write( + getDataType(i) == null ? (byte) -1 : getDataType(i).serialize(), stream); + ReadWriteIOUtils.write(columnCategories[i].getCategory(), stream); + ReadWriteIOUtils.write((byte) (isNull(i) ? 1 : 0), stream); + if (!isNull(i)) { + if (columnCategories[i] == TsTableColumnCategory.TAG) { + ReadWriteIOUtils.write(numericValues.getTag(i), stream); + } else { + ReadWriteIOUtils.write(numericValues.get(i), stream); + } + } + } + } + + private void serializeBody(IWALByteBufferView buffer) { + WALWriteUtils.write(time, buffer); + WALWriteUtils.write(getTableName(), buffer); + WALWriteUtils.write(getValidMeasurementNumber(), buffer); + WALWriteUtils.write((byte) (measurementSchemas == null ? 0 : 1), buffer); + for (int i = 0; i < getValueCount(); i++) { + if (!isPresent(i)) { + continue; + } + if (measurementSchemas == null) { + WALWriteUtils.write(measurements[i], buffer); + } else { + WALWriteUtils.write(measurementSchemas[i], buffer); + } + WALWriteUtils.write(getDataType(i) == null ? (byte) -1 : getDataType(i).serialize(), buffer); + WALWriteUtils.write(columnCategories[i].getCategory(), buffer); + WALWriteUtils.write((byte) (isNull(i) ? 1 : 0), buffer); + if (!isNull(i)) { + if (columnCategories[i] == TsTableColumnCategory.TAG) { + WALWriteUtils.write(numericValues.getTag(i), buffer); + } else { + WALWriteUtils.write(numericValues.get(i), buffer); + } + } + } + } + + private static RelationalInsertNumericRowNode deserializeBody(ByteBuffer buffer) { + RelationalInsertNumericRowNode node = new RelationalInsertNumericRowNode(new PlanNodeId("")); + node.time = buffer.getLong(); + node.targetPath = new PartialPath(ReadWriteIOUtils.readString(buffer), false); + int count = buffer.getInt(); + boolean hasSchemas = buffer.get() != 0; + node.measurements = new String[count]; + node.dataTypes = new TSDataType[count]; + node.measurementSchemas = hasSchemas ? new MeasurementSchema[count] : null; + TsTableColumnCategory[] categories = new TsTableColumnCategory[count]; + String[] tagPayload = new String[count]; + long[] valuePayload = new long[count]; + boolean[] nonNull = new boolean[count]; + for (int i = 0; i < count; i++) { + if (hasSchemas) { + node.measurementSchemas[i] = MeasurementSchema.deserializeFrom(buffer); + node.measurements[i] = node.measurementSchemas[i].getMeasurementName(); + } else { + node.measurements[i] = ReadWriteIOUtils.readString(buffer); + } + byte type = buffer.get(); + node.dataTypes[i] = type == -1 ? null : TSDataType.deserialize(type); + categories[i] = TsTableColumnCategory.deserialize(buffer); + nonNull[i] = buffer.get() == 0; + if (nonNull[i]) { + if (categories[i] == TsTableColumnCategory.TAG && node.dataTypes[i] == TSDataType.STRING) { + tagPayload[i] = ReadWriteIOUtils.readString(buffer); + } else { + if (!NumericRowValues.supports(node.dataTypes[i]) + || categories[i] != TsTableColumnCategory.FIELD) { + throw new UnSupportedDataTypeException( + DataNodeQueryMessages.UNSUPPORTED_DATA_TYPE + node.dataTypes[i]); + } + valuePayload[i] = buffer.getLong(); + } + } + } + node.numericValues = buildNumericValues(categories, tagPayload, valuePayload, nonNull); + node.setColumnCategories(categories); + return node; + } + + public static RelationalInsertNumericRowNode deserializeFromWAL(ByteBuffer buffer) { + long searchIndex = buffer.getLong(); + RelationalInsertNumericRowNode node = deserializeBody(buffer); + node.setSearchIndexFromWAL(searchIndex); + return node; + } + + private static RelationalInsertNumericRowNode deserializeBody(DataInputStream stream) + throws IOException { + RelationalInsertNumericRowNode node = new RelationalInsertNumericRowNode(new PlanNodeId("")); + node.time = stream.readLong(); + node.targetPath = new PartialPath(ReadWriteIOUtils.readString(stream), false); + int count = stream.readInt(); + boolean hasSchemas = stream.readByte() != 0; + node.measurements = new String[count]; + node.dataTypes = new TSDataType[count]; + node.measurementSchemas = hasSchemas ? new MeasurementSchema[count] : null; + TsTableColumnCategory[] categories = new TsTableColumnCategory[count]; + String[] tagPayload = new String[count]; + long[] valuePayload = new long[count]; + boolean[] nonNull = new boolean[count]; + for (int i = 0; i < count; i++) { + if (hasSchemas) { + node.measurementSchemas[i] = MeasurementSchema.deserializeFrom(stream); + node.measurements[i] = node.measurementSchemas[i].getMeasurementName(); + } else { + node.measurements[i] = ReadWriteIOUtils.readString(stream); + } + byte type = stream.readByte(); + node.dataTypes[i] = type == -1 ? null : TSDataType.deserialize(type); + categories[i] = TsTableColumnCategory.deserialize(stream); + nonNull[i] = stream.readByte() == 0; + if (nonNull[i]) { + if (categories[i] == TsTableColumnCategory.TAG && node.dataTypes[i] == TSDataType.STRING) { + tagPayload[i] = ReadWriteIOUtils.readString(stream); + } else { + if (!NumericRowValues.supports(node.dataTypes[i]) + || categories[i] != TsTableColumnCategory.FIELD) { + throw new UnSupportedDataTypeException( + DataNodeQueryMessages.UNSUPPORTED_DATA_TYPE + node.dataTypes[i]); + } + valuePayload[i] = stream.readLong(); + } + } + } + node.numericValues = buildNumericValues(categories, tagPayload, valuePayload, nonNull); + node.setColumnCategories(categories); + return node; + } + + private static NumericRowValues buildNumericValues( + TsTableColumnCategory[] categories, + String[] tagPayload, + long[] valuePayload, + boolean[] nonNull) { + int tagCount = 0; + while (tagCount < categories.length && categories[tagCount] == TsTableColumnCategory.TAG) { + tagCount++; + } + for (int i = tagCount; i < categories.length; i++) { + if (categories[i] == TsTableColumnCategory.TAG) { + throw new IllegalArgumentException(); + } + } + NumericRowValues values = new NumericRowValues(tagCount, categories.length - tagCount); + for (int i = 0; i < categories.length; i++) { + if (!nonNull[i]) { + values.setNull(i); + } else if (i < tagCount) { + values.setTag(i, tagPayload[i]); + } else { + values.setRawValue(i, valuePayload[i]); + } + } + return values; + } + + public static RelationalInsertNumericRowNode deserializeFromWAL(DataInputStream stream) + throws IOException { + long searchIndex = stream.readLong(); + RelationalInsertNumericRowNode node = deserializeBody(stream); + node.setSearchIndexFromWAL(searchIndex); + return node; + } + + public static RelationalInsertNumericRowNode deserialize(ByteBuffer buffer) { + RelationalInsertNumericRowNode node = deserializeBody(buffer); + node.setPlanNodeId(PlanNodeId.deserialize(buffer)); + return node; + } + + @Override + public boolean equals(Object other) { + return other instanceof RelationalInsertNumericRowNode node + && super.equals(other) + && time == node.time + && Objects.equals(numericValues, node.numericValues); + } + + @Override + public int hashCode() { + return Objects.hash(super.hashCode(), time, numericValues); + } + + @Override + public String toString() { + return "RelationalInsertNumericRowNode{table=" + + targetPath + + ", time=" + + time + + ", values=" + + Arrays.toString(getValues()) + + '}'; + } +} 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 db90503d464..fe77996fabe 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 @@ -36,9 +36,11 @@ import org.apache.iotdb.db.queryengine.plan.statement.StatementType; import org.apache.iotdb.db.queryengine.plan.statement.StatementVisitor; import org.apache.iotdb.db.utils.NumericRowValues; +import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.utils.RamUsageEstimator; +import org.apache.tsfile.write.UnSupportedDataTypeException; import java.nio.ByteBuffer; import java.util.Arrays; @@ -87,7 +89,6 @@ 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 { - numericValues = new NumericRowValues(measurements.length); dataTypes = new TSDataType[measurements.length]; for (int i = 0; i < dataTypes.length; i++) { byte type = buffer.get(); @@ -95,7 +96,41 @@ public class InsertNumericRowStatement extends InsertBaseStatement { throw new QueryProcessException(DataNodeQueryMessages.UNSUPPORTED_DATA_TYPE + type); } dataTypes[i] = TSDataType.values()[type]; - numericValues.read(i, dataTypes[i], buffer); + if (columnCategories[i] == TsTableColumnCategory.TAG && dataTypes[i] != TSDataType.STRING) { + throw new IllegalArgumentException( + DataNodeQueryMessages + .QUERY_EXCEPTION_MISSING_OR_INVALID_COLUMN_CATEGORIES_FOR_TABLE_INSERTION_5DF990B9); + } + if (columnCategories[i] == TsTableColumnCategory.FIELD + && !NumericRowValues.supports(dataTypes[i])) { + if (dataTypes[i] == TSDataType.STRING) { + throw new IllegalArgumentException( + DataNodeQueryMessages + .QUERY_EXCEPTION_MISSING_OR_INVALID_COLUMN_CATEGORIES_FOR_TABLE_INSERTION_5DF990B9); + } + throw new UnSupportedDataTypeException( + DataNodeQueryMessages.UNSUPPORTED_DATA_TYPE + dataTypes[i]); + } + } + int tagCount = countTagColumns(columnCategories); + numericValues = new NumericRowValues(tagCount, measurements.length - tagCount); + ColumnCategory[] categories = new ColumnCategory[measurements.length]; + for (int i = 0; i < measurements.length; i++) { + categories[i] = + columnCategories[i] == TsTableColumnCategory.TAG + ? ColumnCategory.TAG + : ColumnCategory.FIELD; + } + numericValues.setColumnCategories(categories); + for (int i = 0; i < dataTypes.length; i++) { + if (columnCategories[i] == TsTableColumnCategory.TAG) { + numericValues.read(i, dataTypes[i], buffer); + } + } + for (int i = 0; i < dataTypes.length; i++) { + if (columnCategories[i] == TsTableColumnCategory.FIELD) { + numericValues.read(i, dataTypes[i], buffer); + } } } @@ -208,37 +243,61 @@ public class InsertNumericRowStatement extends InsertBaseStatement { for (int i = 0; i < mapping.length; i++) { mapping[i] = i < pos ? i : i == pos ? -1 : i - 1; } - numericValues = numericValues.remap(mapping); + numericValues = numericValues.remap(mapping, countTagColumns(columnCategories)); deviceID = null; } @Override public void swapColumn(int src, int target) { super.swapColumn(src, target); - numericValues.swap(src, target); + int[] newToOld = new int[measurements.length]; + for (int i = 0; i < newToOld.length; i++) { + newToOld[i] = i; + } + newToOld[src] = target; + newToOld[target] = src; + numericValues = numericValues.remap(newToOld, countTagColumns(columnCategories)); deviceID = null; } @Override public void rebuildArraysAfterExpansion(int[] mapping, String[] newMeasurements) { super.rebuildArraysAfterExpansion(mapping, newMeasurements); - numericValues = numericValues.remap(mapping); + numericValues = numericValues.remap(mapping, countTagColumns(columnCategories)); deviceID = null; } @Override protected void subRemoveAttributeColumns(List<Integer> columnsToKeep) { numericValues = - numericValues.remap(columnsToKeep.stream().mapToInt(Integer::intValue).toArray()); + numericValues.remap( + columnsToKeep.stream().mapToInt(Integer::intValue).toArray(), + countTagColumns(columnCategories)); deviceID = null; } @Override public void setColumnCategories(TsTableColumnCategory[] categories) { super.setColumnCategories(categories); + if (numericValues != null) { + ColumnCategory[] compactCategories = new ColumnCategory[categories.length]; + for (int i = 0; i < categories.length; i++) { + compactCategories[i] = + categories[i] == TsTableColumnCategory.TAG ? ColumnCategory.TAG : ColumnCategory.FIELD; + } + numericValues.setColumnCategories(compactCategories); + } deviceID = null; } + private static int countTagColumns(TsTableColumnCategory[] categories) { + int count = 0; + while (count < categories.length && categories[count] == TsTableColumnCategory.TAG) { + count++; + } + return count; + } + public IDeviceID getTableDeviceID() { if (deviceID == null) { String[] segments = new String[getTagColumnIndices().size() + 1]; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index ba5349edc0e..1ac6c15488b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -1199,8 +1199,8 @@ public class StorageEngine implements IService { return () -> { if (!isReadyForNonReadWriteFunctions()) { return; - } - for (DataRegion dataRegion : dataRegionMap.values()) { + } + for (DataRegion dataRegion : dataRegionMap.values()) { if (dataRegion != null) { dataRegion.compactFileTimeIndexCache(); } 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 new file mode 100644 index 00000000000..8b9a03a44cc --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/NumericRowValues.java @@ -0,0 +1,307 @@ +/* + * 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.utils; + +import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.db.i18n.DataNodeQueryMessages; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.RamUsageEstimator; +import org.apache.tsfile.utils.ReadWriteIOUtils; +import org.apache.tsfile.write.UnSupportedDataTypeException; + +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.BitSet; +import java.util.Objects; + +/** + * Primitive storage for a numeric row. FLOAT and DOUBLE use raw IEEE 754 bits; BOOLEAN uses 0/1. + * TAGs live in a separate String[] and occupy the first logical columns; FIELDs follow them in + * {@code long[]}. Nulls need separate bits because every long bit pattern is a valid numeric value. + */ +public final class NumericRowValues { + private long[] values; + private final BitSet nulls; + private String[] tagValues; + private final int size; + private int tagCount; + + @TestOnly + public NumericRowValues(int size) { + this(0, size); + } + + public NumericRowValues(int tagLength, int valueLength) { + this.values = new long[valueLength]; + this.nulls = new BitSet(tagLength + valueLength); + this.tagValues = tagLength == 0 ? null : new String[tagLength]; + this.size = tagLength + valueLength; + this.tagCount = tagLength; + } + + @TestOnly + public NumericRowValues(long[] values, BitSet nulls) { + this.values = values; + this.nulls = nulls; + this.size = values.length; + this.tagCount = 0; + } + + public int size() { + return size; + } + + public long[] getValues() { + return values; + } + + public String getTag(int index) { + return index < 0 || index >= size || isNull(index) || tagValues == null || index >= tagCount + ? null + : tagValues[index]; + } + + public void setTag(int index, String value) { + if (index < 0 || index >= tagCount) { + throw new IndexOutOfBoundsException(index); + } + if (value == null) { + tagValues[index] = null; + nulls.set(index); + return; + } + tagValues[index] = value; + nulls.clear(index); + } + + public boolean isNull(int index) { + return nulls.get(index); + } + + public void setNull(int index) { + if (index < 0 || index >= size) { + throw new IndexOutOfBoundsException(index); + } + // Null slots have no payload on the wire; keep their in-memory representation canonical too. + if (index >= tagCount) { + values[index - tagCount] = 0; + } + if (tagValues != null && index < tagCount) { + tagValues[index] = null; + } + nulls.set(index); + } + + public long get(int index) { + if (index < 0 || index >= size) { + throw new IndexOutOfBoundsException(index); + } + return index < tagCount ? 0 : values[index - tagCount]; + } + + public void setRawValue(int index, long value) { + if (index < tagCount) { + throw new IllegalArgumentException(); + } + if (index < 0 || index >= size) { + throw new IndexOutOfBoundsException(index); + } + values[index - tagCount] = value; + nulls.clear(index); + } + + public String[] getTagValues() { + return tagValues; + } + + public void setColumnCategories(ColumnCategory[] categories) { + if (categories == null || categories.length != size) { + throw new IllegalArgumentException(); + } + int newTagCount = 0; + while (newTagCount < size && categories[newTagCount] == ColumnCategory.TAG) { + newTagCount++; + } + for (int i = newTagCount; i < size; i++) { + if (categories[i] == ColumnCategory.TAG) { + throw new IllegalArgumentException(); + } + } + if (newTagCount != tagCount) { + throw new IllegalArgumentException(); + } + } + + public static boolean supports(TSDataType type) { + return type != null + && switch (type) { + case BOOLEAN, INT32, INT64, FLOAT, DOUBLE, DATE, TIMESTAMP -> true; + case TEXT, STRING, BLOB, OBJECT, VECTOR, UNKNOWN -> false; + }; + } + + public Object getObject(int index, TSDataType type) { + if (isNull(index)) { + return null; + } + return switch (type) { + case BOOLEAN -> get(index) != 0; + case INT32, DATE -> (int) get(index); + case INT64, TIMESTAMP -> get(index); + case FLOAT -> Float.intBitsToFloat((int) get(index)); + case DOUBLE -> Double.longBitsToDouble(get(index)); + // Binary is required only at existing per-cell APIs (schema, Pipe and subscription). + case STRING -> { + if (index >= tagCount) { + throw new IllegalArgumentException(); + } + yield new Binary(getTag(index), TSFileConfig.STRING_CHARSET); + } + default -> throw unsupported(type); + }; + } + + // Used only at APIs that consume individual objects (last cache, diagnostics and type coercion). + public void setObject(int index, TSDataType type, Object value) { + if (value == null) { + setNull(index); + return; + } + if (type == TSDataType.STRING) { + setTag( + index, + value instanceof Binary binary + ? binary.getStringValue(TSFileConfig.STRING_CHARSET) + : (String) value); + return; + } + if (index < tagCount || index < 0 || index >= size) { + throw new IllegalArgumentException(); + } + values[index - tagCount] = + switch (type) { + case BOOLEAN -> (boolean) value ? 1 : 0; + case INT32, DATE -> (int) value; + case INT64, TIMESTAMP -> (long) value; + case FLOAT -> Float.floatToRawIntBits((float) value); + case DOUBLE -> Double.doubleToRawLongBits((double) value); + default -> throw unsupported(type); + }; + nulls.clear(index); + } + + public boolean cast(int index, TSDataType source, TSDataType target) { + if (isNull(index)) { + return true; + } + if (source == TSDataType.STRING && target == TSDataType.STRING) { + return true; + } + if (!supports(target) || source == null || !target.isCompatible(source)) { + return false; + } + if (source == target) { + return true; + } + setObject(index, target, target.castFromSingleValue(source, getObject(index, source))); + return true; + } + + /** Returns a new primitive row; -1 in the mapping represents an inserted null column. */ + public NumericRowValues remap(int[] newToOld, int newTagCount) { + NumericRowValues result = new NumericRowValues(newTagCount, newToOld.length - newTagCount); + for (int i = 0; i < newToOld.length; i++) { + int old = newToOld[i]; + if (old < 0 || old >= size() || isNull(old)) { + result.setNull(i); + } else if (i < newTagCount) { + if (tagValues != null && old < this.tagCount && tagValues[old] != null) { + result.setTag(i, tagValues[old]); + } + } else if (old >= this.tagCount) { + result.setRawValue(i, get(old)); + } else { + result.setNull(i); + } + } + return result; + } + + public void read(int index, TSDataType type, ByteBuffer buffer) { + if (type == TSDataType.STRING) { + setTag(index, ReadWriteIOUtils.readString(buffer)); + return; + } + if (index < tagCount || index < 0 || index >= size) { + throw new IllegalArgumentException(); + } + values[index - tagCount] = + switch (type) { + case BOOLEAN -> buffer.get() == 0 ? 0 : 1; + case INT32, DATE, FLOAT -> buffer.getInt(); + case INT64, TIMESTAMP, DOUBLE -> buffer.getLong(); + default -> throw unsupported(type); + }; + } + + public long ramBytesUsed() { + return RamUsageEstimator.shallowSizeOfInstance(NumericRowValues.class) + + RamUsageEstimator.sizeOf(values) + + (tagValues == null ? 0 : RamUsageEstimator.sizeOf(tagValues)) + + RamUsageEstimator.shallowSizeOfInstance(BitSet.class) + + RamUsageEstimator.NUM_BYTES_ARRAY_HEADER + + (long) nulls.size() / Byte.SIZE; + } + + private static UnSupportedDataTypeException unsupported(TSDataType type) { + return new UnSupportedDataTypeException(DataNodeQueryMessages.UNSUPPORTED_DATA_TYPE + type); + } + + @Override + public boolean equals(Object other) { + if (!(other instanceof NumericRowValues row) + || size != row.size + || tagCount != row.tagCount + || !Arrays.equals(values, row.values) + || !nulls.equals(row.nulls)) { + return false; + } + // An allocated array cleared by a partial failure equals an absent all-null TAG array. + for (int i = 0; i < size(); i++) { + if (!Objects.equals(getTag(i), row.getTag(i))) { + return false; + } + } + return true; + } + + @Override + public int hashCode() { + int hash = 31 * (31 * size + tagCount) + Arrays.hashCode(values) + nulls.hashCode(); + for (int i = 0; i < size(); i++) { + hash = 31 * hash + Objects.hashCode(getTag(i)); + } + return hash; + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/NumericTableRowTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/NumericTableRowTest.java new file mode 100644 index 00000000000..0541ecd2173 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/NumericTableRowTest.java @@ -0,0 +1,524 @@ +/* + * 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.queryengine.plan.parser; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.db.pipe.event.common.tablet.parser.TabletInsertionEventTablePatternParser; +import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletBatchReqV2; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertNumericRowNode; +import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis; +import org.apache.iotdb.db.queryengine.plan.relational.planner.TableModelPlanner; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertNumericRow; +import org.apache.iotdb.db.queryengine.plan.statement.StatementType; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; +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.storageengine.dataregion.memtable.AlignedWritableMemChunkGroup; +import org.apache.iotdb.db.storageengine.dataregion.memtable.PrimitiveMemTable; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; +import org.apache.iotdb.db.subscription.broker.consensus.ConsensusLogToTabletConverter; +import org.apache.iotdb.db.utils.NumericRowValues; +import org.apache.iotdb.db.utils.datastructure.AlignedTVList; +import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq; +import org.apache.iotdb.session.util.SessionUtils; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.common.type.TypeFactory; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.ReadWriteIOUtils; +import org.apache.tsfile.write.UnSupportedDataTypeException; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.Test; +import org.mockito.Mockito; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.lang.reflect.Field; +import java.nio.ByteBuffer; +import java.time.LocalDate; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +/** Exercises one numeric row from the client codec through schema edits, WAL and memtable. */ +public class NumericTableRowTest { + private static final List<String> NAMES = Arrays.asList("b", "i", "l", "f", "d", "date", "ts"); + private static final List<TSDataType> TYPES = + Arrays.asList( + TSDataType.BOOLEAN, + TSDataType.INT32, + TSDataType.INT64, + TSDataType.FLOAT, + TSDataType.DOUBLE, + TSDataType.DATE, + TSDataType.TIMESTAMP); + private static final long[] BITS = { + 1, Integer.MIN_VALUE, Long.MIN_VALUE, 0x7fc01234, 0x7ff8000000001234L, 20260910, Long.MAX_VALUE + }; + + private TSInsertRecordReq request() throws Exception { + return new TSInsertRecordReq() + .setPrefixPath("numeric_table") + .setTimestamp(42) + .setMeasurements(NAMES) + .setValues( + SessionUtils.getValueBuffer( + TYPES, + BITS, + NAMES, + Collections.nCopies(NAMES.size(), ColumnCategory.FIELD), + new String[0])) + .setIsAligned(true) + .setIsWriteToTable(true) + .setIsNumericRow(true) + .setColumnCategoryies( + Collections.nCopies(NAMES.size(), (byte) ColumnCategory.FIELD.ordinal())); + } + + @Test + public void testRpcDecodeAndSchemaChangesRetainPrimitiveStorage() throws Exception { + // Missing TAG insertion, column swaps and partial failure recovery must retain long[] values. + InsertNumericRowStatement statement = + (InsertNumericRowStatement) StatementGenerator.createStatement(request()); + assertArrayEquals(BITS, statement.getNumericValues().getValues()); + assertObjectStorageAbsent(InsertNumericRowStatement.class, statement); + assertTrue(statement.isWriteToTable()); + assertTrue(statement.isAligned()); + assertEquals(StatementType.INSERT_NUMERIC_ROW, statement.getType()); + statement.semanticCheck(); + + statement.insertColumn( + 0, + new ColumnSchema( + "tag", TypeFactory.getType(TSDataType.STRING), false, TsTableColumnCategory.TAG)); + assertTrue(statement.getNumericValues().isNull(0)); + assertNull(statement.getFirstValueOfIndex(0)); + assertEquals(20260910, statement.getFirstValueOfIndex(6)); + statement.swapColumn(1, 2); + assertEquals(Integer.MIN_VALUE, statement.getNumericValues().get(1)); + assertEquals(1, statement.getNumericValues().get(2)); + statement.markFailedMeasurement(1, new IllegalArgumentException()); + assertTrue(statement.getNumericValues().isNull(1)); + statement.removeAllFailedMeasurementMarks(); + assertEquals(Integer.MIN_VALUE, statement.getNumericValues().get(1)); + assertObjectStorageAbsent(InsertNumericRowStatement.class, statement); + assertEquals("numeric_table", statement.getTableDeviceID().getTableName()); + } + + @Test + public void testRpcRejectsNumericTreeAndNonStringTagRequests() throws Exception { + // Raw RPC callers must use STRING TAGs and numeric FIELDs in the table model. + TSInsertRecordReq tree = request().setIsWriteToTable(false); + assertThrows(IllegalArgumentException.class, () -> StatementGenerator.createStatement(tree)); + TSInsertRecordReq tag = + request() + .setColumnCategoryies( + Collections.nCopies(NAMES.size(), (byte) ColumnCategory.TAG.ordinal())); + assertThrows(IllegalArgumentException.class, () -> StatementGenerator.createStatement(tag)); + } + + @Test + public void testStringTagsSurviveSchemaChanges() throws Exception { + // The client places TAG metadata before FIELD metadata while retaining dense value arrays. + List<String> names = Arrays.asList("tag2", "tag1", "b", "l"); + List<TSDataType> types = + Arrays.asList(TSDataType.STRING, TSDataType.STRING, TSDataType.BOOLEAN, TSDataType.INT64); + List<ColumnCategory> categories = + Arrays.asList( + ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.FIELD); + TSInsertRecordReq req = + request() + .setMeasurements(names) + .setValues( + SessionUtils.getValueBuffer( + types, + new long[] {1, Long.MIN_VALUE}, + names, + categories, + new String[] {"设备😀", ""})) + .setColumnCategoryies( + Arrays.asList( + (byte) ColumnCategory.TAG.ordinal(), + (byte) ColumnCategory.TAG.ordinal(), + (byte) ColumnCategory.FIELD.ordinal(), + (byte) ColumnCategory.FIELD.ordinal())); + InsertNumericRowStatement statement = + (InsertNumericRowStatement) StatementGenerator.createStatement(req); + assertEquals("设备😀", statement.getNumericValues().getTag(0)); + assertEquals("", statement.getNumericValues().getTag(1)); + assertArrayEquals(new long[] {1, Long.MIN_VALUE}, statement.getValues()); + statement.rebuildArraysAfterExpansion( + new int[] {-1, 1, 0, 2, 3}, new String[] {"missing", "tag1", "tag2", "b", "l"}); + assertArrayEquals( + new String[] {"numeric_table", null, "", "设备😀"}, + statement.getTableDeviceID().getSegments()); + statement.swapColumn(1, 2); + assertEquals("设备😀", statement.getNumericValues().getTag(1)); + statement.markFailedMeasurement(1, new IllegalArgumentException()); + statement.removeAllFailedMeasurementMarks(); + assertEquals("设备😀", statement.getNumericValues().getTag(1)); + assertEquals(1, statement.getValues()[0]); + assertEquals(Long.MIN_VALUE, statement.getValues()[1]); + assertObjectStorageAbsent(InsertNumericRowStatement.class, statement); + } + + @Test + public void testStringTagsSurviveWalConsensusAndTabletAdapters() throws Exception { + // TAGs identify the device through WAL recovery and remain present in Pipe/subscription + // Tablets. + for (String tag : Arrays.asList("设备😀", "", null)) { + RelationalInsertNumericRowNode node = nodeWithNullTag(); + node.getNumericValues().setTag(0, tag); + assertRoundTrips(node); + ByteBuffer bytes = node.serializeToByteBuffer(); + RelationalInsertNumericRowNode decoded = + (RelationalInsertNumericRowNode) PlanNodeType.deserialize(bytes); + assertEquals(tag, decoded.getNumericValues().getTag(0)); + assertEquals(node.getDeviceID(), decoded.getDeviceID()); + assertEquals( + new Binary(tag == null ? "" : tag, TSFileConfig.STRING_CHARSET), + tag == null ? Binary.EMPTY_VALUE : decoded.getValue(0)); + Tablet pipe = + new TabletInsertionEventTablePatternParser(null, null, decoded, null).convertToTablet(); + Tablet subscription = + new ConsensusLogToTabletConverter(null, null, null, "db").convert(decoded).get(0); + for (Tablet tablet : Arrays.asList(pipe, subscription)) { + if (tag == null) { + assertTrue(tablet.getBitMaps()[0].isMarked(0)); + } else { + assertEquals( + tag, + ((Binary[]) tablet.getValues()[0])[0].getStringValue(TSFileConfig.STRING_CHARSET)); + } + } + } + } + + @Test + public void testBulkTagExpansionKeepsValuesAndNulls() throws Exception { + // A reversed schema mapping with two missing TAGs must preserve numeric bits and null slots. + InsertNumericRowStatement statement = + (InsertNumericRowStatement) StatementGenerator.createStatement(request()); + statement.rebuildArraysAfterExpansion( + new int[] {-1, -1, 6, 5, 4, 3, 2, 1, 0}, + new String[] {"tag1", "tag2", "ts", "date", "d", "f", "l", "i", "b"}); + assertTrue(statement.getNumericValues().isNull(0)); + assertTrue(statement.getNumericValues().isNull(1)); + for (int i = 0; i < BITS.length; i++) { + assertEquals(BITS[BITS.length - 1 - i], statement.getNumericValues().get(i + 2)); + } + assertObjectStorageAbsent(InsertNumericRowStatement.class, statement); + } + + @Test + public void testNumericCastsKeepPrimitiveStorage() { + // Widen INT32 to DOUBLE in place; reject STRING coercion while accepting an inserted null TAG. + NumericRowValues values = new NumericRowValues(2); + values.getValues()[0] = Integer.MIN_VALUE; + values.setNull(1); + assertTrue(values.cast(0, TSDataType.INT32, TSDataType.DOUBLE)); + assertEquals(Double.doubleToRawLongBits((double) Integer.MIN_VALUE), values.get(0)); + assertFalse(values.cast(0, TSDataType.DOUBLE, TSDataType.STRING)); + assertTrue(values.cast(1, null, TSDataType.STRING)); + assertTrue(values.isNull(1)); + } + + @Test + public void testConsensusAndWalRetainPrimitiveStorage() throws Exception { + // Both serializers and WAL readers must retain numeric storage, including partial failures. + RelationalInsertNumericRowNode node = nodeWithNullTag(); + assertRoundTrips(node); + node.markFailedMeasurement(2); + assertRoundTrips(node); + node.setMeasurementSchemas(null); + assertRoundTrips(node); + } + + @Test + public void testWalEnvelopeRetainsTypeAndFrozenIndex() throws Exception { + RelationalInsertNumericRowNode node = nodeWithNullTag(); + node.setSearchIndex(123); + node.setLastFragment(true); + WALInfoEntry entry = new WALInfoEntry(7, node); + // WAL queuing freezes the consensus index even if the caller later reuses the node. + node.setSearchIndex(456); + node.setLastFragment(false); + ByteBuffer buffer = ByteBuffer.allocate(entry.serializedSize()); + entry.serialize(new WALByteBufferForTest(buffer)); + assertFalse(buffer.hasRemaining()); + buffer.flip(); + RelationalInsertNumericRowNode consensus = + (RelationalInsertNumericRowNode) WALEntry.deserializeForConsensus(buffer); + assertEquals(123, consensus.getSearchIndex()); + assertTrue(consensus.isLastFragment()); + assertEquals(node.getNumericValues(), consensus.getNumericValues()); + try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(buffer.array()))) { + WALEntry recovered = WALEntry.deserialize(input); + assertEquals(7, recovered.getMemTableId()); + RelationalInsertNumericRowNode row = (RelationalInsertNumericRowNode) recovered.getValue(); + assertEquals(123, row.getSearchIndex()); + assertTrue(row.isLastFragment()); + assertEquals(node.getNumericValues(), row.getNumericValues()); + assertEquals(0, input.available()); + assertObjectStorageAbsent(RelationalInsertNumericRowNode.class, row); + } + } + + @Test + public void testPipeBatchRetainsIndependentNumericStatements() throws Exception { + RelationalInsertNumericRowNode node = nodeWithNullTag(); + PipeTransferTabletBatchReqV2 request = + PipeTransferTabletBatchReqV2.toTPipeTransferReq( + Arrays.asList(node.serializeToByteBuffer(), node.serializeToByteBuffer()), + Collections.emptyList(), + Arrays.asList("db", "db"), + Collections.emptyList()); + List<InsertBaseStatement> statements = + PipeTransferTabletBatchReqV2.fromTPipeTransferReq(request).constructStatements(); + // Batching must not replace numeric statements with the object-valued InsertRowsStatement. + assertEquals(2, statements.size()); + for (InsertBaseStatement statement : statements) { + assertEquals(StatementType.INSERT_NUMERIC_ROW, statement.getType()); + assertEquals("db", statement.getDatabaseName().get()); + assertEquals( + node.getNumericValues(), ((InsertNumericRowStatement) statement).getNumericValues()); + assertObjectStorageAbsent(InsertNumericRowStatement.class, statement); + } + } + + @Test + public void testNumericRowRedirectUsesSingleEndpoint() throws Exception { + InsertNumericRowStatement statement = + (InsertNumericRowStatement) StatementGenerator.createStatement(request()); + Analysis analysis = new Analysis(new InsertNumericRow(statement, null), Collections.emptyMap()); + TEndPoint local = new TEndPoint("127.0.0.1", 6667); + TEndPoint leader = new TEndPoint("127.0.0.1", 6668); + analysis.setRedirectNodeList(Collections.singletonList(leader)); + TableModelPlanner planner = Mockito.mock(TableModelPlanner.class, Mockito.CALLS_REAL_METHODS); + TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + planner.setRedirectInfo(analysis, local, status); + assertEquals(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode(), status.getCode()); + assertEquals(leader, status.getRedirectNode()); + assertFalse(status.isSetSubStatus()); + TSStatus localStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + planner.setRedirectInfo(analysis, leader, localStatus); + assertFalse(localStatus.isSetRedirectNode()); + } + + private void assertRoundTrips(RelationalInsertNumericRowNode node) throws Exception { + ByteBuffer streamed = node.serializeToByteBuffer(); + ByteBuffer direct = ByteBuffer.allocate(streamed.remaining()); + node.serialize(direct); + direct.flip(); + assertEquals(streamed, direct); + RelationalInsertNumericRowNode decoded = + (RelationalInsertNumericRowNode) PlanNodeType.deserialize(direct); + assertObjectStorageAbsent(RelationalInsertNumericRowNode.class, decoded); + assertEquals(node.getNumericValues().isNull(0), decoded.getNumericValues().isNull(0)); + assertEquals(node.getNumericValues().getTag(0), decoded.getNumericValues().getTag(0)); + assertEquals(node.getDeviceID(), decoded.getDeviceID()); + + ByteBuffer wal = ByteBuffer.allocate(node.serializedSize()); + node.setSearchIndex(123); + node.setLastFragment(true); + node.serializeToWAL(new WALByteBufferForTest(wal)); + assertFalse(wal.hasRemaining()); + wal.flip(); + assertEquals(PlanNodeType.RELATIONAL_INSERT_NUMERIC_ROW.getNodeType(), wal.getShort()); + RelationalInsertNumericRowNode recovered = + RelationalInsertNumericRowNode.deserializeFromWAL(wal); + assertFalse(wal.hasRemaining()); + assertEquals(decoded.getNumericValues(), recovered.getNumericValues()); + assertArrayEquals(decoded.getMeasurements(), recovered.getMeasurements()); + assertArrayEquals(decoded.getColumnCategories(), recovered.getColumnCategories()); + assertObjectStorageAbsent(RelationalInsertNumericRowNode.class, recovered); + assertEquals(123, recovered.getSearchIndex()); + assertTrue(recovered.isLastFragment()); + + try (DataInputStream input = new DataInputStream(new ByteArrayInputStream(wal.array()))) { + input.readShort(); + RelationalInsertNumericRowNode fromStream = + RelationalInsertNumericRowNode.deserializeFromWAL(input); + assertEquals(recovered.getNumericValues(), fromStream.getNumericValues()); + assertObjectStorageAbsent(RelationalInsertNumericRowNode.class, fromStream); + assertEquals(0, input.available()); + } + } + + @Test + public void testMemtableAcceptsRecoveredRowWithoutBoxing() throws Exception { + // A WAL-recovered row must reach primitive column arrays without boxing the row or its NaNs. + RelationalInsertNumericRowNode original = nodeWithNullTag(); + ByteBuffer wal = ByteBuffer.allocate(original.serializedSize()); + original.serializeToWAL(new WALByteBufferForTest(wal)); + wal.flip(); + wal.getShort(); + RelationalInsertNumericRowNode recovered = + RelationalInsertNumericRowNode.deserializeFromWAL(wal); + // Fail immediately if the primitive memtable write path boxes even a single cell. + RelationalInsertNumericRowNode row = + new RelationalInsertNumericRowNode(new PlanNodeId("test")) { + @Override + public Object getValue(int index) { + throw new AssertionError("Numeric memtable writes must not box the row"); + } + }; + row.setTargetPath(recovered.getTargetPath()); + row.setTime(42); + row.setAligned(true); + row.setMeasurements(recovered.getMeasurements()); + row.setDataTypes(recovered.getDataTypes()); + row.setMeasurementSchemas(recovered.getMeasurementSchemas()); + row.setColumnCategories(recovered.getColumnCategories()); + row.setNumericValues(recovered.getNumericValues()); + PrimitiveMemTable memtable = new PrimitiveMemTable(); + try { + assertEquals(7, memtable.insertAlignedRow(row)); + AlignedTVList list = + ((AlignedWritableMemChunkGroup) memtable.getMemTableMap().get(row.getDeviceID())) + .getAlignedMemChunk() + .getWorkingTVList(); + assertEquals(42, list.getTime(0)); + assertTrue(list.getBooleanByValueIndex(0, 0)); + assertEquals(Integer.MIN_VALUE, list.getIntByValueIndex(0, 1)); + assertEquals(Long.MIN_VALUE, list.getLongByValueIndex(0, 2)); + assertEquals(0x7fc01234, Float.floatToRawIntBits(list.getFloatByValueIndex(0, 3))); + assertEquals( + 0x7ff8000000001234L, Double.doubleToRawLongBits(list.getDoubleByValueIndex(0, 4))); + assertEquals(20260910, list.getIntByValueIndex(0, 5)); + assertEquals(Long.MAX_VALUE, list.getLongByValueIndex(0, 6)); + assertObjectStorageAbsent(RelationalInsertNumericRowNode.class, row); + } finally { + memtable.release(); + } + } + + @Test + public void testRpcRejectsNonNumericTypes() throws Exception { + // Raw RPC callers cannot smuggle unsupported values into numeric FIELD storage. + for (TSDataType type : + Arrays.asList( + TSDataType.TEXT, + TSDataType.BLOB, + TSDataType.OBJECT, + TSDataType.VECTOR, + TSDataType.UNKNOWN)) { + TSInsertRecordReq invalid = + request().setValues(ByteBuffer.wrap(new byte[] {type.serialize()})); + assertThrows( + UnSupportedDataTypeException.class, () -> StatementGenerator.createStatement(invalid)); + } + // STRING has a valid wire representation for TAGs, but must still be rejected as a FIELD. + ByteBuffer stringValue = ByteBuffer.allocate(32); + ReadWriteIOUtils.write(TSDataType.STRING, stringValue); + ReadWriteIOUtils.write("invalid", stringValue); + stringValue.flip(); + TSInsertRecordReq stringField = + request() + .setMeasurements(Collections.singletonList("s")) + .setColumnCategoryies(Collections.singletonList((byte) ColumnCategory.FIELD.ordinal())) + .setValues(stringValue); + assertThrows( + IllegalArgumentException.class, () -> StatementGenerator.createStatement(stringField)); + } + + @Test + public void testTabletAdaptersDoNotMaterializeObjectRows() throws Exception { + // Pipe and subscription must preserve values and null TAGs without boxing the entire row. + RelationalInsertNumericRowNode node = Mockito.spy(nodeWithNullTag()); + Mockito.doThrow(new AssertionError("Numeric adapters must not box the row")) + .when(node) + .getValues(); + Tablet pipe = + new TabletInsertionEventTablePatternParser(null, null, node, null).convertToTablet(); + List<Tablet> subscription = + new ConsensusLogToTabletConverter(null, null, null, "db").convert(node); + assertEquals(1, subscription.size()); + for (Tablet tablet : Arrays.asList(pipe, subscription.get(0))) { + assertEquals(1, tablet.getRowSize()); + assertEquals("numeric_table", tablet.getTableName()); + assertEquals(42, tablet.getTimestamps()[0]); + assertTrue(tablet.getBitMaps()[0].isMarked(0)); + assertTrue(((boolean[]) tablet.getValues()[1])[0]); + assertEquals(Integer.MIN_VALUE, ((int[]) tablet.getValues()[2])[0]); + assertEquals(Long.MIN_VALUE, ((long[]) tablet.getValues()[3])[0]); + assertEquals(0x7fc01234, Float.floatToRawIntBits(((float[]) tablet.getValues()[4])[0])); + assertEquals( + 0x7ff8000000001234L, Double.doubleToRawLongBits(((double[]) tablet.getValues()[5])[0])); + assertEquals(LocalDate.of(2026, 9, 10), ((LocalDate[]) tablet.getValues()[6])[0]); + assertEquals(Long.MAX_VALUE, ((long[]) tablet.getValues()[7])[0]); + } + assertObjectStorageAbsent(RelationalInsertNumericRowNode.class, node); + } + + private RelationalInsertNumericRowNode nodeWithNullTag() throws Exception { + InsertNumericRowStatement statement = + (InsertNumericRowStatement) StatementGenerator.createStatement(request()); + statement.insertColumn( + 0, + new ColumnSchema( + "tag", TypeFactory.getType(TSDataType.STRING), false, TsTableColumnCategory.TAG)); + MeasurementSchema[] schemas = new MeasurementSchema[statement.getValueCount()]; + for (int i = 0; i < schemas.length; i++) { + schemas[i] = new MeasurementSchema(statement.getMeasurements()[i], statement.getDataType(i)); + } + RelationalInsertNumericRowNode node = + new RelationalInsertNumericRowNode( + new PlanNodeId("numeric"), + statement.getDevicePath(), + statement.getMeasurements(), + statement.getDataTypes(), + schemas, + statement.getTime(), + statement.getNumericValues(), + statement.getColumnCategories()); + node.setNumericValues(statement.getNumericValues()); + return node; + } + + private void assertObjectStorageAbsent(Class<?> owner, Object row) throws Exception { + // A dedicated numeric type must not inherit the object row or retain an Object[] payload. + assertFalse(InsertRowStatement.class.isInstance(row)); + assertFalse(InsertRowNode.class.isInstance(row)); + for (Class<?> type = owner; type != null; type = type.getSuperclass()) { + for (Field field : type.getDeclaredFields()) { + assertFalse(field.getName(), field.getType() == Object[].class); + } + } + } +}
