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 eed7ef0dba99bf2ec390be12b1ff53ca193db2fb Author: Tian Jiang <[email protected]> AuthorDate: Fri Sep 11 12:30:16 2026 +0800 tmp --- .../IoTDBNumericRowInsertPerformanceIT.java | 456 +++++++++++++++++++++ .../it/session/IoTDBTableModelSessionIT.java | 75 ++-- .../org/apache/iotdb/isession/ITableSession.java | 16 +- .../java/org/apache/iotdb/isession/NumericRow.java | 97 +++++ .../apache/iotdb/session/i18n/SessionMessages.java | 4 + .../apache/iotdb/session/i18n/SessionMessages.java | 4 + .../java/org/apache/iotdb/session/Session.java | 51 +++ .../org/apache/iotdb/session/TableSession.java | 6 + .../iotdb/session/pool/TableSessionWrapper.java | 13 + .../apache/iotdb/session/util/SessionUtils.java | 74 ++++ .../java/org/apache/iotdb/session/SessionTest.java | 80 ++++ .../iotdb/session/util/SessionUtilsTest.java | 151 +++++++ .../tablet/PipeInsertNodeTabletInsertionEvent.java | 12 +- .../tablet/parser/TabletInsertionEventParser.java | 19 +- .../TabletInsertionEventTablePatternParser.java | 5 +- .../visitor/PipePlanToStatementVisitor.java | 15 + .../request/PipeTransferTabletBatchReqV2.java | 10 +- .../request/PipeTransferTabletBinaryReq.java | 4 +- .../request/PipeTransferTabletBinaryReqV2.java | 4 +- .../request/PipeTransferTabletInsertNodeReq.java | 4 +- .../request/PipeTransferTabletInsertNodeReqV2.java | 4 +- .../execution/executor/RegionWriteExecutor.java | 7 + .../plan/parser/StatementGenerator.java | 46 ++- .../plan/planner/plan/node/PlanVisitor.java | 5 + .../plan/node/write/AbstractInsertRowNode.java | 64 +++ .../relational/analyzer/StatementAnalyzer.java | 6 + .../plan/relational/planner/RelationPlanner.java | 23 ++ .../plan/relational/planner/TableModelPlanner.java | 14 + .../plan/relational/sql/ast/InsertNumericRow.java | 71 ++++ .../queryengine/plan/statement/StatementType.java | 1 + .../plan/statement/StatementVisitor.java | 5 + .../statement/crud/InsertNumericRowStatement.java | 292 +++++++++++++ .../iotdb/db/storageengine/StorageEngine.java | 4 +- .../dataregion/memtable/TsFileProcessor.java | 55 +-- .../dataregion/wal/buffer/WALEntry.java | 6 +- .../dataregion/wal/buffer/WALInfoEntry.java | 4 +- .../dataregion/wal/node/IWALNode.java | 4 +- .../dataregion/wal/node/WALFakeNode.java | 4 +- .../storageengine/dataregion/wal/node/WALNode.java | 4 +- .../wal/recover/file/TsFilePlanRedoer.java | 9 +- .../plan/parser/StatementGeneratorTest.java | 2 +- .../AlignedWritableMemChunkNumericTest.java | 134 ++++++ .../plan/planner/plan/node/PlanNodeType.java | 1 + 43 files changed, 1767 insertions(+), 98 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 new file mode 100644 index 00000000000..84a52da1451 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/performance/IoTDBNumericRowInsertPerformanceIT.java @@ -0,0 +1,456 @@ +/* + * 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.relational.it.performance; + +import org.apache.iotdb.isession.ITableSession; +import org.apache.iotdb.isession.NumericRow; +import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.write.record.Tablet; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Assume; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Locale; +import java.util.Properties; + +/** + * 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. + * + * <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 + * reproducible commands and interpretation. + */ +@RunWith(IoTDBTestRunner.class) +@Category(TableLocalStandaloneIT.class) +public class IoTDBNumericRowInsertPerformanceIT { + + private static final String PROPERTY_PREFIX = "iotdb.numeric.row.insert.perf."; + 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 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 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); + + private static final String DATABASE = "numeric_row_insert_performance"; + private static final List<TSDataType> TYPE_SEQUENCE = + Arrays.asList( + TSDataType.BOOLEAN, + TSDataType.INT32, + TSDataType.INT64, + TSDataType.FLOAT, + TSDataType.DOUBLE); + private static List<String> columnNames; + private static List<TSDataType> dataTypes; + private static List<ColumnCategory> columnCategories; + + private static boolean environmentInitialized; + + @BeforeClass + public static void setUp() throws Exception { + Assume.assumeTrue(Boolean.getBoolean(ENABLED_PROPERTY)); + // Validate before allocating inputs or starting nodes, including when a count is negative. + assertValidConfiguration(); + columnNames = createColumnNames(); + dataTypes = createDataTypes(); + columnCategories = Collections.nCopies(COLUMN_COUNT, ColumnCategory.FIELD); + environmentInitialized = true; + EnvFactory.getEnv().initClusterEnvironment(); + } + + @AfterClass + public static void tearDown() throws Exception { + if (environmentInitialized) { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + } + + @Test + // Compare identical, prebuilt numeric data; verify every field count outside the timed calls. + public void testNumericRowInsertPerformanceAgainstTablet() throws Exception { + 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 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 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); + } else { + tabletNanos = executeTabletWorkload(session, tabletWorkload); + numericRowNanos = executeNumericRowWorkload(session, numericRowWorkload); + } + assertRowCount(session, numericRowWorkload.tableName); + assertRowCount(session, tabletWorkload.tableName); + + if (round >= WARMUP_ROUND_COUNT) { + final int measuredRound = round - WARMUP_ROUND_COUNT; + numericRowElapsedNanos[measuredRound] = numericRowNanos; + tabletElapsedNanos[measuredRound] = tabletNanos; + } + } + } + + 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); + } + return new NumericRowWorkload(tableName, rows); + } + + private static TabletWorkload prepareTabletWorkload(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])); + } + } + tablets.add(tablet); + rowStart += rowCount; + } + return new TabletWorkload(tableName, tablets); + } + + 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; + } + + private static long executeTabletWorkload(ITableSession session, TabletWorkload tabletWorkload) + throws Exception { + final long startNanos = System.nanoTime(); + for (final Tablet tablet : tabletWorkload.tablets) { + session.insert(tablet); + } + return System.nanoTime() - startNanos; + } + + private static void createTable(ITableSession session, String tableName) throws Exception { + final StringBuilder statement = new StringBuilder("CREATE TABLE "); + statement.append(tableName).append('('); + for (int column = 0; column < COLUMN_COUNT; column++) { + if (column > 0) { + statement.append(','); + } + statement + .append(columnNames.get(column)) + .append(' ') + .append(dataTypes.get(column)) + .append(" FIELD"); + } + statement.append(')'); + session.executeNonQueryStatement(statement.toString()); + } + + private static void assertRowCount(ITableSession session, String tableName) throws Exception { + final List<String> counts = new ArrayList<>(COLUMN_COUNT); + for (String columnName : columnNames) { + counts.add("COUNT(" + columnName + ")"); + } + try (SessionDataSet dataSet = + session.executeQueryStatement( + "SELECT " + String.join(",", counts) + " FROM " + tableName)) { + Assert.assertTrue(dataSet.hasNext()); + for (var field : dataSet.next().getFields()) { + Assert.assertEquals(ROW_COUNT, field.getLongV()); + } + Assert.assertFalse(dataSet.hasNext()); + } + } + + private static long[] createRawValues(long timestamp) { + final long[] values = new long[COLUMN_COUNT]; + for (int column = 0; column < COLUMN_COUNT; column++) { + final TSDataType type = dataTypes.get(column); + // Floating-point values occupy raw IEEE 754 bits in the numeric-row API. + values[column] = + switch (type) { + case BOOLEAN -> (timestamp + column) & 1; + case INT32 -> (int) (timestamp + column); + case INT64 -> timestamp + column; + case FLOAT -> Float.floatToRawIntBits((float) timestamp + column / 10.0F); + case DOUBLE -> Double.doubleToRawLongBits(timestamp + column / 10.0D); + default -> throw new AssertionError(type); + }; + } + return values; + } + + private static Object toTabletValue(TSDataType type, long value) { + return switch (type) { + case BOOLEAN -> value != 0; + case INT32 -> (int) value; + case INT64 -> value; + case FLOAT -> Float.intBitsToFloat((int) value); + case DOUBLE -> Double.longBitsToDouble(value); + default -> throw new AssertionError(type); + }; + } + + private static List<String> createColumnNames() { + final List<String> columnNames = new ArrayList<>(COLUMN_COUNT); + for (int column = 0; column < COLUMN_COUNT; column++) { + columnNames.add("v" + column); + } + return Collections.unmodifiableList(columnNames); + } + + private static List<TSDataType> createDataTypes() { + final List<TSDataType> dataTypes = new ArrayList<>(COLUMN_COUNT); + for (int column = 0; column < COLUMN_COUNT; column++) { + dataTypes.add(TYPE_SEQUENCE.get(column % TYPE_SEQUENCE.size())); + } + return Collections.unmodifiableList(dataTypes); + } + + private static void assertValidConfiguration() { + Assert.assertTrue(ROW_COUNT > 0); + Assert.assertTrue(COLUMN_COUNT > 0); + Assert.assertTrue(TABLET_ROW_COUNT > 0); + Assert.assertTrue(WARMUP_ROUND_COUNT >= 0); + Assert.assertTrue(ROUND_COUNT > 0); + Assert.assertTrue((long) WARMUP_ROUND_COUNT + ROUND_COUNT <= Integer.MAX_VALUE); + } + + private static void writeResult(BenchmarkResult result) throws IOException { + final Properties properties = new Properties(); + properties.setProperty("format.version", "2"); + properties.setProperty("java.version", System.getProperty("java.version")); + properties.setProperty("os.name", System.getProperty("os.name")); + properties.setProperty("os.arch", System.getProperty("os.arch")); + properties.setProperty( + "client.available.processors", + Integer.toString(Runtime.getRuntime().availableProcessors())); + properties.setProperty( + "client.max.heap.bytes", Long.toString(Runtime.getRuntime().maxMemory())); + properties.setProperty("client.count", "1"); + 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( + "insert.tablet.requests.per.round", + Integer.toString((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( + "numeric.row.round.p50.nanos", Long.toString(result.numericRowP50Nanos())); + properties.setProperty( + "numeric.row.round.p95.nanos", Long.toString(result.numericRowP95Nanos())); + properties.setProperty("numeric.row.rows.per.second", format(result.numericRowRowsPerSecond())); + properties.setProperty( + "numeric.row.values.per.second", format(result.numericRowValuesPerSecond())); + properties.setProperty("insert.tablet.round.p50.nanos", Long.toString(result.tabletP50Nanos())); + properties.setProperty("insert.tablet.round.p95.nanos", Long.toString(result.tabletP95Nanos())); + properties.setProperty("insert.tablet.rows.per.second", format(result.tabletRowsPerSecond())); + properties.setProperty( + "insert.tablet.values.per.second", format(result.tabletValuesPerSecond())); + + final Path resultPath = + Paths.get( + System.getProperty( + RESULT_PATH_PROPERTY, "target/numeric-row-insert-performance.properties")); + final Path parent = resultPath.getParent(); + if (parent != null) { + Files.createDirectories(parent); + } + try (OutputStream outputStream = Files.newOutputStream(resultPath)) { + properties.store(outputStream, null); + } + } + + private static String join(List<TSDataType> values) { + final StringBuilder builder = new StringBuilder(); + for (int index = 0; index < values.size(); index++) { + if (index > 0) { + builder.append(','); + } + builder.append(values.get(index)); + } + return builder.toString(); + } + + private static String join(long[] values) { + final StringBuilder builder = new StringBuilder(); + for (int index = 0; index < values.length; index++) { + if (index > 0) { + builder.append(','); + } + builder.append(values[index]); + } + return builder.toString(); + } + + private static String format(double value) { + return String.format(Locale.ROOT, "%.2f", value); + } + + private static long percentile(long[] latencies, double percentile) { + final long[] sorted = latencies.clone(); + Arrays.sort(sorted); + return sorted[(int) Math.ceil(percentile * sorted.length) - 1]; + } + + private static final class NumericRowWorkload { + + private final String tableName; + private final NumericRow[] rows; + + private NumericRowWorkload(String tableName, NumericRow[] rows) { + this.tableName = tableName; + this.rows = rows; + } + } + + private static final class TabletWorkload { + + private final String tableName; + private final List<Tablet> tablets; + + private TabletWorkload(String tableName, List<Tablet> tablets) { + this.tableName = tableName; + this.tablets = tablets; + } + } + + private static final class BenchmarkResult { + + private final long[] numericRowElapsedNanos; + private final long[] tabletElapsedNanos; + + private BenchmarkResult(long[] numericRowElapsedNanos, long[] tabletElapsedNanos) { + this.numericRowElapsedNanos = numericRowElapsedNanos; + this.tabletElapsedNanos = tabletElapsedNanos; + } + + private long numericRowP50Nanos() { + return percentile(numericRowElapsedNanos, 0.5D); + } + + private long numericRowP95Nanos() { + return percentile(numericRowElapsedNanos, 0.95D); + } + + private double numericRowRowsPerSecond() { + return ROW_COUNT * 1_000_000_000.0D / numericRowP50Nanos(); + } + + private double numericRowValuesPerSecond() { + return numericRowRowsPerSecond() * COLUMN_COUNT; + } + + private long tabletP50Nanos() { + return percentile(tabletElapsedNanos, 0.5D); + } + + private long tabletP95Nanos() { + return percentile(tabletElapsedNanos, 0.95D); + } + + private double tabletRowsPerSecond() { + return ROW_COUNT * 1_000_000_000.0D / tabletP50Nanos(); + } + + private double tabletValuesPerSecond() { + return tabletRowsPerSecond() * COLUMN_COUNT; + } + } +} 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 48a5a994dd5..f9c7ca00444 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 @@ -20,6 +20,7 @@ package org.apache.iotdb.relational.it.session; import org.apache.iotdb.isession.ITableSession; +import org.apache.iotdb.isession.NumericRow; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.isession.pool.ITableSessionPool; import org.apache.iotdb.it.env.EnvFactory; @@ -66,7 +67,7 @@ public class IoTDBTableModelSessionIT { @Test public void testNumericRowInsert() throws Exception { - // Exercise both table-session entry points, schema reordering and missing STRING columns. + // Exercise direct and pooled table sessions, schema reordering and missing STRING columns. // The same values must remain readable from MemTable and after flushing to TsFile. try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) { session.executeNonQueryStatement("CREATE DATABASE numeric_rows"); @@ -94,16 +95,26 @@ public class IoTDBTableModelSessionIT { 20260910, 1234 }; - session.insert("readings", 1, names, types, values); + session.insert( + new NumericRow( + "readings", + 1, + names, + types, + Collections.nCopies(types.size(), ColumnCategory.FIELD), + values, + new String[0])); try (ITableSessionPool pool = EnvFactory.getEnv().getTableSessionPool(1, "numeric_rows"); ITableSession pooledSession = pool.getSession()) { pooledSession.insert( - "readings", - 2, - names, - types, - Collections.nCopies(types.size(), ColumnCategory.FIELD), - values); + new NumericRow( + "readings", + 2, + names, + types, + Collections.nCopies(types.size(), ColumnCategory.FIELD), + values, + new String[0])); } assertNumericRows(session); session.executeNonQueryStatement("FLUSH"); @@ -127,28 +138,44 @@ public class IoTDBTableModelSessionIT { Arrays.asList( ColumnCategory.FIELD, ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.TAG); session.insert( - "readings", - 1, - names, - types, - categories, - new long[] {1, 10}, - new String[] {"设备甲😀", "华北"}); + new NumericRow( + "readings", + 1, + names, + types, + categories, + new long[] {1, 10}, + new String[] {"设备甲😀", "华北"})); try (ITableSessionPool pool = EnvFactory.getEnv().getTableSessionPool(1, "numeric_tag_rows"); ITableSession pooled = pool.getSession()) { pooled.insert( - "readings", - 1, - names, - types, - categories, - new long[] {0, 20}, - new String[] {"设备乙", "华北"}); + new NumericRow( + "readings", + 1, + names, + types, + categories, + new long[] {0, 20}, + new String[] {"设备乙", "华北"})); } session.insert( - "readings", 2, names, types, categories, new long[] {1, 30}, new String[] {"", null}); + new NumericRow( + "readings", + 2, + names, + types, + categories, + new long[] {1, 30}, + new String[] {"", null})); session.insert( - "readings", 2, names, types, categories, new long[] {0, 40}, new String[] {null, null}); + new NumericRow( + "readings", + 2, + names, + types, + categories, + new long[] {0, 40}, + new String[] {null, null})); assertNumericTagRows(session); session.executeNonQueryStatement("FLUSH"); assertNumericTagRows(session); 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 b5868e60ecd..af524f9a853 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 @@ -22,12 +22,8 @@ package org.apache.iotdb.isession; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; -import org.apache.tsfile.enums.ColumnCategory; -import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.write.record.Tablet; -import java.util.List; - /** * This interface defines a session for interacting with IoTDB tables. It supports operations such * as data insertion, executing queries, and closing the session. Implementations of this interface @@ -50,6 +46,18 @@ public interface ITableSession extends AutoCloseable { */ void insert(Tablet tablet) throws StatementExecutionException, IoTDBConnectionException; + /** + * Inserts one table-model {@link NumericRow} with numeric FIELDs and optional STRING TAGs. + * + * <p>This API requires a server supporting numeric row requests. The row's metadata and value + * arrays must not be modified until this call returns. + * + * @param row the numeric row to insert + * @throws StatementExecutionException if an error occurs while executing the statement + * @throws IoTDBConnectionException if there is an issue with the IoTDB connection + */ + void insert(NumericRow row) 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 new file mode 100644 index 00000000000..502c75e05e4 --- /dev/null +++ b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRow.java @@ -0,0 +1,97 @@ +/* + * 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.isession; + +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; + +import java.util.List; + +/** + * A table-model row with numeric FIELDs in {@code long[]} and STRING TAGs in {@code String[]}. + * + * <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. + * + * <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 + * Float#floatToRawIntBits(float)} in the low 32 bits and DOUBLE uses {@link + * Double#doubleToRawLongBits(double)}. + * + * <p>Metadata and arrays are retained without copying or boxing. Do not modify them while {@link + * ITableSession#insert(NumericRow)} is in progress. Arguments are validated when inserting the row. + */ +public final class NumericRow { + private final String tableName; + private final long timestamp; + private final List<String> columnNames; + private final List<TSDataType> dataTypes; + private final List<ColumnCategory> columnCategories; + private final long[] values; + private final String[] tagValues; + + /** Creates a row with numeric FIELDs and optional STRING TAGs in separate arrays. */ + public NumericRow( + String tableName, + long timestamp, + List<String> columnNames, + List<TSDataType> dataTypes, + List<ColumnCategory> columnCategories, + long[] values, + String[] tagValues) { + this.tableName = tableName; + this.timestamp = timestamp; + this.columnNames = columnNames; + this.dataTypes = dataTypes; + this.columnCategories = columnCategories; + this.values = values; + this.tagValues = tagValues; + } + + public String getTableName() { + return tableName; + } + + public long getTimestamp() { + return timestamp; + } + + public List<String> getColumnNames() { + return columnNames; + } + + public List<TSDataType> getDataTypes() { + return dataTypes; + } + + public List<ColumnCategory> getColumnCategories() { + return columnCategories; + } + + public long[] getValues() { + return values; + } + + public String[] getTagValues() { + return tagValues; + } +} 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 69b87b2db16..3eb19971a92 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_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."; // Session public static final String NODE_URLS_EMPTY = "nodeUrls shouldn't be empty."; @@ -286,4 +288,6 @@ public final class SessionMessages { public static final String EXCEPTION_RETRY_EXECUTE_STATEMENT_ARG_FAILED_ARG_TIMES_ARG_216C6873 = "retry to execute statement on %s failed %d times: %s"; public static final String EXCEPTION_SESSIONPOOL_DOESN_T_SUPPORT_EXECUTING_ARG_DIRECTLY_B778F701 = "SessionPool doesn't support executing %s directly"; + public static final String EXCEPTION_NUMERIC_ROW_ARGUMENTS_MUST_BE_NON_NULL_AND_COLUMN_NAMES_TYPES_CATEGORIES_AND_VALUES_MUST_HAVE_EQUAL_LENGTHS_AB102B01 = "Numeric row arguments must be non-null and column names, types, categories and values must have equal lengths."; + public static final String EXCEPTION_NUMERIC_ROWS_SUPPORT_ONLY_FIELD_COLUMNS_DA7A8EE7 = "Numeric rows support only FIELD columns."; } 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 2f28590ac88..22f3bea7821 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_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 列。"; // Session public static final String NODE_URLS_EMPTY = "nodeUrls 不能为空。"; @@ -284,4 +286,6 @@ public final class SessionMessages { public static final String EXCEPTION_RETRY_EXECUTE_STATEMENT_ARG_FAILED_ARG_TIMES_ARG_216C6873 = "在 %s 上重试执行语句失败 %d 次：%s"; public static final String EXCEPTION_SESSIONPOOL_DOESN_T_SUPPORT_EXECUTING_ARG_DIRECTLY_B778F701 = "SessionPool 不支持直接执行 %s"; + public static final String EXCEPTION_NUMERIC_ROW_ARGUMENTS_MUST_BE_NON_NULL_AND_COLUMN_NAMES_TYPES_CATEGORIES_AND_VALUES_MUST_HAVE_EQUAL_LENGTHS_AB102B01 = "数值行参数不能为 null，列名、类型、类别和值的数量必须相同。"; + public static final String EXCEPTION_NUMERIC_ROWS_SUPPORT_ONLY_FIELD_COLUMNS_DA7A8EE7 = "数值行仅支持 FIELD 列。"; } 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 39bb9c06cba..716d14fa802 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 @@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TAggregationType; import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.isession.INodeSupplier; import org.apache.iotdb.isession.ISession; +import org.apache.iotdb.isession.NumericRow; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.isession.template.Template; @@ -66,6 +67,7 @@ import org.apache.iotdb.session.util.ThreadUtils; import org.apache.thrift.TException; import org.apache.tsfile.common.conf.TSFileDescriptor; +import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.enums.CompressionType; @@ -2873,6 +2875,55 @@ public class Session implements ISession { } } + /** Inserts numeric FIELDs and STRING TAGs without constructing a boxed row. */ + public void insertRelationalRecord(NumericRow row) + throws IoTDBConnectionException, StatementExecutionException { + if (row == null || row.getTableName() == null) { + throw new IllegalArgumentException( + SessionMessages + .EXCEPTION_NUMERIC_ROW_ARGUMENTS_MUST_BE_NON_NULL_COLUMN_METADATA_LENGTHS_AND_PER_CATEGORY_VALUE_COUNTS_MUST_MATCH_EB12B233); + } + String tableName = row.getTableName(); + long timestamp = row.getTimestamp(); + List<String> measurements = row.getColumnNames(); + List<TSDataType> types = row.getDataTypes(); + List<ColumnCategory> columnCategories = row.getColumnCategories(); + long[] values = row.getValues(); + String[] tagValues = row.getTagValues(); + ByteBuffer valueBuffer = + SessionUtils.getValueBuffer(types, values, measurements, columnCategories, tagValues); + TSInsertRecordReq request = new TSInsertRecordReq(); + request.setPrefixPath(tableName); + request.setTimestamp(timestamp); + request.setMeasurements(measurements); + request.setValues(valueBuffer); + request.setIsWriteToTable(true); + request.setIsNumericRow(true); + request.setIsAligned(true); + request.setColumnCategoryies( + columnCategories.stream() + .map(category -> (byte) category.ordinal()) + .collect(Collectors.toList())); + String[] deviceSegments = new String[tagValues.length + 1]; + deviceSegments[0] = tableName; + System.arraycopy(tagValues, 0, deviceSegments, 1, tagValues.length); + IDeviceID deviceId = IDeviceID.Factory.DEFAULT_FACTORY.create(deviceSegments); + try { + getSessionConnection(deviceId).insertRecord(request); + } catch (RedirectException e) { + handleRedirection(deviceId, e.getEndPoint()); + } catch (IoTDBConnectionException e) { + if (!enableRedirection || tableModelDeviceIdToEndpoint.remove(deviceId) == null) { + throw e; + } + try { + getDefaultSessionConnection().insertRecord(request); + } catch (RedirectException redirect) { + handleRedirection(deviceId, redirect.getEndPoint()); + } + } + } + private void insertRelationalTabletWithLeaderCache(Tablet tablet) throws IoTDBConnectionException, StatementExecutionException { Map<SessionConnection, Tablet> relationalTabletGroup = new HashMap<>(); 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 059fdea3469..38cc6d72cea 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 @@ -20,6 +20,7 @@ package org.apache.iotdb.session; import org.apache.iotdb.isession.ITableSession; +import org.apache.iotdb.isession.NumericRow; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; @@ -39,6 +40,11 @@ public class TableSession implements ITableSession { session.insertRelationalTablet(tablet); } + @Override + public void insert(NumericRow row) throws StatementExecutionException, IoTDBConnectionException { + session.insertRelationalRecord(row); + } + @Override public void executeNonQueryStatement(String sql) throws IoTDBConnectionException, StatementExecutionException { 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 4cf7ca980f3..8f158f4ab4f 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 @@ -20,6 +20,7 @@ package org.apache.iotdb.session.pool; import org.apache.iotdb.isession.ITableSession; +import org.apache.iotdb.isession.NumericRow; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; @@ -68,6 +69,18 @@ public class TableSessionWrapper implements ITableSession { } } + @Override + public void insert(NumericRow row) throws StatementExecutionException, IoTDBConnectionException { + try { + session.insertRelationalRecord(row); + } catch (IoTDBConnectionException e) { + sessionPool.cleanSessionAndMayThrowConnectionException(session); + closed.set(true); + session = null; + throw e; + } + } + @Override public SessionDataSet executeQueryStatement(String sql) throws StatementExecutionException, IoTDBConnectionException { 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 b3b29f0be93..0d838b69254 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 @@ -163,6 +163,80 @@ public class SessionUtils { return buffer; } + /** Encodes separate, densely packed FIELD and TAG arrays in the metadata column order. */ + public static ByteBuffer getValueBuffer( + List<TSDataType> types, + long[] values, + List<String> measurements, + List<ColumnCategory> categories, + String[] tagValues) + throws IoTDBConnectionException { + if (types == null + || values == null + || measurements == null + || categories == null + || tagValues == null + || types.size() != measurements.size() + || types.size() != categories.size()) { + throw invalidNumericRowArguments(); + } + int fieldCount = 0; + int tagCount = 0; + int length = types.size(); + for (int i = 0; i < types.size(); i++) { + TSDataType type = types.get(i); + if (categories.get(i) == ColumnCategory.TAG && type == TSDataType.STRING) { + if (tagCount >= tagValues.length) { + throw invalidNumericRowArguments(); + } + length += ReadWriteIOUtils.sizeToWrite(tagValues[tagCount++]); + } else if (categories.get(i) == ColumnCategory.FIELD && type != null) { + fieldCount++; + length += + 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); + }; + } else { + throw new IllegalArgumentException( + SessionMessages + .EXCEPTION_NUMERIC_ROWS_SUPPORT_NUMERIC_FIELD_COLUMNS_AND_STRING_TAG_COLUMNS_ONLY_87159FFA); + } + } + if (fieldCount != values.length || tagCount != tagValues.length) { + throw invalidNumericRowArguments(); + } + ByteBuffer buffer = ByteBuffer.allocate(length); + int fieldIndex = 0; + int 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. + ReadWriteIOUtils.write(tagValues[tagIndex++], buffer); + } else { + long value = values[fieldIndex++]; + switch (type) { + case BOOLEAN -> buffer.put((byte) (value == 0 ? 0 : 1)); + case INT32, DATE, FLOAT -> buffer.putInt((int) value); + case INT64, TIMESTAMP, DOUBLE -> buffer.putLong(value); + default -> throw new IoTDBConnectionException(MSG_UNSUPPORTED_DATA_TYPE + type); + } + } + } + buffer.flip(); + return buffer; + } + + private static IllegalArgumentException invalidNumericRowArguments() { + return new IllegalArgumentException( + SessionMessages + .EXCEPTION_NUMERIC_ROW_ARGUMENTS_MUST_BE_NON_NULL_COLUMN_METADATA_LENGTHS_AND_PER_CATEGORY_VALUE_COUNTS_MUST_MATCH_EB12B233); + } + 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/SessionTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionTest.java index e4d2dbbb41d..88280ac8b4b 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 @@ -22,13 +22,18 @@ package org.apache.iotdb.session; import org.apache.iotdb.common.rpc.thrift.TAggregationType; import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.isession.ISession; +import org.apache.iotdb.isession.ITableSession; +import org.apache.iotdb.isession.NumericRow; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.isession.util.Version; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq; import org.apache.iotdb.service.rpc.thrift.TSQueryTemplateResp; +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.file.metadata.enums.CompressionType; import org.apache.tsfile.file.metadata.enums.TSEncoding; @@ -41,6 +46,7 @@ import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.Mockito; import org.mockito.MockitoAnnotations; @@ -134,6 +140,80 @@ public class SessionTest { .build(); } + @Test + public void testNumericTableRowRequest() throws Exception { + // A BOOLEAN/DOUBLE row must carry the table RPC flags and explicit FIELD categories. + ITableSession tableSession = new TableSession((Session) session); + List<String> names = Arrays.asList("enabled", "temperature"); + List<TSDataType> types = Arrays.asList(TSDataType.BOOLEAN, TSDataType.DOUBLE); + tableSession.insert( + new NumericRow( + "readings", + 123L, + names, + types, + Collections.nCopies(types.size(), ColumnCategory.FIELD), + new long[] {1L, Double.doubleToRawLongBits(-0.0)}, + new String[0])); + 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()); + Assert.assertTrue(request.isIsAligned()); + assertEquals("readings", request.getPrefixPath()); + assertEquals(123L, request.getTimestamp()); + assertEquals(names, request.getMeasurements()); + assertEquals( + Collections.nCopies(2, (byte) ColumnCategory.FIELD.ordinal()), + request.getColumnCategoryies()); + } + + @Test + public void testNumericTableRowRejectsNonFieldColumnsBeforeRpc() throws Exception { + // Numeric TAG values are invalid and must fail locally without sending a request. + ITableSession tableSession = new TableSession((Session) session); + Assert.assertThrows( + IllegalArgumentException.class, () -> tableSession.insert((NumericRow) null)); + Assert.assertThrows( + IllegalArgumentException.class, + () -> + tableSession.insert( + new NumericRow( + "readings", + 1L, + Collections.singletonList("tag"), + Collections.singletonList(TSDataType.INT64), + Collections.singletonList(ColumnCategory.TAG), + new long[] {1}, + new String[0]))); + Mockito.verify(sessionConnection, Mockito.never()).insertRecord(any(TSInsertRecordReq.class)); + } + + @Test + 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); + tableSession.insert( + new NumericRow( + "readings", 123L, names, types, categories, new long[] {1}, new String[] {"device"})); + 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()), + request.getColumnCategoryies()); + assertEquals( + SessionUtils.getValueBuffer( + types, new long[] {1}, names, categories, new String[] {"device"}), + request.bufferForValues()); + } + @Test public void testTimeZone() throws IoTDBConnectionException, StatementExecutionException { String timeZone = session.getTimeZone(); 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 b8524d4b311..d2c24a4da8e 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 @@ -28,6 +28,7 @@ import org.apache.tsfile.file.metadata.enums.CompressionType; import org.apache.tsfile.file.metadata.enums.TSEncoding; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.utils.ReadWriteIOUtils; import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.IMeasurementSchema; import org.apache.tsfile.write.schema.MeasurementSchema; @@ -45,6 +46,156 @@ import static org.junit.Assert.assertThrows; public class SessionUtilsTest { + @Test + public void testNumericRowEncodingPreservesRawBits() throws IoTDBConnectionException { + List<TSDataType> types = + Arrays.asList( + TSDataType.BOOLEAN, + TSDataType.BOOLEAN, + TSDataType.INT32, + TSDataType.INT64, + TSDataType.FLOAT, + TSDataType.DOUBLE, + TSDataType.DATE, + TSDataType.TIMESTAMP); + // Noncanonical NaN payloads must survive encoding without a float/double round trip. + long[] values = { + 0, + -7, + Integer.MIN_VALUE, + Long.MIN_VALUE, + 0x7fc01234, + 0x7ff8000000001234L, + 20260910, + Long.MAX_VALUE + }; + ByteBuffer actual = + SessionUtils.getValueBuffer( + types, + values, + Collections.nCopies(8, "s"), + 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); + expected.flip(); + Assert.assertEquals(expected, actual); + } + + @Test + public void testNumericRowRejectsInvalidValues() { + // Reject mismatched lengths, null types and every unsupported type before encoding. + assertThrows( + IllegalArgumentException.class, + () -> + SessionUtils.getValueBuffer( + Collections.singletonList(TSDataType.INT64), + new long[0], + Collections.singletonList("s"), + Collections.singletonList(ColumnCategory.FIELD), + new String[0])); + assertThrows( + IllegalArgumentException.class, + () -> + SessionUtils.getValueBuffer( + Collections.singletonList(null), + new long[] {1}, + Collections.singletonList("s"), + Collections.singletonList(ColumnCategory.FIELD), + new String[0])); + for (TSDataType type : + Arrays.asList( + TSDataType.TEXT, + TSDataType.STRING, + TSDataType.BLOB, + TSDataType.OBJECT, + TSDataType.VECTOR, + TSDataType.UNKNOWN)) { + assertThrows( + IoTDBConnectionException.class, + () -> + SessionUtils.getValueBuffer( + Collections.singletonList(type), + new long[] {1}, + Collections.singletonList("s"), + Collections.singletonList(ColumnCategory.FIELD), + new String[0])); + } + } + + @Test + public void testNumericRowWithSeparateTags() throws Exception { + // TAG strings, including null and empty, must not consume slots from the compact long[]. + List<TSDataType> types = + Arrays.asList( + TSDataType.STRING, + TSDataType.DOUBLE, + TSDataType.STRING, + TSDataType.BOOLEAN, + TSDataType.STRING); + List<ColumnCategory> categories = + Arrays.asList( + ColumnCategory.TAG, + ColumnCategory.FIELD, + ColumnCategory.TAG, + ColumnCategory.FIELD, + ColumnCategory.TAG); + ByteBuffer buffer = + SessionUtils.getValueBuffer( + types, + new long[] {0x7ff8000000001234L, -1}, + Arrays.asList("a", "d", "b", "enabled", "c"), + categories, + new String[] {"设备😀", null, ""}); + Assert.assertEquals(TSDataType.STRING.serialize(), buffer.get()); + Assert.assertEquals("设备😀", ReadWriteIOUtils.readString(buffer)); + Assert.assertEquals(TSDataType.DOUBLE.serialize(), buffer.get()); + 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()); + } + + @Test + public void testNumericRowRejectsInvalidTagArrays() { + // Mismatched per-category counts and ATTRIBUTE values must fail before sending the RPC. + List<TSDataType> types = Arrays.asList(TSDataType.STRING, TSDataType.INT64); + List<String> names = Arrays.asList("tag", "field"); + List<ColumnCategory> categories = Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD); + for (String[] tags : Arrays.asList(new String[0], new String[] {"a", "b"})) { + assertThrows( + IllegalArgumentException.class, + () -> SessionUtils.getValueBuffer(types, new long[] {1}, names, categories, tags)); + } + assertThrows( + IllegalArgumentException.class, + () -> + SessionUtils.getValueBuffer(types, new long[0], names, categories, new String[] {"a"})); + assertThrows( + IllegalArgumentException.class, + () -> SessionUtils.getValueBuffer(types, new long[] {1}, names, categories, null)); + assertThrows( + IllegalArgumentException.class, + () -> + SessionUtils.getValueBuffer( + types, + new long[] {1}, + names, + Arrays.asList(ColumnCategory.ATTRIBUTE, ColumnCategory.FIELD), + new String[] {"a"})); + } + @Test public void testGetTimeBuffer() { List<IMeasurementSchema> schemas = new ArrayList<>(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java index 3bebbc91153..dc8a8877b6d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java @@ -46,10 +46,12 @@ import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager; import org.apache.iotdb.db.pipe.resource.memory.InsertNodeMemoryEstimator; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil; import org.apache.iotdb.db.pipe.resource.memory.PipeTabletMemoryBlock; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertNumericRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; @@ -366,8 +368,8 @@ public class PipeInsertNodeTabletInsertionEvent extends PipeInsertionEvent return true; } - if (insertNode instanceof InsertRowNode) { - final long timestamp = ((InsertRowNode) insertNode).getTime(); + if (insertNode instanceof AbstractInsertRowNode row) { + final long timestamp = row.getTime(); return startTime <= timestamp && timestamp <= endTime; } @@ -409,13 +411,14 @@ public class PipeInsertNodeTabletInsertionEvent extends PipeInsertionEvent try { final InsertNode insertNode = getInsertNode(); - if (insertNode instanceof RelationalInsertRowNode + if (insertNode instanceof RelationalInsertNumericRowNode + || insertNode instanceof RelationalInsertRowNode || insertNode instanceof RelationalInsertTabletNode || insertNode instanceof RelationalInsertRowsNode) { return true; } - if (insertNode instanceof InsertRowNode || insertNode instanceof InsertTabletNode) { + if (insertNode instanceof AbstractInsertRowNode || insertNode instanceof InsertTabletNode) { final PartialPath devicePartialPath = insertNode.getTargetPath(); return Objects.isNull(devicePartialPath) || treePattern.mayOverlapWithDevice(devicePartialPath.getIDeviceIDAsFullDevice()); @@ -539,6 +542,7 @@ public class PipeInsertNodeTabletInsertionEvent extends PipeInsertionEvent pipeTaskMeta, this, insertRowNode, treePattern, userEntity)); } break; + case RELATIONAL_INSERT_NUMERIC_ROW: case RELATIONAL_INSERT_ROW: case RELATIONAL_INSERT_TABLET: eventParsers.add( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/parser/TabletInsertionEventParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/parser/TabletInsertionEventParser.java index 545a587aec4..886b070a84d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/parser/TabletInsertionEventParser.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/parser/TabletInsertionEventParser.java @@ -25,7 +25,7 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.pipe.api.access.Row; import org.apache.iotdb.pipe.api.collector.RowCollector; @@ -115,7 +115,7 @@ public abstract class TabletInsertionEventParser { //////////////////////////// parse //////////////////////////// - protected void parse(final InsertRowNode insertRowNode) { + protected void parse(final AbstractInsertRowNode insertRowNode) { final String[] originColumnNameStringList = insertRowNode.getMeasurements(); final int originColumnSize = originColumnNameStringList == null ? 0 : originColumnNameStringList.length; @@ -133,7 +133,8 @@ public abstract class TabletInsertionEventParser { final MeasurementSchema[] originMeasurementSchemaList = insertRowNode.getMeasurementSchemas(); final TsTableColumnCategory[] originColumnCategories = insertRowNode.getColumnCategories(); final TSDataType[] originValueDataTypes = insertRowNode.getDataTypes(); - final Object[] originValues = insertRowNode.getValues(); + // Read individual columns so numeric rows never need an Object[] row snapshot. + final int originValueCount = insertRowNode.getValueCount(); generateColumnIndexMapper( originColumnNameStringList, originColumnIndex2FilteredColumnIndexMapperList); @@ -147,8 +148,7 @@ public abstract class TabletInsertionEventParser { originMeasurementSchemaList, originValueDataTypes, i) - && originValues != null - && i < originValues.length); + && i < originValueCount); this.measurementSchemaList = new MeasurementSchema[filteredColumnSize]; this.columnNameStringList = new String[filteredColumnSize]; @@ -170,7 +170,7 @@ public abstract class TabletInsertionEventParser { : ColumnCategory.FIELD; this.valueColumnDataTypes[filteredColumnIndex] = originValueDataTypes[i]; final BitMap bitMap = new BitMap(this.timestampColumn.length); - if (Objects.isNull(originValues[i]) || Objects.isNull(originValueDataTypes[i])) { + if (insertRowNode.isNull(i) || Objects.isNull(originValueDataTypes[i])) { fillNullValue( originValueDataTypes[i], this.valueColumns, @@ -180,7 +180,12 @@ public abstract class TabletInsertionEventParser { } else { this.valueColumns[filteredColumnIndex] = filterValueColumnsByRowIndexList( - originValueDataTypes[i], originValues[i], rowIndexList, true, null, bitMap); + originValueDataTypes[i], + insertRowNode.getValue(i), + rowIndexList, + true, + null, + bitMap); } this.nullValueColumnBitmaps[filteredColumnIndex] = bitMap; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/parser/TabletInsertionEventTablePatternParser.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/parser/TabletInsertionEventTablePatternParser.java index 14708aa4be9..84dfb70bacc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/parser/TabletInsertionEventTablePatternParser.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/parser/TabletInsertionEventTablePatternParser.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.pipe.event.EnrichedEvent; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.event.common.tablet.PipeTabletUtils; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertNumericRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.pipe.api.access.Row; @@ -60,7 +61,9 @@ public class TabletInsertionEventTablePatternParser extends TabletInsertionEvent super(pipeTaskMeta, sourceEvent); this.pattern = pattern; - if (insertNode instanceof RelationalInsertRowNode) { + if (insertNode instanceof RelationalInsertNumericRowNode numeric) { + parse(numeric); + } else if (insertNode instanceof RelationalInsertRowNode) { parse((RelationalInsertRowNode) insertNode); } else if (insertNode instanceof RelationalInsertTabletNode) { parse((RelationalInsertTabletNode) insertNode); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipePlanToStatementVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipePlanToStatementVisitor.java index 6158c19bb99..210defbeb70 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipePlanToStatementVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/visitor/PipePlanToStatementVisitor.java @@ -42,6 +42,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNod import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertNumericRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceAttributeUpdateNode; @@ -50,6 +51,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Delete; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Update; import org.apache.iotdb.db.queryengine.plan.statement.Statement; import org.apache.iotdb.db.queryengine.plan.statement.crud.DeleteDataStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertNumericRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; @@ -84,6 +86,19 @@ public class PipePlanToStatementVisitor implements PlanVisitor<Object, Void> { node)); } + @Override + public InsertNumericRowStatement visitRelationalInsertNumericRow( + final RelationalInsertNumericRowNode node, final Void context) { + InsertNumericRowStatement statement = new InsertNumericRowStatement(); + statement.setDevicePath(node.getTargetPath()); + statement.setTime(node.getTime()); + statement.setMeasurements(node.getMeasurements()); + statement.setDataTypes(node.getDataTypes()); + statement.setNumericValues(node.getNumericValues()); + statement.setColumnCategories(node.getColumnCategories()); + return statement; + } + @Override public InsertRowStatement visitInsertRow(final InsertRowNode node, final Void context) { // MeasurementSchema is only allowed to be set at the analysis type stage. However, InsertNode diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2.java index 2cec219fa75..e8b765366c1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2.java @@ -28,6 +28,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.PlanFragment; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertMultiTabletsStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertNumericRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; @@ -78,9 +79,12 @@ public class PipeTransferTabletBatchReqV2 extends TPipeTransferReq { continue; } if (statement.isWriteToTable()) { - if (statement instanceof InsertRowStatement) { - addTableModelInsertRowStatement( - tableModelDatabaseInsertRowStatementMap, (InsertRowStatement) statement); + if (statement instanceof InsertNumericRowStatement) { + // Keep numeric rows independent: grouping into InsertRowsStatement would box their + // values. + statements.add(statement); + } else if (statement instanceof InsertRowStatement row) { + addTableModelInsertRowStatement(tableModelDatabaseInsertRowStatementMap, row); } else if (statement instanceof InsertTabletStatement) { statements.add(statement); } else if (statement instanceof InsertRowsStatement) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBinaryReq.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBinaryReq.java index cf4a746cb0e..cf89d64461a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBinaryReq.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBinaryReq.java @@ -24,8 +24,8 @@ import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.receiver.protocol.thrift.IoTDBDataNodeReceiver; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; @@ -55,7 +55,7 @@ public class PipeTransferTabletBinaryReq extends TPipeTransferReq { public InsertBaseStatement constructStatement() { final InsertNode insertNode = parseByteBuffer(); - if (!(insertNode instanceof InsertRowNode + if (!(insertNode instanceof AbstractInsertRowNode || insertNode instanceof InsertTabletNode || insertNode instanceof InsertRowsNode)) { throw new UnsupportedOperationException( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBinaryReqV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBinaryReqV2.java index 196af0b16d2..d764dc7ce98 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBinaryReqV2.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBinaryReqV2.java @@ -24,8 +24,8 @@ import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.receiver.protocol.thrift.IoTDBDataNodeReceiver; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; @@ -56,7 +56,7 @@ public class PipeTransferTabletBinaryReqV2 extends PipeTransferTabletBinaryReq { public InsertBaseStatement constructStatement() { final InsertNode insertNode = parseByteBuffer(); - if (!(insertNode instanceof InsertRowNode + if (!(insertNode instanceof AbstractInsertRowNode || insertNode instanceof InsertTabletNode || insertNode instanceof InsertRowsNode)) { throw new UnsupportedOperationException( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletInsertNodeReq.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletInsertNodeReq.java index 42f353c2506..40fd29ad98f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletInsertNodeReq.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletInsertNodeReq.java @@ -24,8 +24,8 @@ import org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.receiver.protocol.thrift.IoTDBDataNodeReceiver; import org.apache.iotdb.db.queryengine.plan.planner.plan.PlanFragment; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; @@ -51,7 +51,7 @@ public class PipeTransferTabletInsertNodeReq extends TPipeTransferReq { } public InsertBaseStatement constructStatement() { - if (!(insertNode instanceof InsertRowNode + if (!(insertNode instanceof AbstractInsertRowNode || insertNode instanceof InsertTabletNode || insertNode instanceof InsertRowsNode)) { throw new UnsupportedOperationException( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletInsertNodeReqV2.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletInsertNodeReqV2.java index 4fc18398dda..45dd2607ef6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletInsertNodeReqV2.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletInsertNodeReqV2.java @@ -25,8 +25,8 @@ import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.receiver.protocol.thrift.IoTDBDataNodeReceiver; import org.apache.iotdb.db.queryengine.plan.planner.plan.PlanFragment; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; @@ -56,7 +56,7 @@ public class PipeTransferTabletInsertNodeReqV2 extends PipeTransferTabletInsertN @Override public InsertBaseStatement constructStatement() { - if (!(insertNode instanceof InsertRowNode + if (!(insertNode instanceof AbstractInsertRowNode || insertNode instanceof InsertTabletNode || insertNode instanceof InsertRowsNode)) { throw new UnsupportedOperationException( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java index 6fc521a7f72..a49fd33161c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/executor/RegionWriteExecutor.java @@ -71,6 +71,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOf import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertNumericRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode; import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableNodeLocationAddNode; @@ -231,6 +232,12 @@ public class RegionWriteExecutor { } } + @Override + public RegionExecutionResult visitRelationalInsertNumericRow( + final RelationalInsertNumericRowNode node, final WritePlanNodeExecutionContext context) { + return executeDataInsert(node, context); + } + @Override public RegionExecutionResult visitInsertRow( final InsertRowNode node, final WritePlanNodeExecutionContext context) { 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 7464e6d99fb..0303eed2777 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 @@ -47,7 +47,9 @@ import org.apache.iotdb.db.queryengine.plan.statement.component.ResultColumn; import org.apache.iotdb.db.queryengine.plan.statement.component.SelectComponent; import org.apache.iotdb.db.queryengine.plan.statement.component.WhereCondition; import org.apache.iotdb.db.queryengine.plan.statement.crud.DeleteDataStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertMultiTabletsStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertNumericRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsOfOneDeviceStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; @@ -70,6 +72,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.metadata.template.ShowSche import org.apache.iotdb.db.queryengine.plan.statement.metadata.template.UnsetSchemaTemplateStatement; import org.apache.iotdb.db.schemaengine.schemaregion.utils.MetaFormatUtils; import org.apache.iotdb.db.schemaengine.template.TemplateQueryType; +import org.apache.iotdb.db.utils.NumericRowValues; import org.apache.iotdb.db.utils.QueryDataSetUtils; import org.apache.iotdb.db.utils.TabletDecoder; import org.apache.iotdb.mpp.rpc.thrift.TFetchTimeseriesReq; @@ -278,19 +281,38 @@ public class StatementGenerator { return queryStatement; } - public static InsertRowStatement createStatement(TSInsertRecordReq insertRecordReq) + public static InsertBaseStatement createStatement(TSInsertRecordReq insertRecordReq) throws IllegalPathException, QueryProcessException { final long startTime = System.nanoTime(); // construct insert statement - InsertRowStatement insertStatement = new InsertRowStatement(); + InsertBaseStatement insertStatement = + insertRecordReq.isIsNumericRow() + ? new InsertNumericRowStatement() + : new InsertRowStatement(); insertStatement.setDevicePath( - DEVICE_PATH_CACHE.getPartialPath(insertRecordReq.getPrefixPath())); + insertRecordReq.isIsWriteToTable() + ? new PartialPath(insertRecordReq.getPrefixPath(), false) + : DEVICE_PATH_CACHE.getPartialPath(insertRecordReq.getPrefixPath())); TimestampPrecisionUtils.checkTimestampPrecision(insertRecordReq.getTimestamp()); - insertStatement.setTime(insertRecordReq.getTimestamp()); + insertStatement.setMeasurements(insertRecordReq.getMeasurements().toArray(new String[0])); insertStatement.setAligned(insertRecordReq.isAligned); - insertStatement.fillValues(insertRecordReq.values); - if (insertRecordReq.isSetIsWriteToTable()) { + if (insertRecordReq.isIsNumericRow()) { + if (!insertRecordReq.isIsWriteToTable()) { + throw new IllegalArgumentException( + DataNodeQueryMessages + .QUERY_EXCEPTION_MISSING_OR_INVALID_COLUMN_CATEGORIES_FOR_TABLE_INSERTION_5DF990B9); + } + InsertNumericRowStatement numeric = (InsertNumericRowStatement) insertStatement; + numeric.setTime(insertRecordReq.getTimestamp()); + numeric.fillValues(insertRecordReq.values); + insertStatement.setAligned(true); + } else { + InsertRowStatement row = (InsertRowStatement) insertStatement; + row.setTime(insertRecordReq.getTimestamp()); + row.fillValues(insertRecordReq.values); + } + if (insertRecordReq.isIsWriteToTable()) { insertStatement.setWriteToTable(insertRecordReq.isIsWriteToTable()); if (!insertRecordReq.isSetColumnCategoryies() || insertRecordReq.getColumnCategoryiesSize() != insertRecordReq.getMeasurementsSize()) { @@ -305,6 +327,18 @@ public class StatementGenerator { TsTableColumnCategory.fromTsFileColumnCategory( ColumnCategory.values()[insertRecordReq.getColumnCategoryies().get(i).intValue()]); } + if (insertRecordReq.isIsNumericRow()) { + for (int i = 0; i < columnCategories.length; i++) { + TSDataType type = insertStatement.getDataType(i); + if (!(columnCategories[i] == TsTableColumnCategory.TAG && type == TSDataType.STRING) + && !(columnCategories[i] == TsTableColumnCategory.FIELD + && NumericRowValues.supports(type))) { + throw new IllegalArgumentException( + DataNodeQueryMessages + .QUERY_EXCEPTION_MISSING_OR_INVALID_COLUMN_CATEGORIES_FOR_TABLE_INSERTION_5DF990B9); + } + } + } insertStatement.setColumnCategories(columnCategories); } PERFORMANCE_OVERVIEW_METRICS.recordParseCost(System.nanoTime() - startTime); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java index a5b886ff4b0..864c7401c9c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java @@ -117,6 +117,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOf import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertNumericRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; @@ -581,6 +582,10 @@ public interface PlanVisitor<R, C> extends ICoreQueryPlanVisitor<R, C> { return visitPlan(node, context); } + default R visitRelationalInsertNumericRow(RelationalInsertNumericRowNode node, C context) { + return visitPlan(node, context); + } + default R visitRelationalInsertRow(RelationalInsertRowNode node, C context) { return visitInsertRow(node, context); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/AbstractInsertRowNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/AbstractInsertRowNode.java new file mode 100644 index 00000000000..f1c0785afa3 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/AbstractInsertRowNode.java @@ -0,0 +1,64 @@ +/* + * 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.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.LastCacheUpdateSource; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryValue; + +import org.apache.tsfile.enums.TSDataType; + +/** + * Storage-facing contract for single-row writes. Concrete nodes own their value representation and + * wire format; sharing this contract never converts a primitive row to an object row. + */ +public abstract class AbstractInsertRowNode extends InsertNode + implements WALEntryValue, LastCacheUpdateSource { + protected AbstractInsertRowNode(PlanNodeId id) { + super(id); + } + + protected AbstractInsertRowNode( + PlanNodeId id, + PartialPath path, + boolean aligned, + String[] measurements, + TSDataType[] dataTypes) { + super(id, path, aligned, measurements, dataTypes); + } + + public abstract long getTime(); + + public abstract int getValueCount(); + + public abstract boolean isNull(int index); + + /** Boxes only the requested cell for cache and Tablet consumers. */ + public abstract Object getValue(int index); + + /** Returns the original backing storage for allocation accounting, without copying it. */ + public abstract Object getValuesForMemoryAccounting(); + + public abstract void updateLastCache(String databaseName); + + public abstract void serializeToWAL(IWALByteBufferView buffer, long encodedSearchIndex); +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java index 777e0de391a..20bef3dc8ee 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java @@ -185,6 +185,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Explain; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExplainAnalyze; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.FetchDevice; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Insert; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertNumericRow; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertRow; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertRows; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertTablet; @@ -1223,6 +1224,11 @@ public class StatementAnalyzer { insert, scope, Field.newUnqualified(Insert.ROWS, Insert.ROWS_TYPE, Insert.ROWS_CATEGORY)); } + @Override + public Scope visitInsertNumericRow(InsertNumericRow node, Optional<Scope> context) { + return visitInsert(node, context); + } + @Override public Scope visitInsertRow(InsertRow node, Optional<Scope> context) { return visitInsert(node, context); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java index 6f5f3c278ed..c047be7c2b4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java @@ -100,6 +100,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedW import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertNumericRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; @@ -130,11 +131,13 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.rowpattern.rowpat import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AsofJoinOn; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AstVisitor; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Delete; +import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertNumericRow; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertRow; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertRows; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertTablet; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.LoadTsFile; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.PipeEnriched; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertNumericRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; @@ -1370,6 +1373,26 @@ public class RelationPlanner implements AstVisitor<RelationPlan, Void> { insertNode, analysis.getRootScope(), Collections.emptyList(), outerContext); } + @Override + public RelationPlan visitInsertNumericRow(InsertNumericRow node, Void context) { + InsertNumericRowStatement statement = node.getInnerTreeStatement(); + String[] measurements = statement.getMeasurements(); + MeasurementSchema[] schemas = statement.getMeasurementSchemas(); + stayConsistent(measurements, schemas); + RelationalInsertNumericRowNode insert = + new RelationalInsertNumericRowNode( + idAllocator.genPlanNodeId(), + statement.getDevicePath(), + measurements, + statement.getDataTypes(), + schemas, + statement.getTime(), + statement.getNumericValues(), + statement.getColumnCategories()); + insert.setFailedMeasurementNumber(statement.getFailedMeasurementNumber()); + return new RelationPlan(insert, analysis.getRootScope(), Collections.emptyList(), outerContext); + } + @Override public RelationPlan visitInsertRow(InsertRow node, Void context) { InsertRowStatement insertRowStatement = node.getInnerTreeStatement(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableModelPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableModelPlanner.java index cca616e0a3b..c91eb3e63fc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableModelPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/TableModelPlanner.java @@ -55,6 +55,7 @@ import org.apache.iotdb.db.queryengine.plan.scheduler.ClusterScheduler; import org.apache.iotdb.db.queryengine.plan.scheduler.IScheduler; import org.apache.iotdb.db.queryengine.plan.scheduler.load.LoadTsFileScheduler; 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.InsertRowsStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; import org.apache.iotdb.rpc.RpcUtils; @@ -239,6 +240,19 @@ public class TableModelPlanner implements IPlanner { ((WrappedInsertStatement) statementToRedirect).getInnerTreeStatement(); if (!analysis.isFinishQueryAfterAnalyze()) { + if (insertStatement instanceof InsertNumericRowStatement) { + // A numeric row uses insertRecord, whose redirect response carries one endpoint. + if (tsstatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + List<TEndPoint> endpoints = analysis.getRedirectNodeList(); + if (endpoints != null + && !endpoints.isEmpty() + && !localEndPoint.equals(endpoints.get(0))) { + tsstatus.setCode(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()); + tsstatus.setRedirectNode(endpoints.get(0)); + } + } + return; + } // Table Model Session supports insertTablet and pipe-generated insertRows statements. if (insertStatement instanceof InsertTabletStatement || insertStatement instanceof InsertRowsStatement) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertNumericRow.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertNumericRow.java new file mode 100644 index 00000000000..38f29f01d2c --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertNumericRow.java @@ -0,0 +1,71 @@ +/* + * 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.relational.sql.ast; + +import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.IAstVisitor; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertNumericRowStatement; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +/** Relational analysis entry for a primitive numeric row. */ +public class InsertNumericRow extends WrappedInsertStatement { + public InsertNumericRow(InsertNumericRowStatement statement, MPPQueryContext context) { + super(statement, context); + } + + @Override + public <R, C> R accept(IAstVisitor<R, C> visitor, C context) { + return ((AstVisitor<R, C>) visitor).visitInsertNumericRow(this, context); + } + + @Override + public InsertNumericRowStatement getInnerTreeStatement() { + return (InsertNumericRowStatement) super.getInnerTreeStatement(); + } + + @Override + public void updateAfterSchemaValidation(MPPQueryContext context) { + // Numeric rows are already typed; schema validation casts values in their primitive storage. + } + + @Override + public String getTableName() { + return getInnerTreeStatement().getTableName(); + } + + @Override + public List<Object[]> getDeviceIdList() { + Object[] segments = getInnerTreeStatement().getTableDeviceID().getSegments(); + return Collections.singletonList(Arrays.copyOfRange(segments, 1, segments.length)); + } + + @Override + public List<String> getAttributeColumnNameList() { + return Collections.emptyList(); + } + + @Override + public List<Object[]> getAttributeValueList() { + return Collections.singletonList(new Object[0]); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementType.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementType.java index 00d36a03e23..6488f76f56d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementType.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementType.java @@ -203,4 +203,5 @@ public enum StatementType { SHOW_EXTERNAL_SERVICE, SHOW_DISK_USAGE, + INSERT_NUMERIC_ROW, } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java index 5f2f72f4313..d107a5fcde8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java @@ -22,6 +22,7 @@ package org.apache.iotdb.db.queryengine.plan.statement; import org.apache.iotdb.db.queryengine.plan.statement.crud.DeleteDataStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertMultiTabletsStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertNumericRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsOfOneDeviceStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; @@ -405,6 +406,10 @@ public abstract class StatementVisitor<R, C> { return visitStatement(loadTsFileStatement, context); } + public R visitInsertNumericRow(InsertNumericRowStatement statement, C context) { + return visitInsertBase(statement, context); + } + public R visitInsertRow(InsertRowStatement insertRowStatement, C context) { return visitInsertBase(insertRowStatement, context); } 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 new file mode 100644 index 00000000000..db90503d464 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertNumericRowStatement.java @@ -0,0 +1,292 @@ +/* + * 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.statement.crud; + +import org.apache.iotdb.calc.exception.QueryProcessException; +import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; +import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.plan.relational.metadata.ColumnSchema; +import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; +import org.apache.iotdb.commons.utils.TimePartitionUtils; +import org.apache.iotdb.db.i18n.DataNodeQueryMessages; +import org.apache.iotdb.db.pipe.resource.memory.InsertNodeMemoryEstimator; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaValidation; +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.StatementVisitor; +import org.apache.iotdb.db.utils.NumericRowValues; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.utils.RamUsageEstimator; + +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; + +/** A table-only numeric row. Schema edits and partial failures keep the payload in long[]. */ +public class InsertNumericRowStatement extends InsertBaseStatement { + private static final long INSTANCE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(InsertNumericRowStatement.class); + private long time; + private NumericRowValues numericValues; + private IDeviceID deviceID; + + public InsertNumericRowStatement() { + statementType = StatementType.INSERT_NUMERIC_ROW; + writeToTable = true; + isAligned = true; + } + + public long getTime() { + return time; + } + + public void setTime(long time) { + this.time = time; + } + + public NumericRowValues getNumericValues() { + return numericValues; + } + + public long[] getValues() { + return numericValues.getValues(); + } + + public void setNumericValues(NumericRowValues values) { + numericValues = values; + deviceID = null; + } + + public int getValueCount() { + return numericValues == null ? 0 : numericValues.size(); + } + + /** 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(); + if (type < 0 || type >= TSDataType.values().length) { + throw new QueryProcessException(DataNodeQueryMessages.UNSUPPORTED_DATA_TYPE + type); + } + dataTypes[i] = TSDataType.values()[type]; + numericValues.read(i, dataTypes[i], buffer); + } + } + + @Override + public <R, C> R accept(StatementVisitor<R, C> visitor, C context) { + return visitor.visitInsertNumericRow(this, context); + } + + @Override + public Statement toRelationalStatement(MPPQueryContext context) { + return new InsertNumericRow(this, context); + } + + @Override + public List<PartialPath> getPaths() { + return Collections.emptyList(); + } + + @Override + public long getMinTime() { + return time; + } + + public TTimePartitionSlot getTimePartitionSlot() { + return TimePartitionUtils.getTimePartitionSlot(time); + } + + @Override + public boolean isEmpty() { + return getValueCount() == 0; + } + + @Override + public Object getFirstValueOfIndex(int index) { + return index < 0 || index >= getValueCount() + ? null + : numericValues.getObject(index, getDataType(index)); + } + + @Override + public boolean isColumnPresent(int index) { + return super.isColumnPresent(index) && index < getValueCount(); + } + + @Override + protected boolean checkAndCastDataType(int index, TSDataType type) { + if (index < 0 + || index >= getValueCount() + || !numericValues.cast(index, getDataType(index), type)) { + return false; + } + dataTypes[index] = type; + return true; + } + + @Override + public void semanticCheck() { + super.semanticCheck(); + if (measurements.length != getValueCount()) { + throw new SemanticException( + String.format( + DataNodeQueryMessages + .THE_MEASUREMENTLIST_S_SIZE_D_IS_NOT_CONSISTENT_WITH_THE_VALUELIST_S_SIZE_D, + measurements.length, + getValueCount())); + } + } + + @Override + public void markFailedMeasurement(int index, Exception cause) { + if (!isColumnPresent(index)) { + return; + } + if (failedMeasurementIndex2Info == null) { + failedMeasurementIndex2Info = new HashMap<>(); + } + failedMeasurementIndex2Info.putIfAbsent( + index, + new FailedMeasurementInfo( + measurements[index], getDataType(index), getFirstValueOfIndex(index), cause)); + measurements[index] = null; + if (dataTypes != null && index < dataTypes.length) { + dataTypes[index] = null; + } + numericValues.setNull(index); + tagColumnIndices = null; + deviceID = null; + } + + @Override + public void removeAllFailedMeasurementMarks() { + if (failedMeasurementIndex2Info == null) { + return; + } + failedMeasurementIndex2Info.forEach( + (index, info) -> { + measurements[index] = info.getMeasurement(); + dataTypes[index] = info.getDataType(); + numericValues.setObject(index, info.getDataType(), info.getValue()); + }); + failedMeasurementIndex2Info.clear(); + tagColumnIndices = null; + deviceID = null; + } + + @Override + public void insertColumn(int pos, ColumnSchema columnSchema) { + super.insertColumn(pos, columnSchema); + int[] mapping = new int[measurements.length]; + for (int i = 0; i < mapping.length; i++) { + mapping[i] = i < pos ? i : i == pos ? -1 : i - 1; + } + numericValues = numericValues.remap(mapping); + deviceID = null; + } + + @Override + public void swapColumn(int src, int target) { + super.swapColumn(src, target); + numericValues.swap(src, target); + deviceID = null; + } + + @Override + public void rebuildArraysAfterExpansion(int[] mapping, String[] newMeasurements) { + super.rebuildArraysAfterExpansion(mapping, newMeasurements); + numericValues = numericValues.remap(mapping); + deviceID = null; + } + + @Override + protected void subRemoveAttributeColumns(List<Integer> columnsToKeep) { + numericValues = + numericValues.remap(columnsToKeep.stream().mapToInt(Integer::intValue).toArray()); + deviceID = null; + } + + @Override + public void setColumnCategories(TsTableColumnCategory[] categories) { + super.setColumnCategories(categories); + deviceID = null; + } + + public IDeviceID getTableDeviceID() { + if (deviceID == null) { + String[] segments = new String[getTagColumnIndices().size() + 1]; + segments[0] = getTableName(); + for (int i = 0; i < getTagColumnIndices().size(); i++) { + segments[i + 1] = numericValues.getTag(getTagColumnIndices().get(i)); + } + deviceID = IDeviceID.Factory.DEFAULT_FACTORY.create(segments); + } + return deviceID; + } + + @Override + protected long calculateBytesUsed() { + return INSTANCE_SIZE + + (numericValues == null ? 0 : numericValues.ramBytesUsed()) + + InsertNodeMemoryEstimator.sizeOfIDeviceID(deviceID); + } + + // Tree schema validation and logical view splitting do not apply to numeric table rows. + @Override + public ISchemaValidation getSchemaValidation() { + throw new UnsupportedOperationException(); + } + + @Override + public List<ISchemaValidation> getSchemaValidationList() { + throw new UnsupportedOperationException(); + } + + @Override + public InsertBaseStatement removeLogicalView() { + return this; + } + + @Override + public String getPipeLoggingString() { + return "InsertNumericRowStatement{table=" + + getTableName() + + ", time=" + + time + + ", valueLength=" + + getValueCount() + + '}'; + } + + @Override + public String toString() { + return getPipeLoggingString() + Arrays.toString(getValues()); + } +} 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 1ac6c15488b..ba5349edc0e 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/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index de1a5a09b7d..df24cefdf30 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -42,6 +42,7 @@ import org.apache.iotdb.db.queryengine.common.DeviceContext; import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; import org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet; import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; @@ -80,6 +81,7 @@ import org.apache.iotdb.db.utils.CommonUtils; import org.apache.iotdb.db.utils.EncryptDBUtils; import org.apache.iotdb.db.utils.MemUtils; import org.apache.iotdb.db.utils.ModificationUtils; +import org.apache.iotdb.db.utils.NumericRowValues; import org.apache.iotdb.db.utils.datastructure.AlignedTVList; import org.apache.iotdb.db.utils.datastructure.TVList; import org.apache.iotdb.rpc.RpcUtils; @@ -290,7 +292,7 @@ public class TsFileProcessor { } } - private static void clearDataRegionReplicaSet(final InsertRowNode insertRowNode) { + private static void clearDataRegionReplicaSet(final AbstractInsertRowNode insertRowNode) { insertRowNode.setDataRegionReplicaSet(null); } @@ -310,7 +312,7 @@ public class TsFileProcessor { * * @param insertRowNode physical plan of insertion */ - public void insert(InsertRowNode insertRowNode, long[] infoForMetrics) + public void insert(AbstractInsertRowNode insertRowNode, long[] infoForMetrics) throws WriteProcessException { ensureMemTable(infoForMetrics); @@ -329,7 +331,7 @@ public class TsFileProcessor { insertRowNode.getDeviceID(), insertRowNode.getMeasurements(), insertRowNode.getDataTypes(), - insertRowNode.getValues(), + insertRowNode.getValuesForMemoryAccounting(), insertRowNode.getColumnCategories()); } else { memIncrements = @@ -337,7 +339,7 @@ public class TsFileProcessor { insertRowNode.getDeviceID(), insertRowNode.getMeasurements(), insertRowNode.getDataTypes(), - insertRowNode.getValues(), + insertRowNode.getValuesForMemoryAccounting(), insertRowNode.getColumnCategories()); } // recordScheduleMemoryBlockCost @@ -390,7 +392,7 @@ public class TsFileProcessor { if (insertRowNode.isAligned()) { pointInserted = workMemTable.insertAlignedRow(insertRowNode); } else { - pointInserted = workMemTable.insert(insertRowNode); + pointInserted = workMemTable.insert((InsertRowNode) insertRowNode); } } finally { reconcileAlignedTVListRamCost(alignedRamCostSnapshot, memIncrements[0]); @@ -514,7 +516,7 @@ public class TsFileProcessor { if (insertRowNode.isAligned()) { pointInserted += workMemTable.insertAlignedRow(insertRowNode); } else { - pointInserted += workMemTable.insert(insertRowNode); + pointInserted += workMemTable.insert((InsertRowNode) insertRowNode); } // update start time of this memtable tsFileResource.updateStartTime(insertRowNode.getDeviceID(), insertRowNode.getTime()); @@ -770,7 +772,7 @@ public class TsFileProcessor { IDeviceID deviceId, String[] measurements, TSDataType[] dataTypes, - Object[] values, + Object values, TsTableColumnCategory[] columnCategories) throws WriteProcessException { // Fixed-size TVList structures and materialized value primitive arrays. @@ -798,8 +800,8 @@ public class TsFileProcessor { } } // TEXT data mem size - if (dataTypes[i].isBinary() && values[i] != null) { - textDataIncrement += MemUtils.getBinarySize((Binary) values[i]); + if (dataTypes[i].isBinary() && values instanceof Object[] rowValues && rowValues[i] != null) { + textDataIncrement += MemUtils.getBinarySize((Binary) rowValues[i]); } } updateMemoryInfo(memTableIncrement, chunkMetadataIncrement, textDataIncrement); @@ -818,7 +820,7 @@ public class TsFileProcessor { for (InsertRowNode insertRowNode : insertRowNodeList) { IDeviceID deviceId = insertRowNode.getDeviceID(); TSDataType[] dataTypes = insertRowNode.getDataTypes(); - Object[] values = insertRowNode.getValues(); + Object values = insertRowNode.getValuesForMemoryAccounting(); String[] measurements = insertRowNode.getMeasurements(); for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { // Skip failed Measurements @@ -852,8 +854,10 @@ public class TsFileProcessor { increasingMemTableInfo.get(deviceId).computeIfPresent(measurements[i], (k, v) -> v + 1); } // TEXT data mem size - if (dataTypes[i].isBinary() && values[i] != null) { - textDataIncrement += MemUtils.getBinarySize((Binary) values[i]); + if (dataTypes[i].isBinary() + && values instanceof Object[] rowValues + && rowValues[i] != null) { + textDataIncrement += MemUtils.getBinarySize((Binary) rowValues[i]); } } } @@ -866,7 +870,7 @@ public class TsFileProcessor { IDeviceID deviceId, String[] measurements, TSDataType[] dataTypes, - Object[] values, + Object values, TsTableColumnCategory[] columnCategories) throws WriteProcessException { // Fixed-size TVList structures and materialized value primitive arrays. @@ -915,9 +919,10 @@ public class TsFileProcessor { } for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { if (isWritableFieldMeasurement(measurements, dataTypes, values, columnCategories, i) - && dataTypes[i].isBinary()) { + && dataTypes[i].isBinary() + && values instanceof Object[] rowValues) { // Binary values keep their variable-length content outside the fixed Binary[] array. - textDataIncrement += MemUtils.getBinarySize((Binary) values[i]); + textDataIncrement += MemUtils.getBinarySize((Binary) rowValues[i]); } } updateMemoryInfo(memTableIncrement, chunkMetadataIncrement, textDataIncrement); @@ -939,7 +944,7 @@ public class TsFileProcessor { for (InsertRowNode insertRowNode : insertRowNodeList) { IDeviceID deviceId = insertRowNode.getDeviceID(); TSDataType[] dataTypes = insertRowNode.getDataTypes(); - Object[] values = insertRowNode.getValues(); + Object values = insertRowNode.getValuesForMemoryAccounting(); String[] measurements = insertRowNode.getMeasurements(); IWritableMemChunk memChunk = @@ -1031,9 +1036,11 @@ public class TsFileProcessor { measurements, dataTypes, values, insertRowNode.getColumnCategories(), i)) { continue; } - if (dataTypes[i].isBinary() && values[i] != null) { + if (dataTypes[i].isBinary() + && values instanceof Object[] rowValues + && rowValues[i] != null) { // Binary values keep their variable-length content outside the fixed Binary[] array. - textDataIncrement += MemUtils.getBinarySize((Binary) values[i]); + textDataIncrement += MemUtils.getBinarySize((Binary) rowValues[i]); } } } @@ -1326,7 +1333,7 @@ public class TsFileProcessor { private static TSDataType[] getWritableFieldDataTypes( String[] measurementIds, TSDataType[] dataTypes, - Object[] valuesOrColumns, + Object valuesOrColumns, TsTableColumnCategory[] columnCategories) { List<TSDataType> writableFieldDataTypes = new ArrayList<>(); for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { @@ -1341,13 +1348,15 @@ public class TsFileProcessor { private static boolean isWritableFieldMeasurement( String[] measurementIds, TSDataType[] dataTypes, - Object[] valuesOrColumns, + Object valuesOrColumns, TsTableColumnCategory[] columnCategories, int index) { return isFieldMeasurement(measurementIds, dataTypes, columnCategories, index) - && valuesOrColumns != null - && index < valuesOrColumns.length - && valuesOrColumns[index] != null; + && (valuesOrColumns instanceof NumericRowValues row + ? index < row.size() && !row.isNull(index) + : valuesOrColumns instanceof Object[] columns + && index < columns.length + && columns[index] != null); } private static boolean isFieldMeasurement( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALEntry.java index 8c0a5e35329..346e3007c7e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALEntry.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALEntry.java @@ -22,9 +22,9 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.buffer; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; import org.apache.iotdb.db.i18n.StorageEngineMessages; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ContinuousSameSearchIndexSeparatorNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode; @@ -65,7 +65,7 @@ public abstract class WALEntry implements SerializedSize { this.value = value; if (value instanceof IMemTable) { this.type = WALEntryType.MEMORY_TABLE_SNAPSHOT; - } else if (value instanceof InsertRowNode) { + } else if (value instanceof AbstractInsertRowNode) { this.type = WALEntryType.INSERT_ROW_NODE; } else if (value instanceof InsertTabletNode) { this.type = WALEntryType.INSERT_TABLET_NODE; @@ -121,7 +121,7 @@ public abstract class WALEntry implements SerializedSize { value = AbstractMemTable.Factory.createFromOldMemTableSnapshot(stream); break; case INSERT_ROW_NODE: - value = (InsertRowNode) PlanNodeType.deserializeFromWAL(stream); + value = (AbstractInsertRowNode) PlanNodeType.deserializeFromWAL(stream); break; case INSERT_TABLET_NODE: value = (InsertTabletNode) PlanNodeType.deserializeFromWAL(stream); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java index 957cf03d892..37cc8aff863 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java @@ -22,9 +22,9 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.buffer; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.StorageEngineMessages; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode; @@ -102,7 +102,7 @@ public class WALInfoEntry extends WALEntry { .serializeToWAL(buffer, tabletInfo.tabletRangeList, encodedSearchIndex); break; case INSERT_ROW_NODE: - ((InsertRowNode) value).serializeToWAL(buffer, encodedSearchIndex); + ((AbstractInsertRowNode) value).serializeToWAL(buffer, encodedSearchIndex); break; case INSERT_ROWS_NODE: ((InsertRowsNode) value).serializeToWAL(buffer, encodedSearchIndex); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/IWALNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/IWALNode.java index a8fbbee0dc4..b2689a5f99a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/IWALNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/IWALNode.java @@ -21,9 +21,9 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.node; import org.apache.iotdb.consensus.common.DataSet; import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ContinuousSameSearchIndexSeparatorNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode; @@ -38,7 +38,7 @@ import java.util.List; public interface IWALNode extends FlushListener, AutoCloseable, ConsensusReqReader, DataSet { /** Log InsertRowNode. */ - WALFlushListener log(long memTableId, InsertRowNode insertRowNode); + WALFlushListener log(long memTableId, AbstractInsertRowNode insertRowNode); /** Log InsertRowsNode. */ WALFlushListener log(long memTableId, InsertRowsNode insertRowsNode); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNode.java index d24d1a1d823..93ac79c05a9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALFakeNode.java @@ -20,9 +20,9 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.node; import org.apache.iotdb.db.i18n.StorageEngineMessages; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ContinuousSameSearchIndexSeparatorNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode; @@ -53,7 +53,7 @@ public class WALFakeNode implements IWALNode { } @Override - public WALFlushListener log(long memTableId, InsertRowNode insertRowNode) { + public WALFlushListener log(long memTableId, AbstractInsertRowNode insertRowNode) { return getResult(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java index 0e66509d350..825a83ffe2c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java @@ -29,9 +29,9 @@ import org.apache.iotdb.consensus.common.request.IoTConsensusRequest; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.StorageEngineMessages; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ContinuousSameSearchIndexSeparatorNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode; @@ -222,7 +222,7 @@ public class WALNode implements IWALNode { } @Override - public WALFlushListener log(long memTableId, InsertRowNode insertRowNode) { + public WALFlushListener log(long memTableId, AbstractInsertRowNode insertRowNode) { logger.debug( StorageEngineMessages.STORAGE_LOG_WAL_NODE_LOGS_INSERTROWNODE_THE_SEARCH_INDEX_IS_027450AC, identifier, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java index 221d39cdfcc..dbd26029418 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.service.metric.MetricService; import org.apache.iotdb.commons.service.metric.enums.Metric; import org.apache.iotdb.commons.service.metric.enums.Tag; import org.apache.iotdb.db.exception.WriteProcessException; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; @@ -96,8 +97,8 @@ public class TsFilePlanRedoer { // the last chunk group may contain the same data with the logs, ignore such logs in seq file Optional<Long> lastEndTime = tsFileResource.getEndTime(node.getDeviceID()); long minTimeInNode; - if (node instanceof InsertRowNode) { - minTimeInNode = ((InsertRowNode) node).getTime(); + if (node instanceof AbstractInsertRowNode row) { + minTimeInNode = row.getTime(); } else { minTimeInNode = ((InsertTabletNode) node).getTimes()[0]; } @@ -107,9 +108,9 @@ public class TsFilePlanRedoer { } int pointsInserted; - if (node instanceof InsertRowNode) { + if (node instanceof AbstractInsertRowNode row) { if (node.isAligned()) { - pointsInserted = recoveryMemTable.insertAlignedRow((InsertRowNode) node); + pointsInserted = recoveryMemTable.insertAlignedRow(row); } else { pointsInserted = recoveryMemTable.insert((InsertRowNode) node); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGeneratorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGeneratorTest.java index 5512e312cf0..7a56ac2ef03 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGeneratorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGeneratorTest.java @@ -271,7 +271,7 @@ public class StatementGeneratorTest { Arrays.asList("s1", "s2"), ByteBuffer.wrap(new byte[] {0, 0, 0, 0}), 1000L); - InsertRowStatement statement = StatementGenerator.createStatement(req); + InsertRowStatement statement = (InsertRowStatement) StatementGenerator.createStatement(req); assertEquals(1000L, statement.getTime()); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkNumericTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkNumericTest.java new file mode 100644 index 00000000000..8620f17f71b --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkNumericTest.java @@ -0,0 +1,134 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.storageengine.dataregion.memtable; + +import org.apache.iotdb.db.utils.NumericRowValues; +import org.apache.iotdb.db.utils.datastructure.AlignedTVList; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.Test; + +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.BitSet; +import java.util.Collections; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +public class AlignedWritableMemChunkNumericTest { + + @Test + public void testMappingAllocatedOnlyForNumericRowsAndReused() throws Exception { + // Ordinary row/Tablet writes must not allocate scratch space; numeric rows share it. + List<IMeasurementSchema> schemas = + Collections.singletonList(new MeasurementSchema("a", TSDataType.INT64)); + AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(new ArrayList<>(schemas), true); + try { + assertNull(mapping(chunk)); + chunk.writeAlignedPoints(1, new Object[] {10L}, schemas); + chunk.writeAlignedTablet( + new long[] {2}, new Object[] {new long[] {20}}, null, schemas, 0, 1, null); + assertNull(mapping(chunk)); + + chunk.writeAlignedNumericPoints(3, row(30), schemas); + int[] mapping = mapping(chunk); + assertNotNull(mapping); + chunk.writeAlignedNumericPoints(4, row(40), schemas); + assertSame(mapping, mapping(chunk)); + assertEquals(30, chunk.getWorkingTVList().getLongByValueIndex(2, 0)); + assertEquals(40, chunk.getWorkingTVList().getLongByValueIndex(3, 0)); + } finally { + chunk.release(); + } + } + + @Test + public void testMappingResetAndGrowthWithMixedWrites() throws Exception { + // Reordered/absent columns must stay correct after reuse and schema growth by either API. + IMeasurementSchema a = new MeasurementSchema("a", TSDataType.INT64); + IMeasurementSchema b = new MeasurementSchema("b", TSDataType.INT64); + IMeasurementSchema c = new MeasurementSchema("c", TSDataType.INT64); + IMeasurementSchema d = new MeasurementSchema("d", TSDataType.INT64); + AlignedWritableMemChunk chunk = + new AlignedWritableMemChunk(new ArrayList<>(Arrays.asList(a, b)), true); + try { + chunk.writeAlignedNumericPoints(1, row(20, 10), Arrays.asList(b, a)); + int[] mapping = mapping(chunk); + chunk.writeAlignedNumericPoints(2, row(30, 999), Arrays.asList(a, null)); + assertSame(mapping, mapping(chunk)); + AlignedTVList list = chunk.getWorkingTVList(); + assertEquals(10, list.getLongByValueIndex(0, 0)); + assertEquals(20, list.getLongByValueIndex(0, 1)); + assertEquals(30, list.getLongByValueIndex(1, 0)); + assertTrue(list.isNullValue(1, 1)); + + chunk.writeAlignedNumericPoints(3, row(40, 50), Arrays.asList(c, b)); + int[] extendedMapping = mapping(chunk); + assertEquals(3, extendedMapping.length); + assertTrue(list.isNullValue(2, 0)); + assertEquals(50, list.getLongByValueIndex(2, 1)); + assertEquals(40, list.getLongByValueIndex(2, 2)); + + chunk.writeAlignedTablet( + new long[] {4}, + new Object[] {new long[] {60}}, + null, + Collections.singletonList(d), + 0, + 1, + null); + assertSame(extendedMapping, mapping(chunk)); + chunk.writeAlignedNumericPoints(5, row(70), Collections.singletonList(a)); + assertEquals(4, mapping(chunk).length); + assertEquals(60, list.getLongByValueIndex(3, 3)); + assertEquals(70, list.getLongByValueIndex(4, 0)); + assertTrue(list.isNullValue(4, 1)); + assertTrue(list.isNullValue(4, 2)); + assertTrue(list.isNullValue(4, 3)); + + mapping = mapping(chunk); + chunk.writeAlignedNumericPoints(6, row(80), Collections.singletonList(d)); + assertSame(mapping, mapping(chunk)); + assertTrue(list.isNullValue(5, 0)); + assertEquals(80, list.getLongByValueIndex(5, 3)); + } finally { + chunk.release(); + } + } + + private static NumericRowValues row(long... values) { + return new NumericRowValues(values, new BitSet()); + } + + private static int[] mapping(AlignedWritableMemChunk chunk) throws Exception { + // Array identity verifies the allocation contract without exposing scratch state as public API. + Field field = AlignedWritableMemChunk.class.getDeclaredField("numericColumnMapping"); + field.setAccessible(true); + return (int[]) field.get(chunk); + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/planner/plan/node/PlanNodeType.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/planner/plan/node/PlanNodeType.java index e36796fca43..64da340870e 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/planner/plan/node/PlanNodeType.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/planner/plan/node/PlanNodeType.java @@ -209,6 +209,7 @@ public enum PlanNodeType { RELATIONAL_INSERT_ROWS((short) 2002), RELATIONAL_DELETE_DATA((short) 2003), OBJECT_FILE_NODE((short) 2004), + RELATIONAL_INSERT_NUMERIC_ROW((short) 2005), ; private static final IPlanNodeDeserializer DESERIALIZER;
