You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/10/12 03:38:46 UTC
(iotdb) 03/20: tmp
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 <jt...@163.com>
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;