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 d29c9c420a32bd20e6e1480e7313c8d040c1eb07 Author: Tian Jiang <[email protected]> AuthorDate: Mon Oct 12 11:37:31 2026 +0800 Finalize prepared numeric row API and performance coverage Remove the non-prepared numeric row API and its obsolete tests. Extend the opt-in single-row benchmark with partial-column workloads, latency, write-stage timing and process resource sampling. --- .../it/env/cluster/config/MppDataNodeConfig.java | 6 + .../it/env/remote/config/RemoteDataNodeConfig.java | 5 + .../apache/iotdb/itbase/env/DataNodeConfig.java | 2 + .../manual/basic/IoTDBPipeNumericRowIT.java | 10 +- .../IoTDBNumericRowInsertPerformanceIT.java | 538 -------------------- .../it/session/IoTDBTableModelSessionIT.java | 540 ++++++++++++++++----- .../session/WritePerformanceResourceTracker.java | 234 +++++++++ .../org/apache/iotdb/isession/ITableSession.java | 29 +- .../java/org/apache/iotdb/isession/NumericRow.java | 122 ----- .../isession/NumericRowPreparedStatement.java | 15 +- .../java/org/apache/iotdb/session/Session.java | 85 ---- .../org/apache/iotdb/session/TableSession.java | 6 - .../iotdb/session/pool/TableSessionWrapper.java | 14 - .../apache/iotdb/session/util/SessionUtils.java | 103 ---- .../session/PreparedNumericRowBufferTest.java | 56 +++ .../java/org/apache/iotdb/session/SessionTest.java | 104 ---- .../session/pool/PreparedNumericRowPoolTest.java | 23 - .../iotdb/session/util/SessionUtilsTest.java | 202 +------- .../plan/parser/StatementGenerator.java | 37 +- .../statement/crud/InsertNumericRowStatement.java | 65 --- .../session/PreparedNumericRowInfoTest.java | 50 +- .../plan/parser/NumericTableRowTest.java | 128 ++--- .../thrift-datanode/src/main/thrift/client.thrift | 8 +- 23 files changed, 819 insertions(+), 1563 deletions(-) diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java index 8c5f56ad9f1..cfa55537dcb 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java @@ -56,6 +56,12 @@ public class MppDataNodeConfig extends MppBaseConfig implements DataNodeConfig { return this; } + @Override + public DataNodeConfig setMetricPrometheusReporterAsyncUpdate(boolean asyncUpdate) { + properties.setProperty("prometheus_reporter_async_update", String.valueOf(asyncUpdate)); + return this; + } + @Override public DataNodeConfig setMetricPrometheusReporterUsername(String username) { properties.setProperty("metric_prometheus_reporter_username", username); diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java index c97e2b5065a..f754b71ebe1 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java @@ -28,6 +28,11 @@ public class RemoteDataNodeConfig implements DataNodeConfig { return this; } + @Override + public DataNodeConfig setMetricPrometheusReporterAsyncUpdate(boolean asyncUpdate) { + return this; + } + @Override public DataNodeConfig setMetricPrometheusReporterUsername(String username) { return this; diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java index fb969c778ea..c820e1325e9 100644 --- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java @@ -25,6 +25,8 @@ import java.util.List; public interface DataNodeConfig { DataNodeConfig setMetricReporterType(List<String> metricReporterTypes); + DataNodeConfig setMetricPrometheusReporterAsyncUpdate(boolean asyncUpdate); + DataNodeConfig setMetricPrometheusReporterUsername(String username); DataNodeConfig setMetricPrometheusReporterPassword(String password); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeNumericRowIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeNumericRowIT.java index 5211f290771..cc2e614b43a 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeNumericRowIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeNumericRowIT.java @@ -31,7 +31,6 @@ import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq; import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.isession.ITableSession; -import org.apache.iotdb.isession.NumericRow; import org.apache.iotdb.isession.NumericRowPreparedStatement; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.it.env.MultiEnvFactory; @@ -127,7 +126,7 @@ public class IoTDBPipeNumericRowIT extends AbstractPipeTableModelDualManualIT { @Test public void testRealtimeWithoutBatch() throws Exception { - // Unflushed NumericRow and prepared rows must arrive through individual InsertNode requests. + // Unflushed prepared rows must arrive through individual InsertNode requests. testRealtime(false); } @@ -381,12 +380,7 @@ public class IoTDBPipeNumericRowIT extends AbstractPipeTableModelDualManualIT { if (row % 3 == 0) { nulls.mark(row % 7); } - if (row % 2 == 0) { - session.executePreparedInsertNumericRow(prepared, row, values, tags, nulls); - } else { - session.insert( - new NumericRow("readings", row, NAMES, TYPES, CATEGORIES, values, tags, nulls)); - } + session.executePreparedInsertNumericRow(prepared, row, values, tags, nulls); } } } 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 deleted file mode 100644 index e2d6a481743..00000000000 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/performance/IoTDBNumericRowInsertPerformanceIT.java +++ /dev/null @@ -1,538 +0,0 @@ -/* - * 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; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.Future; - -/** - * Manual end-to-end benchmark for the numeric table-row API and relational {@link Tablet} writes. - * - * <p>Enable this test with {@code -Diotdb.numeric.row.insert.perf.enabled=true}. The timed region - * includes the concurrent Session calls and server write confirmations. Schema creation and - * client-side input preparation are outside the timed region. By default 16 client sessions write - * one row per request; set {@code iotdb.numeric.row.insert.perf.client.count} or {@code - * iotdb.numeric.row.insert.perf.tablet.row.count} to vary concurrency or Tablet batching. - * - * <p>Results are written to {@code target/numeric-row-insert-performance.properties}. Percentiles - * describe elapsed time per complete round, not individual RPC latency. See the adjacent README for - * 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 CLIENT_COUNT_PROPERTY = PROPERTY_PREFIX + "client.count"; - private static final String TABLET_ROW_COUNT_PROPERTY = PROPERTY_PREFIX + "tablet.row.count"; - private static final String WARMUP_ROUND_COUNT_PROPERTY = PROPERTY_PREFIX + "warmup.round.count"; - private static final String ROUND_COUNT_PROPERTY = PROPERTY_PREFIX + "round.count"; - private static final String RESULT_PATH_PROPERTY = PROPERTY_PREFIX + "result.path"; - - private static final int ROW_COUNT = Integer.getInteger(ROW_COUNT_PROPERTY, 10_000); - private static final int COLUMN_COUNT = Integer.getInteger(COLUMN_COUNT_PROPERTY, 100); - private static final int CLIENT_COUNT = Integer.getInteger(CLIENT_COUNT_PROPERTY, 16); - private static final int TABLET_ROW_COUNT = Integer.getInteger(TABLET_ROW_COUNT_PROPERTY, 1); - private static final int WARMUP_ROUND_COUNT = Integer.getInteger(WARMUP_ROUND_COUNT_PROPERTY, 1); - private static final int ROUND_COUNT = Integer.getInteger(ROUND_COUNT_PROPERTY, 5); - - 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]; - - final List<ITableSession> sessions = new ArrayList<>(CLIENT_COUNT); - final ExecutorService executor = Executors.newFixedThreadPool(CLIENT_COUNT); - try { - final ITableSession setupSession = EnvFactory.getEnv().getTableSessionConnection(); - setupSession.executeNonQueryStatement("CREATE DATABASE " + DATABASE); - setupSession.executeNonQueryStatement("USE " + DATABASE); - sessions.add(setupSession); - for (int client = 1; client < CLIENT_COUNT; client++) { - final ITableSession session = EnvFactory.getEnv().getTableSessionConnection(); - session.executeNonQueryStatement("USE " + DATABASE); - sessions.add(session); - } - - final int totalRoundCount = WARMUP_ROUND_COUNT + ROUND_COUNT; - for (int round = 0; round < totalRoundCount; round++) { - final List<NumericRowWorkload> numericRowWorkloads = - prepareNumericRowWorkloads(setupSession, round); - final List<TabletWorkload> tabletWorkloads = prepareTabletWorkloads(setupSession, round); - - final long numericRowNanos; - final long tabletNanos; - // Alternate execution order so one path does not consistently receive a warmer server. - if ((round & 1) == 0) { - numericRowNanos = executeNumericRowWorkload(sessions, numericRowWorkloads, executor); - tabletNanos = executeTabletWorkload(sessions, tabletWorkloads, executor); - } else { - tabletNanos = executeTabletWorkload(sessions, tabletWorkloads, executor); - numericRowNanos = executeNumericRowWorkload(sessions, numericRowWorkloads, executor); - } - for (NumericRowWorkload workload : numericRowWorkloads) { - assertRowCount(setupSession, workload.tableName); - } - for (TabletWorkload workload : tabletWorkloads) { - assertRowCount(setupSession, workload.tableName); - } - - if (round >= WARMUP_ROUND_COUNT) { - final int measuredRound = round - WARMUP_ROUND_COUNT; - numericRowElapsedNanos[measuredRound] = numericRowNanos; - tabletElapsedNanos[measuredRound] = tabletNanos; - } - } - } finally { - executor.shutdownNow(); - for (ITableSession session : sessions) { - session.close(); - } - } - - final BenchmarkResult result = new BenchmarkResult(numericRowElapsedNanos, tabletElapsedNanos); - writeResult(result); - } - - private static List<NumericRowWorkload> prepareNumericRowWorkloads( - ITableSession session, int round) throws Exception { - final List<NumericRowWorkload> workloads = new ArrayList<>(CLIENT_COUNT); - for (int client = 0; client < CLIENT_COUNT; client++) { - final String tableName = "numeric_row_" + round + "_" + client; - createTable(session, tableName); - final long startTimestamp = ((long) round * CLIENT_COUNT + client) * ROW_COUNT; - final NumericRow[] rows = new NumericRow[ROW_COUNT]; - final String[] tagValues = new String[0]; - for (int row = 0; row < ROW_COUNT; row++) { - rows[row] = - new NumericRow( - tableName, - startTimestamp + row, - columnNames, - dataTypes, - columnCategories, - createRawValues(startTimestamp + row), - tagValues); - } - workloads.add(new NumericRowWorkload(tableName, rows)); - } - return workloads; - } - - private static List<TabletWorkload> prepareTabletWorkloads(ITableSession session, int round) - throws Exception { - final List<TabletWorkload> workloads = new ArrayList<>(CLIENT_COUNT); - for (int client = 0; client < CLIENT_COUNT; client++) { - final String tableName = "tablet_" + round + "_" + client; - createTable(session, tableName); - final long startTimestamp = ((long) round * CLIENT_COUNT + client) * ROW_COUNT; - final List<Tablet> tablets = new ArrayList<>((ROW_COUNT - 1) / TABLET_ROW_COUNT + 1); - for (int rowStart = 0; rowStart < ROW_COUNT; ) { - final int rowCount = Math.min(TABLET_ROW_COUNT, ROW_COUNT - rowStart); - final Tablet tablet = - new Tablet(tableName, columnNames, dataTypes, columnCategories, rowCount); - for (int row = 0; row < rowCount; row++) { - final int rowIndex = tablet.getRowSize(); - final long timestamp = startTimestamp + rowStart + row; - final long[] values = createRawValues(timestamp); - tablet.addTimestamp(rowIndex, timestamp); - for (int column = 0; column < COLUMN_COUNT; column++) { - tablet.addValue( - columnNames.get(column), - rowIndex, - toTabletValue(dataTypes.get(column), values[column])); - } - } - tablets.add(tablet); - rowStart += rowCount; - } - workloads.add(new TabletWorkload(tableName, tablets)); - } - return workloads; - } - - private static long executeNumericRowWorkload( - List<ITableSession> sessions, List<NumericRowWorkload> workloads, ExecutorService executor) - throws Exception { - return executeConcurrently( - sessions, - workloads, - executor, - (session, workload) -> { - for (NumericRow row : workload.rows) { - session.insert(row); - } - }); - } - - private static long executeTabletWorkload( - List<ITableSession> sessions, List<TabletWorkload> workloads, ExecutorService executor) - throws Exception { - return executeConcurrently( - sessions, - workloads, - executor, - (session, workload) -> { - for (Tablet tablet : workload.tablets) { - session.insert(tablet); - } - }); - } - - private static <T> long executeConcurrently( - List<ITableSession> sessions, - List<T> workloads, - ExecutorService executor, - WorkloadWriter<T> writer) - throws Exception { - final CountDownLatch ready = new CountDownLatch(CLIENT_COUNT); - final CountDownLatch start = new CountDownLatch(1); - final List<Future<?>> futures = new ArrayList<>(CLIENT_COUNT); - for (int client = 0; client < CLIENT_COUNT; client++) { - final ITableSession session = sessions.get(client); - final T workload = workloads.get(client); - futures.add( - executor.submit( - () -> { - ready.countDown(); - start.await(); - writer.write(session, workload); - return null; - })); - } - ready.await(); - final long startNanos = System.nanoTime(); - start.countDown(); - for (Future<?> future : futures) { - future.get(); - } - return System.nanoTime() - startNanos; - } - - @FunctionalInterface - private interface WorkloadWriter<T> { - - void write(ITableSession session, T workload) throws Exception; - } - - private static void createTable(ITableSession session, String tableName) throws Exception { - final StringBuilder statement = new StringBuilder("CREATE TABLE "); - statement.append(tableName).append('('); - 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(CLIENT_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", Integer.toString(CLIENT_COUNT)); - properties.setProperty("row.count", Integer.toString(ROW_COUNT)); - properties.setProperty("column.count", Integer.toString(COLUMN_COUNT)); - properties.setProperty("tablet.row.count", Integer.toString(TABLET_ROW_COUNT)); - properties.setProperty("warmup.round.count", Integer.toString(WARMUP_ROUND_COUNT)); - properties.setProperty("round.count", Integer.toString(ROUND_COUNT)); - properties.setProperty("column.types", join(dataTypes)); - properties.setProperty( - "numeric.row.requests.per.round", Integer.toString(CLIENT_COUNT * ROW_COUNT)); - properties.setProperty( - "insert.tablet.requests.per.round", - Integer.toString(CLIENT_COUNT * ((ROW_COUNT - 1) / TABLET_ROW_COUNT + 1))); - properties.setProperty("numeric.row.round.elapsed.nanos", join(result.numericRowElapsedNanos)); - properties.setProperty("insert.tablet.round.elapsed.nanos", join(result.tabletElapsedNanos)); - properties.setProperty( - "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 CLIENT_COUNT * 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 CLIENT_COUNT * 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 e196ff1edf0..6f6ee44f129 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,7 +20,6 @@ package org.apache.iotdb.relational.it.session; import org.apache.iotdb.isession.ITableSession; -import org.apache.iotdb.isession.NumericRow; import org.apache.iotdb.isession.NumericRowPreparedStatement; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.isession.pool.ITableSessionPool; @@ -28,8 +27,10 @@ import org.apache.iotdb.it.env.EnvFactory; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.TableClusterIT; import org.apache.iotdb.itbase.category.TableLocalStandaloneIT; +import org.apache.iotdb.jdbc.Config; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.rpc.UrlUtils; import org.apache.tsfile.enums.ColumnCategory; import org.apache.tsfile.enums.TSDataType; @@ -48,8 +49,10 @@ import java.time.LocalDate; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.SplittableRandom; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -57,6 +60,8 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.function.IntConsumer; +import java.util.regex.Matcher; +import java.util.regex.Pattern; import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.showTablesColumnHeaders; import static org.junit.Assert.assertEquals; @@ -72,15 +77,26 @@ public class IoTDBTableModelSessionIT { @Before public void setUp() throws Exception { - if (Boolean.getBoolean("iotdb.numeric.row.insert.perf.enabled") - && Boolean.getBoolean("iotdb.numeric.row.insert.perf.io.enabled")) { - // Isolate per-round storage accounting from compaction of earlier rounds. - EnvFactory.getEnv() - .getConfig() - .getCommonConfig() - .setEnableSeqSpaceCompaction(false) - .setEnableUnseqSpaceCompaction(false) - .setEnableCrossSpaceCompaction(false); + if (Boolean.getBoolean("iotdb.numeric.row.insert.perf.enabled")) { + // A saturated single DataNode can delay ConfigNode heartbeats while the benchmark is running. + // Keep the test focused on write-path throughput instead of fencing the node for that delay. + EnvFactory.getEnv().getConfig().getConfigNodeConfig().setMetadataLeaseFenceMs(120_000); + if (Boolean.getBoolean("iotdb.numeric.row.insert.perf.breakdown.enabled")) { + EnvFactory.getEnv() + .getConfig() + .getDataNodeConfig() + .setMetricReporterType(Collections.singletonList("PROMETHEUS")) + .setMetricPrometheusReporterAsyncUpdate(false); + } + if (Boolean.getBoolean("iotdb.numeric.row.insert.perf.io.enabled")) { + // Isolate per-round storage accounting from compaction of earlier rounds. + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setEnableSeqSpaceCompaction(false) + .setEnableUnseqSpaceCompaction(false) + .setEnableCrossSpaceCompaction(false); + } } EnvFactory.getEnv().initClusterEnvironment(); } @@ -91,7 +107,7 @@ public class IoTDBTableModelSessionIT { } @Test - public void testNumericRowInsert() throws Exception { + public void testPreparedNumericRowInsert() throws Exception { // 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()) { @@ -120,26 +136,20 @@ public class IoTDBTableModelSessionIT { 20260910, 1234 }; - session.insert( - new NumericRow( - "readings", - 1, - names, - types, - Collections.nCopies(types.size(), ColumnCategory.FIELD), - values, - new String[0])); + try (NumericRowPreparedStatement rowStatement = + session.prepareInsertNumericRow( + "readings", names, types, Collections.nCopies(types.size(), ColumnCategory.FIELD))) { + session.executePreparedInsertNumericRow(rowStatement, 1, values, new String[0]); + } try (ITableSessionPool pool = EnvFactory.getEnv().getTableSessionPool(1, "numeric_rows"); - ITableSession pooledSession = pool.getSession()) { - pooledSession.insert( - new NumericRow( - "readings", - 2, - names, - types, - Collections.nCopies(types.size(), ColumnCategory.FIELD), - values, - new String[0])); + ITableSession pooledSession = pool.getSession(); + NumericRowPreparedStatement pooledRow = + pooledSession.prepareInsertNumericRow( + "readings", + names, + types, + Collections.nCopies(types.size(), ColumnCategory.FIELD))) { + pooledSession.executePreparedInsertNumericRow(pooledRow, 2, values, new String[0]); } assertNumericRows(session); session.executeNonQueryStatement("FLUSH"); @@ -148,7 +158,7 @@ public class IoTDBTableModelSessionIT { } @Test - public void testNumericRowInsertWithTags() throws Exception { + public void testPreparedNumericRowInsertWithTags() throws Exception { // Different TAGs at the same timestamp must stay separate, with identical behavior after flush. try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) { session.executeNonQueryStatement("CREATE DATABASE numeric_tag_rows"); @@ -162,45 +172,29 @@ public class IoTDBTableModelSessionIT { List<ColumnCategory> categories = Arrays.asList( ColumnCategory.FIELD, ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.TAG); - session.insert( - new NumericRow( - "readings", - 1, - names, - types, - categories, - new long[] {1, 10}, - new String[] {"设备甲😀", "华北"})); + try (NumericRowPreparedStatement rowStatement = + session.prepareInsertNumericRow("readings", names, types, categories)) { + session.executePreparedInsertNumericRow( + rowStatement, 1, new long[] {1, 10}, new String[] {"设备甲😀", "华北"}); + } try (ITableSessionPool pool = EnvFactory.getEnv().getTableSessionPool(1, "numeric_tag_rows"); ITableSession pooled = pool.getSession()) { - pooled.insert( - new NumericRow( - "readings", - 1, - names, - types, - categories, - new long[] {0, 20}, - new String[] {"设备乙", "华北"})); - } - session.insert( - new NumericRow( - "readings", - 2, - names, - types, - categories, - new long[] {1, 30}, - new String[] {"", null})); - session.insert( - new NumericRow( - "readings", - 2, - names, - types, - categories, - new long[] {0, 40}, - new String[] {null, null})); + try (NumericRowPreparedStatement rowStatement = + pooled.prepareInsertNumericRow("readings", names, types, categories)) { + pooled.executePreparedInsertNumericRow( + rowStatement, 1, new long[] {0, 20}, new String[] {"设备乙", "华北"}); + } + } + try (NumericRowPreparedStatement rowStatement = + session.prepareInsertNumericRow("readings", names, types, categories)) { + session.executePreparedInsertNumericRow( + rowStatement, 2, new long[] {1, 30}, new String[] {"", null}); + } + try (NumericRowPreparedStatement rowStatement = + session.prepareInsertNumericRow("readings", names, types, categories)) { + session.executePreparedInsertNumericRow( + rowStatement, 2, new long[] {0, 40}, new String[] {null, null}); + } assertNumericTagRows(session); session.executeNonQueryStatement("FLUSH"); assertNumericTagRows(session); @@ -278,17 +272,16 @@ public class IoTDBTableModelSessionIT { nulls.mark(7); nulls.mark(8); String[] tags = {"device"}; - session.insert(new NumericRow("readings", 1, names, types, categories, values, tags, nulls)); try (NumericRowPreparedStatement prepared = session.prepareInsertNumericRow("readings", names, types, categories)) { + session.executePreparedInsertNumericRow(prepared, 1, values, tags, nulls); session.executePreparedInsertNumericRow(prepared, 2, values, tags, nulls); session.executePreparedInsertNumericRow(prepared, 3, values, tags); // All-null table rows retain their timestamp/TAGs; null slots must never become zeros. BitMap allNulls = new BitMap(9); allNulls.markAll(); session.executePreparedInsertNumericRow(prepared, 3, values, tags, allNulls); - session.insert( - new NumericRow("readings", 10, names, types, categories, values, tags, allNulls)); + session.executePreparedInsertNumericRow(prepared, 10, values, tags, allNulls); session.executePreparedInsertNumericRow(prepared, 11, values, tags, allNulls); session.executeNonQueryStatement("ALTER TABLE readings ADD COLUMN tag2 STRING TAG"); session.executePreparedInsertNumericRow(prepared, 4, values, tags, nulls); @@ -297,7 +290,7 @@ public class IoTDBTableModelSessionIT { ITableSession pooled = pool.getSession(); NumericRowPreparedStatement prepared = pooled.prepareInsertNumericRow("readings", names, types, categories)) { - pooled.insert(new NumericRow("readings", 5, names, types, categories, values, tags, nulls)); + pooled.executePreparedInsertNumericRow(prepared, 5, values, tags, nulls); pooled.executePreparedInsertNumericRow(prepared, 6, values, tags, nulls); } assertNumericNullRows(session); @@ -441,17 +434,64 @@ public class IoTDBTableModelSessionIT { @Test public void testPreparedNumericRowAndTabletWritePerformance() throws Exception { + compareNumericRowWritePerformance(false); + } + + @Test + public void testPreparedNumericRowAndTabletSparseWritePerformance() throws Exception { + compareNumericRowWritePerformance(true); + } + + @Test + public void testPreparedNumericRowAndTabletWriteRatioPerformance() throws Exception { + Assume.assumeTrue( + "Performance benchmark disabled; set iotdb.numeric.row.insert.perf.enabled=true to run", + Boolean.getBoolean("iotdb.numeric.row.insert.perf.enabled")); + final int fieldCount = Integer.getInteger("iotdb.numeric.row.insert.perf.fields", 99); + boolean first = true; + for (int ratio : configuredSparseWriteRatios()) { + if (!first) { + // Recreate the instance so earlier ratios cannot contribute flush or compaction work. + EnvFactory.getEnv().cleanClusterEnvironment(); + EnvFactory.getEnv() + .setTestMethodName("testPreparedNumericRowAndTabletWriteRatioPerformance_" + ratio); + setUp(); + } + first = false; + int writtenFields = Math.max(1, (fieldCount * ratio + 99) / 100); + compareNumericRowWritePerformance(true, writtenFields, "sparse-" + ratio + "%"); + } + } + + private void compareNumericRowWritePerformance(boolean sparse) throws Exception { + int writtenFields = + sparse ? Integer.getInteger("iotdb.numeric.row.insert.perf.sparse.fields", 10) : 99; + compareNumericRowWritePerformance(sparse, writtenFields, sparse ? "sparse" : "full-row"); + } + + private void compareNumericRowWritePerformance( + boolean sparse, int writtenFields, String scenarioName) throws Exception { // Compare synchronous single-row paths and optionally account for their Linux TCP/storage I/O. Assume.assumeTrue( "Performance benchmark disabled; set iotdb.numeric.row.insert.perf.enabled=true to run", Boolean.getBoolean("iotdb.numeric.row.insert.perf.enabled")); final int concurrency = Integer.getInteger("iotdb.numeric.row.insert.perf.concurrency", 16); - // Compare sustained single-row writes after warming all three paths; verify all acknowledged + // Compare sustained single-row writes after warming both paths; verify all acknowledged // rows. final int warmupRounds = 2; final int measuredRounds = 6; - final int fieldCount = 99; - final String databasePrefix = "numeric_perf_"; + final int fieldCount = Integer.getInteger("iotdb.numeric.row.insert.perf.fields", 99); + final int deviceCount = + sparse ? Integer.getInteger("iotdb.numeric.row.insert.perf.sparse.devices", 100) : 1; + final long seed = Long.getLong("iotdb.numeric.row.insert.perf.sparse.seed", 20261008L); + assertTrue("Concurrency and device count must be positive", concurrency > 0 && deviceCount > 0); + assertTrue( + "Written FIELD count must be positive and within the table schema", + writtenFields > 0 && writtenFields <= fieldCount); + final String databasePrefix = + (sparse ? "numeric_sparse_perf_" : "numeric_perf_") + + scenarioName.replace('%', 'p').replace('-', '_') + + "_"; try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) { StringBuilder ddl = new StringBuilder("CREATE TABLE prepared_perf (tag STRING TAG"); for (int i = 0; i < fieldCount; i++) { @@ -464,7 +504,6 @@ public class IoTDBTableModelSessionIT { session.executeNonQueryStatement("USE " + databasePrefix + i); session.executeNonQueryStatement(ddl.toString()); session.executeNonQueryStatement(ddl.toString().replace("prepared_perf", "tablet_perf")); - session.executeNonQueryStatement(ddl.toString().replace("prepared_perf", "row_perf")); } } @@ -482,11 +521,12 @@ public class IoTDBTableModelSessionIT { ITableSession[] sessions = new ITableSession[concurrency]; NumericRowPreparedStatement[] prepared = new NumericRowPreparedStatement[concurrency]; Tablet[] tablets = new Tablet[concurrency]; - NumericRow[] rows = new NumericRow[concurrency]; String[][] tags = new String[concurrency][]; long[][] numericValues = new long[concurrency][fieldCount]; - long[][] pathRows = new long[3][concurrency]; + long[][] pathRows = new long[2][concurrency]; + SparseNumericRowInput[][] sparseRows = new SparseNumericRowInput[2][concurrency]; ExecutorService executor = Executors.newFixedThreadPool(concurrency); + WritePerformanceResourceTracker resourceTracker = null; try { // Connections, USE, prepare and reusable input allocation are outside every timed round. for (int i = 0; i < concurrency; i++) { @@ -497,17 +537,32 @@ public class IoTDBTableModelSessionIT { tablets[i] = new Tablet("tablet_perf", names, types, categories, 1); tags[i] = new String[] {"device-" + i}; Arrays.fill(numericValues[i], Double.doubleToRawLongBits(1.0)); - rows[i] = - new NumericRow("row_perf", 0, names, types, categories, numericValues[i], tags[i]); + if (sparse) { + // Independent generators replay the same device/column sequence for each path. + for (int path = 0; path < sparseRows.length; path++) { + sparseRows[path][i] = + new SparseNumericRowInput(deviceCount, fieldCount, writtenFields, seed + i); + } + } } IntConsumer preparedWrite = threadId -> { try { - sessions[threadId].executePreparedInsertNumericRow( - prepared[threadId], - pathRows[0][threadId], - numericValues[threadId], - tags[threadId]); + if (sparse) { + SparseNumericRowInput input = sparseRows[0][threadId]; + sessions[threadId].executePreparedInsertNumericRow( + prepared[threadId], + pathRows[0][threadId], + numericValues[threadId], + input.tagValues, + input.nulls); + } else { + sessions[threadId].executePreparedInsertNumericRow( + prepared[threadId], + pathRows[0][threadId], + numericValues[threadId], + tags[threadId]); + } } catch (Exception e) { throw new RuntimeException(e); } @@ -515,32 +570,22 @@ public class IoTDBTableModelSessionIT { IntConsumer tabletPrepare = threadId -> { Tablet tablet = tablets[threadId]; - tablet.addTimestamp(0, pathRows[2][threadId]); - tablet.addValue("tag", 0, tags[threadId][0]); + tablet.addTimestamp(0, pathRows[1][threadId]); + SparseNumericRowInput input = sparseRows[1][threadId]; + if (sparse) { + input.next(); + } + tablet.addValue("tag", 0, sparse ? input.tagValues[0] : tags[threadId][0]); for (int i = 0; i < fieldCount; i++) { - tablet.addValue(names.get(i + 1), 0, 1.0D); + if (sparse && input.nulls.isMarked(i)) { + tablet.addValue(names.get(i + 1), 0, null); + } else { + tablet.addValue(names.get(i + 1), 0, 1.0D); + } } }; - IntConsumer rowPrepare = - threadId -> - rows[threadId] = - new NumericRow( - "row_perf", - pathRows[1][threadId], - names, - types, - categories, - numericValues[threadId], - tags[threadId]); IntConsumer[] paths = { preparedWrite, - threadId -> { - try { - sessions[threadId].insert(rows[threadId]); - } catch (Exception e) { - throw new RuntimeException(e); - } - }, threadId -> { try { sessions[threadId].insert(tablets[threadId]); @@ -549,17 +594,31 @@ public class IoTDBTableModelSessionIT { } } }; - IntConsumer[] prepareWrites = {null, rowPrepare, tabletPrepare}; + IntConsumer[] prepareWrites = { + sparse ? threadId -> sparseRows[0][threadId].next() : null, tabletPrepare + }; IntConsumer[] completeWrites = { - threadId -> pathRows[0][threadId]++, - threadId -> pathRows[1][threadId]++, threadId -> { + if (sparse) sparseRows[0][threadId].acknowledge(); + pathRows[0][threadId]++; + }, + threadId -> { + if (sparse) sparseRows[1][threadId].acknowledge(); tablets[threadId].reset(); - pathRows[2][threadId]++; + pathRows[1][threadId]++; } }; - String[] pathNames = {"prepared numeric row", "insert NumericRow", "insertTablet batch=1"}; - System.out.printf("configuration: concurrency=%d, tabletBatch=1%n", concurrency); + String[] pathNames = {"prepared numeric row", "insertTablet batch=1"}; + System.out.printf( + "configuration: scenario=%s, concurrency=%d, tabletBatch=1, devicesPerWriter=%d, " + + "totalFields=%d, writtenFields=%d, actualWriteRatio=%.2f%%, seed=%d%n", + scenarioName, + concurrency, + deviceCount, + fieldCount, + writtenFields, + writtenFields * 100.0 / fieldCount, + seed); WritePerformanceIoTracker ioTracker = null; if (Boolean.getBoolean("iotdb.numeric.row.insert.perf.io.enabled")) { assertEquals( @@ -576,29 +635,46 @@ public class IoTDBTableModelSessionIT { EnvFactory.getEnv().getConfigNodeWrapperList().get(0).getPid(), EnvFactory.getEnv().getDataNodeWrapperList().get(0).getPort()); } + resourceTracker = + new WritePerformanceResourceTracker( + EnvFactory.getEnv().getDataNodeWrapperList().get(0).getPid()); double[][] rates = new double[paths.length][measuredRounds]; long[] totalRows = new long[paths.length]; long[] totalNanos = new long[paths.length]; + double[] peakCpuPercent = new double[paths.length]; + Arrays.fill(peakCpuPercent, Double.NaN); + long[] peakResidentBytes = new long[paths.length]; LongSamples[] requestLatencies = new LongSamples[paths.length]; for (int path = 0; path < paths.length; path++) { requestLatencies[path] = new LongSamples(); } + Map<String, MetricSample> previousMetrics = + Boolean.getBoolean("iotdb.numeric.row.insert.perf.breakdown.enabled") + ? readPerformanceMetrics() + : Collections.emptyMap(); for (int round = 0; round < warmupRounds + measuredRounds; round++) { boolean warmup = round < warmupRounds; int seconds = warmup ? 3 : 5; - // Rotate all three paths so each occupies every position during measured rounds. + // Rotate both paths so each occupies every position during measured rounds. for (int position = 0; position < paths.length; position++) { int path = (round + position) % paths.length; WritePerformanceIoTracker.Snapshot ioBefore = ioTracker == null ? null : ioTracker.beforeRound(sessions[0]); - WritePerformance result = - runConcurrentWrites( - executor, - concurrency, - seconds, - prepareWrites[path], - paths[path], - completeWrites[path]); + resourceTracker.beginRound(); + WritePerformance result = null; + WritePerformanceResourceTracker.RoundPeak resourcePeak; + try { + result = + runConcurrentWrites( + executor, + concurrency, + seconds, + prepareWrites[path], + paths[path], + completeWrites[path]); + } finally { + resourcePeak = resourceTracker.endRound(); + } if (ioTracker != null) { System.out.printf( "io %s round %d position %d %s: rows=%d, %s%n", @@ -619,11 +695,37 @@ public class IoTDBTableModelSessionIT { result.nanos() / 1_000_000_000.0, result.rowsPerSecond(), summarizeRequestLatencies(result.requestNanos())); + System.out.printf( + "resource %s round %d position %d %s: %s%n", + warmup ? "warmup" : "measured", + warmup ? round + 1 : round - warmupRounds + 1, + position + 1, + pathNames[path], + resourcePeak.format()); + if (Boolean.getBoolean("iotdb.numeric.row.insert.perf.breakdown.enabled")) { + Map<String, MetricSample> currentMetrics = readPerformanceMetrics(); + printPerformanceBreakdown( + warmup ? "warmup" : "measured", + warmup ? round + 1 : round - warmupRounds + 1, + position + 1, + pathNames[path], + previousMetrics, + currentMetrics); + previousMetrics = currentMetrics; + } if (!warmup) { rates[path][round - warmupRounds] = result.rowsPerSecond(); totalRows[path] += result.rows(); totalNanos[path] += result.nanos(); requestLatencies[path].appendAll(result.requestNanos()); + if (!Double.isNaN(resourcePeak.peakCpuPercent())) { + peakCpuPercent[path] = + Double.isNaN(peakCpuPercent[path]) + ? resourcePeak.peakCpuPercent() + : Math.max(peakCpuPercent[path], resourcePeak.peakCpuPercent()); + } + peakResidentBytes[path] = + Math.max(peakResidentBytes[path], resourcePeak.peakResidentBytes()); } } } @@ -631,20 +733,26 @@ public class IoTDBTableModelSessionIT { Arrays.sort(rates[path]); System.out.printf( "summary %s: median %.1f rows/s, aggregate %.1f rows/s, min %.1f, max %.1f, " - + "measured API latency %s%n", + + "measured API latency %s, peak DataNode CPU %s, peak DataNode RSS %s, " + + "scenario=%s, writtenFields=%d/%d, nonNullFieldsPerSecond=%.1f%n", pathNames[path], (rates[path][measuredRounds / 2 - 1] + rates[path][measuredRounds / 2]) / 2, totalRows[path] * 1_000_000_000.0 / totalNanos[path], rates[path][0], rates[path][measuredRounds - 1], - summarizeRequestLatencies(requestLatencies[path].toArray())); + summarizeRequestLatencies(requestLatencies[path].toArray()), + formatPeakCpu(peakCpuPercent[path]), + formatPeakMemory(peakResidentBytes[path]), + scenarioName, + writtenFields, + fieldCount, + totalRows[path] * 1_000_000_000.0 / totalNanos[path] * writtenFields); } // Per-device monotonic timestamps span warmup and measurement, so no rows overwrite others. for (int path = 0; path < paths.length; path++) { String table = switch (path) { case 0 -> "prepared_perf"; - case 1 -> "row_perf"; default -> "tablet_perf"; }; long[] rowsByThread = pathRows[path]; @@ -658,15 +766,30 @@ public class IoTDBTableModelSessionIT { assertFalse(dataSet.hasNext()); } } - System.out.printf("verified %s: %d rows%n", table, expectedRows); + if (sparse) { + // Check counts for every device and FIELD; stale Tablet bits must not fill omitted cells. + for (int i = 0; i < concurrency; i++) { + sparseRows[path][i].verify(sessions[i], table); + } + } + System.out.printf( + "verified %s: %d rows, %d non-null FIELD values%n", + table, expectedRows, expectedRows * writtenFields); } } finally { executor.shutdownNow(); try { assertTrue(executor.awaitTermination(1, TimeUnit.MINUTES)); } finally { - // Closing each session also releases its server-side prepared metadata. Exception closeFailure = null; + if (resourceTracker != null) { + try { + resourceTracker.close(); + } catch (Exception e) { + closeFailure = e; + } + } + // Closing each session also releases its server-side prepared metadata. for (ITableSession session : sessions) { if (session != null) { try { @@ -682,6 +805,171 @@ public class IoTDBTableModelSessionIT { } } + private static final Pattern PERFORMANCE_TIMER_SUM_PATTERN = + Pattern.compile( + "^(performance_overview[^\\s{]+)_sum\\{[^}]*stage=\\\"([^\\\"]+)\\\"[^}]*\\}\\s+([-+0-9.eE]+)$"); + private static final Pattern PERFORMANCE_TIMER_COUNT_PATTERN = + Pattern.compile( + "^(performance_overview[^\\s{]+)_count\\{[^}]*stage=\\\"([^\\\"]+)\\\"[^}]*\\}\\s+([-+0-9.eE]+)$"); + + private static Map<String, MetricSample> readPerformanceMetrics() { + Map<String, double[]> values = new LinkedHashMap<>(); + for (var dataNode : EnvFactory.getEnv().getDataNodeWrapperList()) { + String content = + EnvFactory.getEnv() + .getUrlContent( + Config.IOTDB_HTTP_URL_PREFIX + + UrlUtils.formatTEndPointIpv4AndIpv6Url( + dataNode.getIp(), dataNode.getMetricPort()) + + "/metrics", + null); + if (content == null) { + continue; + } + for (String line : content.split("\\n")) { + Matcher sumMatcher = PERFORMANCE_TIMER_SUM_PATTERN.matcher(line.trim()); + if (sumMatcher.matches()) { + values + .computeIfAbsent( + sumMatcher.group(1) + "." + sumMatcher.group(2), key -> new double[2])[0] += + Double.parseDouble(sumMatcher.group(3)); + continue; + } + Matcher countMatcher = PERFORMANCE_TIMER_COUNT_PATTERN.matcher(line.trim()); + if (countMatcher.matches()) { + values + .computeIfAbsent( + countMatcher.group(1) + "." + countMatcher.group(2), key -> new double[2])[ + 1] += + Double.parseDouble(countMatcher.group(3)); + } + } + } + Map<String, MetricSample> result = new LinkedHashMap<>(); + values.forEach((key, value) -> result.put(key, new MetricSample(value[0], value[1]))); + return result; + } + + private static void printPerformanceBreakdown( + String phase, + int round, + int position, + String path, + Map<String, MetricSample> previous, + Map<String, MetricSample> current) { + StringBuilder output = + new StringBuilder( + String.format("breakdown %s round %d position %d %s:", phase, round, position, path)); + for (Map.Entry<String, MetricSample> entry : current.entrySet()) { + MetricSample before = previous.getOrDefault(entry.getKey(), MetricSample.ZERO); + double count = entry.getValue().count() - before.count(); + double seconds = entry.getValue().seconds() - before.seconds(); + if (count > 0 && seconds >= 0) { + output.append( + String.format( + " %s count=%.0f total=%.3f ms avg=%.3f us;", + entry.getKey(), count, seconds * 1000, seconds * 1_000_000 / count)); + } + } + System.out.println(output); + } + + private record MetricSample(double seconds, double count) { + private static final MetricSample ZERO = new MetricSample(0, 0); + } + + private static int[] configuredSparseWriteRatios() { + String configured = + System.getProperty("iotdb.numeric.row.insert.perf.sparse.ratios", "10,50,100"); + String[] tokens = configured.split(",", -1); + int[] ratios = new int[tokens.length]; + for (int i = 0; i < tokens.length; i++) { + ratios[i] = Integer.parseInt(tokens[i].trim()); + assertTrue( + "Sparse write ratios must be between 1 and 100", ratios[i] > 0 && ratios[i] <= 100); + } + return ratios; + } + + /** Reproducible random device and distinct FIELD selection, with no per-request allocations. */ + private static final class SparseNumericRowInput { + private final SplittableRandom random; + private final String[] devices; + private final String[] tagValues = new String[1]; + private final int[] columns; + private final BitMap nulls; + private final int writtenFields; + private final long[][] expected; + private int device; + + private SparseNumericRowInput(int deviceCount, int fieldCount, int writtenFields, long seed) { + random = new SplittableRandom(seed); + devices = new String[deviceCount]; + for (int i = 0; i < deviceCount; i++) devices[i] = "device-" + i; + columns = new int[fieldCount]; + for (int i = 0; i < fieldCount; i++) columns[i] = i; + nulls = new BitMap(fieldCount); + this.writtenFields = writtenFields; + // Slot zero counts rows, followed by one counter for each FIELD. + expected = new long[deviceCount][fieldCount + 1]; + } + + private void next() { + device = random.nextInt(devices.length); + tagValues[0] = devices[device]; + nulls.markAll(); + // Partial Fisher-Yates sampling gives exactly writtenFields distinct columns per request. + for (int i = 0; i < writtenFields; i++) { + int other = random.nextInt(i, columns.length); + int column = columns[other]; + columns[other] = columns[i]; + columns[i] = column; + nulls.unmark(column); + } + } + + private void acknowledge() { + expected[device][0]++; + for (int i = 0; i < writtenFields; i++) expected[device][columns[i] + 1]++; + } + + private void verify(ITableSession session, String table) throws Exception { + StringBuilder query = new StringBuilder("SELECT tag, count(*)"); + for (int i = 0; i < columns.length; i++) query.append(", count(f").append(i).append(')'); + query.append(" FROM ").append(table).append(" GROUP BY tag"); + boolean[] seen = new boolean[devices.length]; + try (SessionDataSet result = session.executeQueryStatement(query.toString())) { + while (result.hasNext()) { + RowRecord row = result.next(); + int index = + Integer.parseInt( + row.getFields().get(0).getStringValue().substring("device-".length())); + assertTrue("Unexpected device", index >= 0 && index < devices.length); + assertEquals(devices[index], row.getFields().get(0).getStringValue()); + assertFalse("Duplicate device", seen[index]); + seen[index] = true; + for (int i = 0; i <= columns.length; i++) { + assertEquals( + table + "/" + devices[index] + "/count column " + i, + expected[index][i], + row.getFields().get(i + 1).getLongV()); + } + } + } + for (int i = 0; i < devices.length; i++) assertEquals(expected[i][0] > 0, seen[i]); + } + } + + private static String formatPeakCpu(double peakCpuPercent) { + return Double.isNaN(peakCpuPercent) ? "unavailable" : String.format("%.1f%%", peakCpuPercent); + } + + private static String formatPeakMemory(long peakResidentBytes) { + return peakResidentBytes <= 0 + ? "unavailable" + : String.format("%.1f MiB", peakResidentBytes / 1024.0 / 1024.0); + } + private record WritePerformance(long rows, long nanos, long[] requestNanos) { double rowsPerSecond() { return rows * 1_000_000_000.0 / nanos; diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/session/WritePerformanceResourceTracker.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/session/WritePerformanceResourceTracker.java new file mode 100644 index 00000000000..64a680875af --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/session/WritePerformanceResourceTracker.java @@ -0,0 +1,234 @@ +/* + * 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.session; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +/** Samples the DataNode process resource usage while one benchmark round is running. */ +final class WritePerformanceResourceTracker implements AutoCloseable { + private static final Path PROC = Path.of("/proc"); + private static final long SAMPLE_INTERVAL_MILLIS = 100; + + private final long dataNodePid; + private final long clockTicksPerSecond; + private final Thread sampler; + private final AtomicReference<Round> currentRound = new AtomicReference<>(); + private final AtomicReference<Throwable> samplingFailure = new AtomicReference<>(); + private final CountDownLatch stopped = new CountDownLatch(1); + + WritePerformanceResourceTracker(long dataNodePid) throws IOException { + this.dataNodePid = dataNodePid; + if (!isSupported()) { + clockTicksPerSecond = 0; + sampler = null; + System.out.println("resource configuration: DataNode CPU/memory sampling unavailable"); + return; + } + if (dataNodePid <= 0) { + throw new IOException("DataNode PID unavailable: " + dataNodePid); + } + clockTicksPerSecond = readClockTicksPerSecond(); + readProcessSample(); + sampler = new Thread(this::sampleLoop, "numeric-row-resource-sampler"); + sampler.setDaemon(true); + sampler.start(); + System.out.printf( + "resource configuration: dataNodePid=%d, sampleIntervalMs=%d, cpu=process CPU, " + + "memory=resident set size%n", + dataNodePid, SAMPLE_INTERVAL_MILLIS); + } + + boolean isSupported() { + return Files.isDirectory(PROC); + } + + void beginRound() throws IOException { + checkFailure(); + if (!isSupported()) { + return; + } + Round round = new Round(); + round.sample(); + currentRound.set(round); + } + + RoundPeak endRound() throws IOException { + if (!isSupported()) { + return RoundPeak.UNAVAILABLE; + } + Round round = currentRound.getAndSet(null); + if (round == null) { + throw new IllegalStateException("Resource round was not started"); + } + RoundPeak peak = round.finish(); + checkFailure(); + return peak; + } + + @Override + public void close() throws Exception { + if (sampler == null) { + return; + } + sampler.interrupt(); + if (!stopped.await(5, TimeUnit.SECONDS)) { + throw new IllegalStateException("Resource sampler did not stop"); + } + checkFailure(); + } + + private void sampleLoop() { + try { + while (!Thread.currentThread().isInterrupted()) { + Round round = currentRound.get(); + if (round != null) { + round.sample(); + } + Thread.sleep(SAMPLE_INTERVAL_MILLIS); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (Throwable t) { + samplingFailure.compareAndSet(null, t); + } finally { + stopped.countDown(); + } + } + + private void checkFailure() throws IOException { + Throwable failure = samplingFailure.get(); + if (failure == null) { + return; + } + if (failure instanceof IOException ioException) { + throw ioException; + } + throw new IOException("Failed to sample DataNode resources", failure); + } + + private ProcessSample readProcessSample() throws IOException { + Path processDirectory = PROC.resolve(Long.toString(dataNodePid)); + String stat = Files.readString(processDirectory.resolve("stat")); + int closingParenthesis = stat.lastIndexOf(") "); + if (closingParenthesis < 0) { + throw new IOException("Malformed /proc stat for DataNode PID " + dataNodePid); + } + String[] fields = stat.substring(closingParenthesis + 2).trim().split("\\s+"); + if (fields.length <= 12) { + throw new IOException("Incomplete /proc stat for DataNode PID " + dataNodePid); + } + long cpuTicks = Long.parseLong(fields[11]) + Long.parseLong(fields[12]); + long residentBytes = 0; + for (String line : Files.readAllLines(processDirectory.resolve("status"))) { + if (line.startsWith("VmRSS:")) { + residentBytes = Long.parseLong(line.replaceAll("[^0-9]", "").trim()) * 1024; + break; + } + } + if (residentBytes <= 0) { + throw new IOException("DataNode RSS unavailable for PID " + dataNodePid); + } + return new ProcessSample(System.nanoTime(), cpuTicks, residentBytes); + } + + private static long readClockTicksPerSecond() throws IOException { + Process process = new ProcessBuilder("getconf", "CLK_TCK").redirectErrorStream(true).start(); + try { + if (!process.waitFor(5, TimeUnit.SECONDS)) { + throw new IOException("getconf CLK_TCK timed out"); + } + String output = + new String(process.getInputStream().readAllBytes(), StandardCharsets.UTF_8).trim(); + if (process.exitValue() != 0) { + throw new IOException("getconf CLK_TCK failed: " + output); + } + long result = Long.parseLong(output); + if (result <= 0) { + throw new IOException("Invalid CLK_TCK: " + result); + } + return result; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted while reading CLK_TCK", e); + } finally { + process.destroyForcibly(); + } + } + + private final class Round { + private long lastCpuTicks; + private long lastSampleNanos; + private double peakCpuPercent = Double.NaN; + private long peakResidentBytes; + private boolean active = true; + + private synchronized void sample() throws IOException { + if (!active) { + return; + } + // Read and apply samples under the same lock so the final sample cannot overtake the sampler. + ProcessSample sample = readProcessSample(); + peakResidentBytes = Math.max(peakResidentBytes, sample.residentBytes); + long elapsedNanos = sample.nanoTime - lastSampleNanos; + if (lastSampleNanos != 0) { + // Keep the CPU baseline for short intervals: tick quantization can inflate a tail sample. + if (elapsedNanos < TimeUnit.MILLISECONDS.toNanos(SAMPLE_INTERVAL_MILLIS)) { + return; + } + long cpuTicks = sample.cpuTicks - lastCpuTicks; + double cpuPercent = + cpuTicks * 100.0 / clockTicksPerSecond / (elapsedNanos / 1_000_000_000.0); + peakCpuPercent = + Double.isNaN(peakCpuPercent) ? cpuPercent : Math.max(peakCpuPercent, cpuPercent); + } + lastCpuTicks = sample.cpuTicks; + lastSampleNanos = sample.nanoTime; + } + + private synchronized RoundPeak finish() throws IOException { + try { + sample(); + return new RoundPeak(peakCpuPercent, peakResidentBytes); + } finally { + active = false; + } + } + } + + private record ProcessSample(long nanoTime, long cpuTicks, long residentBytes) {} + + record RoundPeak(double peakCpuPercent, long peakResidentBytes) { + private static final RoundPeak UNAVAILABLE = new RoundPeak(Double.NaN, -1); + + String format() { + if (this == UNAVAILABLE) { + return "peakDataNodeCpu=unavailable, peakDataNodeRss=unavailable"; + } + return String.format( + "peakDataNodeCpu=%s, peakDataNodeRss=%.1f MiB", + Double.isNaN(peakCpuPercent) ? "unavailable" : String.format("%.1f%%", peakCpuPercent), + peakResidentBytes / 1024.0 / 1024.0); + } + } +} 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 1cfb0b86eed..7fa1df842f8 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 @@ -52,17 +52,12 @@ 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. + * Prepares repeated single-row writes with STRING TAGs and BOOLEAN, INT32, INT64, FLOAT, DOUBLE, + * DATE or TIMESTAMP FIELDs. ATTRIBUTE columns and other FIELD types are unsupported. Column + * names, types and categories must have matching lengths and order. * - * <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 + * <p>The returned statement belongs to this session and is not thread-safe. Close it after use. */ - void insert(NumericRow row) throws StatementExecutionException, IoTDBConnectionException; - NumericRowPreparedStatement prepareInsertNumericRow( String tableName, List<String> columnNames, @@ -71,10 +66,18 @@ public interface ITableSession extends AutoCloseable { throws StatementExecutionException, IoTDBConnectionException; /** - * Executes a prepared row, following write redirection recommendations for subsequent rows. - * Destination handles are prepared lazily and released when the statement is closed. An execution - * error reported by the server closes the statement; prepare it again before reuse. Transport - * failures are propagated without replaying a write whose outcome is unknown. + * Executes one row using FIELD and TAG arrays in their respective occurrence order in the + * prepared column list. TAGs may be null or empty. + * + * <p>BOOLEAN uses zero for false and nonzero for true. INT32 and DATE use the low 32 bits (DATE + * is 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)}. Do not modify input arrays during execution. + * + * <p>Follows write redirection recommendations for subsequent rows. Destination handles are + * prepared lazily and released when the statement is closed. An execution error reported by the + * server closes the statement; prepare it again before reuse. Transport failures are propagated + * without replaying a write whose outcome is unknown. */ void executePreparedInsertNumericRow( NumericRowPreparedStatement prepared, long timestamp, long[] values, String[] tagValues) 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 deleted file mode 100644 index 040cb7632ba..00000000000 --- a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRow.java +++ /dev/null @@ -1,122 +0,0 @@ -/* - * 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 org.apache.tsfile.utils.BitMap; - -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. An optional null bitmap marks null - * FIELDs by their index in {@code values}; marked slots retain their position but are not sent. - * ATTRIBUTE columns and nonnumeric FIELDs are not supported. - * - * <p>BOOLEAN uses zero for false and nonzero for true. INT32 and DATE use the low 32 bits (DATE is - * encoded as yyyyMMdd); INT64 and TIMESTAMP use the long value. FLOAT uses {@link - * 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; - private final BitMap nullBitmap; - - /** 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, timestamp, columnNames, dataTypes, columnCategories, values, tagValues, null); - } - - /** - * Creates a row with an optional FIELD-only null bitmap, sized exactly to {@code values.length}. - * A marked bit denotes null. The bitmap is retained and must not change during insertion. - * Requires a server supporting numeric row null bitmaps when the bitmap is supplied. - */ - public NumericRow( - String tableName, - long timestamp, - List<String> columnNames, - List<TSDataType> dataTypes, - List<ColumnCategory> columnCategories, - long[] values, - String[] tagValues, - BitMap nullBitmap) { - this.tableName = tableName; - this.timestamp = timestamp; - this.columnNames = columnNames; - this.dataTypes = dataTypes; - this.columnCategories = columnCategories; - this.values = values; - this.tagValues = tagValues; - this.nullBitmap = nullBitmap; - } - - 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; - } - - public BitMap getNullBitmap() { - return nullBitmap; - } -} diff --git a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRowPreparedStatement.java b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRowPreparedStatement.java index 2fa9ee21257..077cc946f24 100644 --- a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRowPreparedStatement.java +++ b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/NumericRowPreparedStatement.java @@ -2,8 +2,21 @@ * 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. + * 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.iotdb.isession.i18n.ISessionMessages; 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 11a545da714..57754c22b10 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,7 +23,6 @@ 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; @@ -67,7 +66,6 @@ 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; @@ -2874,89 +2872,6 @@ 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(); - if (measurements == null - || types == null - || columnCategories == null - || values == null - || tagValues == null - || measurements.size() != types.size() - || measurements.size() != columnCategories.size()) { - throw new IllegalArgumentException( - SessionMessages - .EXCEPTION_NUMERIC_ROW_ARGUMENTS_MUST_BE_NON_NULL_COLUMN_METADATA_LENGTHS_AND_PER_CATEGORY_VALUE_COUNTS_MUST_MATCH_EB12B233); - } - // Keep wire metadata aligned with the compact payload: TAG columns precede FIELD columns. - List<String> rpcMeasurements = new ArrayList<>(measurements.size()); - List<TSDataType> rpcTypes = new ArrayList<>(types.size()); - List<ColumnCategory> rpcColumnCategories = new ArrayList<>(columnCategories.size()); - for (int pass = 0; pass < 2; pass++) { - ColumnCategory category = pass == 0 ? ColumnCategory.TAG : ColumnCategory.FIELD; - for (int i = 0; i < columnCategories.size(); i++) { - if (columnCategories.get(i) == category) { - rpcMeasurements.add(measurements.get(i)); - rpcTypes.add(types.get(i)); - rpcColumnCategories.add(category); - } - } - } - if (rpcMeasurements.size() != measurements.size()) { - throw new IllegalArgumentException( - SessionMessages - .EXCEPTION_NUMERIC_ROWS_SUPPORT_NUMERIC_FIELD_COLUMNS_AND_STRING_TAG_COLUMNS_ONLY_87159FFA); - } - ByteBuffer valueBuffer = - SessionUtils.getValueBuffer( - rpcTypes, values, rpcMeasurements, rpcColumnCategories, tagValues, row.getNullBitmap()); - TSInsertRecordReq request = new TSInsertRecordReq(); - request.setPrefixPath(tableName); - request.setTimestamp(timestamp); - request.setMeasurements(rpcMeasurements); - request.setValues(valueBuffer); - request.setIsWriteToTable(true); - request.setIsNumericRow(true); - if (row.getNullBitmap() != null) { - request.setNullBitmap(SessionUtils.encodeNumericRowNullBitmap(row.getNullBitmap(), null)); - } - request.setIsAligned(true); - request.setColumnCategoryies( - rpcColumnCategories.stream() - .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 c453596c1ee..6d7ddc1f64c 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,7 +20,6 @@ package org.apache.iotdb.session; import org.apache.iotdb.isession.ITableSession; -import org.apache.iotdb.isession.NumericRow; import org.apache.iotdb.isession.NumericRowPreparedStatement; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.rpc.IoTDBConnectionException; @@ -55,11 +54,6 @@ public class TableSession implements ITableSession { session.insertRelationalTablet(tablet); } - @Override - public void insert(NumericRow row) throws StatementExecutionException, IoTDBConnectionException { - session.insertRelationalRecord(row); - } - @Override public NumericRowPreparedStatement prepareInsertNumericRow( String tableName, 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 7684beba9ab..331a078ae9e 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,7 +20,6 @@ package org.apache.iotdb.session.pool; import org.apache.iotdb.isession.ITableSession; -import org.apache.iotdb.isession.NumericRow; import org.apache.iotdb.isession.NumericRowPreparedStatement; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.rpc.IoTDBConnectionException; @@ -78,19 +77,6 @@ public class TableSessionWrapper implements ITableSession { } } - @Override - public void insert(NumericRow row) throws StatementExecutionException, IoTDBConnectionException { - checkOpen(); - try { - session.insertRelationalRecord(row); - } catch (IoTDBConnectionException e) { - sessionPool.cleanSessionAndMayThrowConnectionException(session); - closed.set(true); - session = null; - throw e; - } - } - @Override public NumericRowPreparedStatement prepareInsertNumericRow( String tableName, 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 fd1dd00cc38..3c316824334 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,109 +163,6 @@ 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 { - return getValueBuffer(types, values, measurements, categories, tagValues, null); - } - - /** Null FIELD slots retain their indices but consume no payload bytes. */ - public static ByteBuffer getValueBuffer( - List<TSDataType> types, - long[] values, - List<String> measurements, - List<ColumnCategory> categories, - String[] tagValues, - BitMap nullBitmap) - throws IoTDBConnectionException { - if (types == null - || values == null - || measurements == null - || categories == null - || tagValues == null - || types.size() != measurements.size() - || types.size() != categories.size()) { - throw invalidNumericRowArguments(); - } - validateNumericRowNullBitmap(nullBitmap, values.length); - 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) { - if (fieldCount >= values.length) { - throw invalidNumericRowArguments(); - } - int valueSize = - switch (type) { - case BOOLEAN -> Byte.BYTES; - case INT32, DATE, FLOAT -> Integer.BYTES; - case INT64, TIMESTAMP, DOUBLE -> Long.BYTES; - default -> throw new IoTDBConnectionException(MSG_UNSUPPORTED_DATA_TYPE + type); - }; - if (nullBitmap == null || !nullBitmap.isMarked(fieldCount)) { - length += valueSize; - } - fieldCount++; - } else { - throw new IllegalArgumentException( - SessionMessages - .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 (TSDataType type : types) { - ReadWriteIOUtils.write(type, buffer); - } - tagIndex = 0; - for (int i = 0; i < types.size(); i++) { - TSDataType type = types.get(i); - if (categories.get(i) == ColumnCategory.TAG && type == TSDataType.STRING) { - ReadWriteIOUtils.write(tagValues[tagIndex++], buffer); - } - } - fieldIndex = 0; - for (int i = 0; i < types.size(); i++) { - TSDataType type = types.get(i); - if (categories.get(i) == ColumnCategory.FIELD) { - long value = values[fieldIndex++]; - if (nullBitmap != null && nullBitmap.isMarked(fieldIndex - 1)) { - continue; - } - switch (type) { - case BOOLEAN -> buffer.put((byte) (value == 0 ? 0 : 1)); - case INT32, DATE, FLOAT -> buffer.putInt((int) value); - 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 void validateNumericRowNullBitmap(BitMap nullBitmap, int fieldCount) { if (nullBitmap != null && nullBitmap.getSize() != fieldCount) { throw new IllegalArgumentException( diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/PreparedNumericRowBufferTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/PreparedNumericRowBufferTest.java index 7a44a165361..6e889f56875 100644 --- a/iotdb-client/session/src/test/java/org/apache/iotdb/session/PreparedNumericRowBufferTest.java +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/PreparedNumericRowBufferTest.java @@ -30,6 +30,7 @@ import org.junit.Test; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import static org.junit.Assert.assertEquals; @@ -42,9 +43,64 @@ import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class PreparedNumericRowBufferTest { + @Test + public void testAllNumericTypesPreserveRawBits() throws Exception { + Session session = mock(Session.class); + SessionConnection connection = mock(SessionConnection.class); + when(session.getDefaultSessionConnection()).thenReturn(connection); + when(connection.prepareNumericRow(any())).thenReturn(1L); + TableSession tableSession = new TableSession(session); + NumericRowPreparedStatement prepared = + tableSession.prepareInsertNumericRow( + "t", + List.of("b", "i", "l", "f", "d", "date", "ts"), + List.of( + TSDataType.BOOLEAN, + TSDataType.INT32, + TSDataType.INT64, + TSDataType.FLOAT, + TSDataType.DOUBLE, + TSDataType.DATE, + TSDataType.TIMESTAMP), + Collections.nCopies(7, ColumnCategory.FIELD)); + doAnswer( + invocation -> { + ByteBuffer values = invocation.getArgument(2); + assertEquals(37, values.remaining()); + assertEquals(1, values.get()); + assertEquals(Integer.MIN_VALUE, values.getInt()); + assertEquals(Long.MIN_VALUE, values.getLong()); + assertEquals(0x7fc01234, values.getInt()); + assertEquals(0x7ff8000000001234L, values.getLong()); + assertEquals(20260910, values.getInt()); + assertEquals(Long.MAX_VALUE, values.getLong()); + assertFalse(values.hasRemaining()); + return null; + }) + .when(connection) + .executePreparedNumericRow(eq(1L), anyLong(), any(ByteBuffer.class), any(ByteBuffer.class)); + // High bits of 32-bit fields are ignored; noncanonical NaNs must survive without normalization. + tableSession.executePreparedInsertNumericRow( + prepared, + 42, + new long[] { + -1, + 0x1234567880000000L, + Long.MIN_VALUE, + 0x123456787fc01234L, + 0x7ff8000000001234L, + 0x1234567800000000L | 20260910, + Long.MAX_VALUE + }, + new String[0]); + verify(connection) + .executePreparedNumericRow(eq(1L), eq(42L), any(ByteBuffer.class), any(ByteBuffer.class)); + } + // Reusing the statement after sparse/all-null rows must reset both payload and bitmap cursors. @Test public void testNullBitmapBufferReuse() throws Exception { 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 9f79ee200d9..e4d2dbbb41d 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,18 +22,13 @@ 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; @@ -46,7 +41,6 @@ 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; @@ -140,104 +134,6 @@ 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("enabled", "tag", "value", "region"); - List<TSDataType> types = - Arrays.asList(TSDataType.BOOLEAN, TSDataType.STRING, TSDataType.DOUBLE, TSDataType.STRING); - List<ColumnCategory> categories = - Arrays.asList( - ColumnCategory.FIELD, ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.TAG); - List<String> rpcNames = Arrays.asList("tag", "region", "enabled", "value"); - List<TSDataType> rpcTypes = - Arrays.asList(TSDataType.STRING, TSDataType.STRING, TSDataType.BOOLEAN, TSDataType.DOUBLE); - List<ColumnCategory> rpcCategories = - Arrays.asList( - ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.FIELD); - tableSession.insert( - new NumericRow( - "readings", - 123L, - names, - types, - categories, - new long[] {1, Double.doubleToRawLongBits(2.5)}, - new String[] {"device", "cn"})); - ArgumentCaptor<TSInsertRecordReq> captor = ArgumentCaptor.forClass(TSInsertRecordReq.class); - Mockito.verify(sessionConnection).insertRecord(captor.capture()); - TSInsertRecordReq request = captor.getValue(); - Assert.assertTrue(request.isIsNumericRow()); - Assert.assertTrue(request.isIsWriteToTable()); - assertEquals( - Arrays.asList( - (byte) ColumnCategory.TAG.ordinal(), - (byte) ColumnCategory.TAG.ordinal(), - (byte) ColumnCategory.FIELD.ordinal(), - (byte) ColumnCategory.FIELD.ordinal()), - request.getColumnCategoryies()); - assertEquals(rpcNames, request.getMeasurements()); - assertEquals( - SessionUtils.getValueBuffer( - rpcTypes, - new long[] {1, Double.doubleToRawLongBits(2.5)}, - rpcNames, - rpcCategories, - new String[] {"device", "cn"}), - 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/pool/PreparedNumericRowPoolTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/pool/PreparedNumericRowPoolTest.java index f65f761c7b9..7a356747001 100644 --- a/iotdb-client/session/src/test/java/org/apache/iotdb/session/pool/PreparedNumericRowPoolTest.java +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/pool/PreparedNumericRowPoolTest.java @@ -20,7 +20,6 @@ package org.apache.iotdb.session.pool; import org.apache.iotdb.common.rpc.thrift.TSStatus; -import org.apache.iotdb.isession.NumericRow; import org.apache.iotdb.isession.NumericRowPreparedStatement; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.StatementExecutionException; @@ -114,28 +113,6 @@ public class PreparedNumericRowPoolTest { verify(pool, times(2)).putBack(session); } - @Test - public void testClosedWrapperCannotInsertNumericRow() throws Exception { - wrapper.close(); - - IoTDBConnectionException error = - assertThrows( - IoTDBConnectionException.class, - () -> - wrapper.insert( - new NumericRow( - "t", - 1L, - List.of("tag", "v"), - List.of(TSDataType.STRING, TSDataType.INT64), - List.of(ColumnCategory.TAG, ColumnCategory.FIELD), - new long[] {42L}, - new String[] {"a"}))); - - assertEquals(SessionMessages.EXCEPTION_TABLE_SESSION_IS_CLOSED_77FE359B, error.getMessage()); - verifyZeroInteractions(client); - } - @Test public void testPrepareTransportFailureDiscardsSession() throws Exception { TException transportFailure = new TException("transport closed"); 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 632fbbda8fb..977c6113be9 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,7 +28,6 @@ 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; @@ -46,38 +45,8 @@ import static org.junit.Assert.assertThrows; public class SessionUtilsTest { - // Null FIELDs are omitted without shifting indices around interleaved TAG metadata or NaN bits. @Test - public void testNumericRowNullBitmapEncoding() throws Exception { - BitMap nulls = new BitMap(9); - nulls.mark(0); - nulls.mark(7); - nulls.mark(8); - List<TSDataType> types = new ArrayList<>(Collections.nCopies(9, TSDataType.DOUBLE)); - List<ColumnCategory> categories = new ArrayList<>(Collections.nCopies(9, ColumnCategory.FIELD)); - types.add(3, TSDataType.STRING); - categories.add(3, ColumnCategory.TAG); - long[] values = new long[9]; - Arrays.fill(values, 0x7ff8000000001234L); - ByteBuffer payload = - SessionUtils.getValueBuffer( - types, values, Collections.nCopies(10, "v"), categories, new String[] {"tag"}, nulls); - for (TSDataType type : types) { - Assert.assertEquals(type.serialize(), payload.get()); - } - Assert.assertEquals("tag", ReadWriteIOUtils.readString(payload)); - for (int i = 0; i < 6; i++) { - Assert.assertEquals(0x7ff8000000001234L, payload.getLong()); - } - Assert.assertFalse(payload.hasRemaining()); - Assert.assertEquals( - ByteBuffer.wrap(new byte[] {(byte) 0x81, 1}), - SessionUtils.encodeNumericRowNullBitmap(nulls, null)); - } - - // markAll may set padding in BitMap; the wire representation uses exactly ceil(FIELDs/8) bytes. - @Test - public void testNumericRowNullBitmapBoundaries() throws Exception { + public void testNumericRowNullBitmapBoundaries() { for (int size : new int[] {1, 7, 8, 9, 16}) { BitMap nulls = new BitMap(size); nulls.markAll(); @@ -87,177 +56,10 @@ public class SessionUtilsTest { Assert.assertEquals( size % 8 == 0 ? 255 : (1 << (size % 8)) - 1, bitmap.get(bitmap.limit() - 1) & 255); Assert.assertArrayEquals(original, nulls.getByteArray()); - ByteBuffer payload = - SessionUtils.getValueBuffer( - Collections.nCopies(size, TSDataType.INT64), - new long[size], - Collections.nCopies(size, "v"), - Collections.nCopies(size, ColumnCategory.FIELD), - new String[0], - nulls); - Assert.assertEquals(size, payload.remaining()); } assertThrows( IllegalArgumentException.class, - () -> - SessionUtils.getValueBuffer( - List.of(TSDataType.INT64), - new long[1], - List.of("v"), - List.of(ColumnCategory.FIELD), - new String[0], - new BitMap(2))); - } - - @Test - public void testNumericRowEncodingPreservesRawBits() throws IoTDBConnectionException { - List<TSDataType> types = - 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); - for (TSDataType type : types) { - expected.put(type.serialize()); - } - expected.put((byte) 0); - expected.put((byte) 1); - expected.putInt(Integer.MIN_VALUE); - expected.putLong(Long.MIN_VALUE); - expected.putInt(0x7fc01234); - expected.putLong(0x7ff8000000001234L); - expected.putInt(20260910); - expected.putLong(Long.MAX_VALUE); - expected.flip(); - Assert.assertEquals(expected, actual); - } - - @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, ""}); - for (TSDataType type : types) { - Assert.assertEquals(type.serialize(), buffer.get()); - } - Assert.assertEquals("设备😀", ReadWriteIOUtils.readString(buffer)); - Assert.assertNull(ReadWriteIOUtils.readString(buffer)); - Assert.assertEquals("", ReadWriteIOUtils.readString(buffer)); - Assert.assertEquals(0x7ff8000000001234L, buffer.getLong()); - Assert.assertEquals(1, buffer.get()); - 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"})); + () -> SessionUtils.validateNumericRowNullBitmap(new BitMap(2), 1)); } @Test 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 eb8cebad679..874f8ffa02e 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 @@ -49,7 +49,6 @@ 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; @@ -72,7 +71,6 @@ 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; @@ -285,10 +283,7 @@ public class StatementGenerator { throws IllegalPathException, QueryProcessException { final long startTime = System.nanoTime(); // construct insert statement - InsertBaseStatement insertStatement = - insertRecordReq.isIsNumericRow() - ? new InsertNumericRowStatement() - : new InsertRowStatement(); + InsertRowStatement insertStatement = new InsertRowStatement(); insertStatement.setDevicePath( insertRecordReq.isIsWriteToTable() ? new PartialPath(insertRecordReq.getPrefixPath(), false) @@ -297,20 +292,8 @@ public class StatementGenerator { insertStatement.setMeasurements(insertRecordReq.getMeasurements().toArray(new String[0])); insertStatement.setAligned(insertRecordReq.isAligned); - 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()); - insertStatement.setAligned(true); - } else { - InsertRowStatement row = (InsertRowStatement) insertStatement; - row.setTime(insertRecordReq.getTimestamp()); - row.fillValues(insertRecordReq.values); - } + insertStatement.setTime(insertRecordReq.getTimestamp()); + insertStatement.fillValues(insertRecordReq.values); if (insertRecordReq.isIsWriteToTable()) { insertStatement.setWriteToTable(insertRecordReq.isIsWriteToTable()); if (!insertRecordReq.isSetColumnCategoryies() @@ -327,20 +310,6 @@ public class StatementGenerator { ColumnCategory.values()[insertRecordReq.getColumnCategoryies().get(i).intValue()]); } insertStatement.setColumnCategories(columnCategories); - if (insertRecordReq.isIsNumericRow()) { - ((InsertNumericRowStatement) insertStatement) - .fillValues(insertRecordReq.values, insertRecordReq.bufferForNullBitmap()); - for (int i = 0; i < columnCategories.length; i++) { - TSDataType type = insertStatement.getDataType(i); - if (!(columnCategories[i] == TsTableColumnCategory.TAG && type == TSDataType.STRING) - && !(columnCategories[i] == TsTableColumnCategory.FIELD - && NumericRowValues.supports(type))) { - throw new IllegalArgumentException( - DataNodeQueryMessages - .QUERY_EXCEPTION_MISSING_OR_INVALID_COLUMN_CATEGORIES_FOR_TABLE_INSERTION_5DF990B9); - } - } - } } PERFORMANCE_OVERVIEW_METRICS.recordParseCost(System.nanoTime() - startTime); return insertStatement; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertNumericRowStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertNumericRowStatement.java index 62970448eb4..46b70fea2c6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertNumericRowStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertNumericRowStatement.java @@ -19,7 +19,6 @@ 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; @@ -42,9 +41,7 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.utils.Pair; import org.apache.tsfile.utils.RamUsageEstimator; -import org.apache.tsfile.write.UnSupportedDataTypeException; -import java.nio.ByteBuffer; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -117,68 +114,6 @@ public class InsertNumericRowStatement extends InsertBaseStatement { 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 { - fillValues(buffer, null); - } - - public void fillValues(ByteBuffer buffer, ByteBuffer nullBitmap) throws QueryProcessException { - dataTypes = new TSDataType[measurements.length]; - for (int i = 0; i < dataTypes.length; i++) { - byte type = buffer.get(); - if (type < 0 || type >= TSDataType.values().length) { - throw new QueryProcessException(DataNodeQueryMessages.UNSUPPORTED_DATA_TYPE + type); - } - dataTypes[i] = TSDataType.values()[type]; - if (columnCategories[i] == TsTableColumnCategory.TAG && dataTypes[i] != TSDataType.STRING) { - throw new IllegalArgumentException( - DataNodeQueryMessages - .QUERY_EXCEPTION_MISSING_OR_INVALID_COLUMN_CATEGORIES_FOR_TABLE_INSERTION_5DF990B9); - } - if (columnCategories[i] == TsTableColumnCategory.FIELD - && !NumericRowValues.supports(dataTypes[i])) { - if (dataTypes[i] == TSDataType.STRING) { - throw new IllegalArgumentException( - DataNodeQueryMessages - .QUERY_EXCEPTION_MISSING_OR_INVALID_COLUMN_CATEGORIES_FOR_TABLE_INSERTION_5DF990B9); - } - throw new UnSupportedDataTypeException( - DataNodeQueryMessages.UNSUPPORTED_DATA_TYPE + dataTypes[i]); - } - } - int tagCount = countTagColumns(columnCategories); - NumericRowValues.validateNullBitmap(nullBitmap, measurements.length - tagCount); - numericValues = new NumericRowValues(tagCount, measurements.length - tagCount); - ColumnCategory[] categories = new ColumnCategory[measurements.length]; - for (int i = 0; i < measurements.length; i++) { - categories[i] = - columnCategories[i] == TsTableColumnCategory.TAG - ? ColumnCategory.TAG - : ColumnCategory.FIELD; - } - numericValues.setColumnCategories(categories); - for (int i = 0; i < dataTypes.length; i++) { - if (columnCategories[i] == TsTableColumnCategory.TAG) { - numericValues.read(i, dataTypes[i], buffer); - } - } - for (int i = 0; i < dataTypes.length; i++) { - if (columnCategories[i] == TsTableColumnCategory.FIELD) { - if (NumericRowValues.isNullField(nullBitmap, i - tagCount)) { - numericValues.setNull(i); - } else { - numericValues.read(i, dataTypes[i], buffer); - } - } - } - if (buffer.hasRemaining()) { - throw new IllegalArgumentException( - String.format( - DataNodeQueryMessages.EXCEPTION_NUMERIC_ROW_PAYLOAD_HAS_ARG_TRAILING_BYTES_89E46A34, - buffer.remaining())); - } - } - @Override public <R, C> R accept(StatementVisitor<R, C> visitor, C context) { return visitor.visitInsertNumericRow(this, context); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfoTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfoTest.java index cf3533296b1..9fc462abfff 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfoTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfoTest.java @@ -30,14 +30,12 @@ import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager; import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType; import org.apache.iotdb.db.queryengine.plan.analyze.schema.SchemaValidator; -import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator; import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.TableHeaderSchemaValidator; import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertNumericRow; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertNumericRowStatement; import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; -import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq; import org.apache.iotdb.service.rpc.thrift.TSPrepareNumericRowReq; import org.apache.tsfile.enums.ColumnCategory; @@ -289,25 +287,6 @@ public class PreparedNumericRowInfoTest { InsertNumericRowStatement prepared = info.bind(1, tagBuffer(1), values, bitmap); assertEquals(1, bitmap.position()); assertEquals(0, values.position()); - ByteBuffer payload = ByteBuffer.allocate(10 + 5 + values.remaining()); - payload.put(TSDataType.STRING.serialize()); - for (int i = 0; i < 9; i++) { - payload.put(TSDataType.DOUBLE.serialize()); - } - payload.put(tagBuffer(1)).put(values.duplicate()).flip(); - TSInsertRecordReq insertReq = new TSInsertRecordReq(); - insertReq - .setPrefixPath("t") - .setTimestamp(1) - .setMeasurements(req.getMeasurements()) - .setColumnCategoryies(req.getColumnCategories()) - .setIsNumericRow(true) - .setIsWriteToTable(true) - .setValues(payload) - .setNullBitmap(bitmap); - InsertNumericRowStatement direct = - (InsertNumericRowStatement) StatementGenerator.createStatement(insertReq); - assertEquals(prepared.getNumericValues(), direct.getNumericValues()); assertTrue(prepared.getNumericValues().isNull(1)); assertTrue(prepared.getNumericValues().isNull(8)); assertTrue(prepared.getNumericValues().isNull(9)); @@ -317,17 +296,11 @@ public class PreparedNumericRowInfoTest { assertThrows( IllegalArgumentException.class, () -> info.bind(1, tagBuffer(1), values, ByteBuffer.wrap(bad))); - insertReq.setValues(payload.duplicate().rewind()).setNullBitmap(bad); - assertThrows( - IllegalArgumentException.class, () -> StatementGenerator.createStatement(insertReq)); } assertThrows( SemanticException.class, () -> info.bind(1, tagBuffer(1), ByteBuffer.allocate(9 * 8), bitmap)); ByteBuffer allNulls = ByteBuffer.wrap(new byte[] {(byte) 255, 1}); - insertReq.setValues(payload.duplicate().rewind()).setNullBitmap(allNulls); - assertThrows( - IllegalArgumentException.class, () -> StatementGenerator.createStatement(insertReq)); InsertNumericRowStatement allNull = info.bind(1, tagBuffer(1), ByteBuffer.allocate(0), allNulls); for (int i = 1; i <= 9; i++) { @@ -335,6 +308,29 @@ public class PreparedNumericRowInfoTest { } } + @Test + public void testUnsupportedColumnTypesRejectedDuringPrepare() { + for (TSDataType type : + new TSDataType[] { + TSDataType.TEXT, + TSDataType.STRING, + TSDataType.BLOB, + TSDataType.OBJECT, + TSDataType.VECTOR, + TSDataType.UNKNOWN + }) { + TSPrepareNumericRowReq req = request("v"); + req.getDataTypes().set(0, type.ordinal()); + assertThrows(SemanticException.class, () -> new PreparedNumericRowInfo(DATABASE, req)); + } + TSPrepareNumericRowReq req = request("tag", "v"); + req.getDataTypes().set(0, TSDataType.INT64.ordinal()); + assertThrows(SemanticException.class, () -> new PreparedNumericRowInfo(DATABASE, req)); + req.getDataTypes().set(0, TSDataType.STRING.ordinal()); + req.getColumnCategories().set(0, (byte) ColumnCategory.ATTRIBUTE.ordinal()); + assertThrows(SemanticException.class, () -> new PreparedNumericRowInfo(DATABASE, req)); + } + private static void install(TsTable table) { DataNodeTableCache.getInstance().preUpdateTable(DATABASE, table, null); DataNodeTableCache.getInstance().commitUpdateTable(DATABASE, "t", null); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/NumericTableRowTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/NumericTableRowTest.java index fbb69b4be04..cc85949aaad 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/NumericTableRowTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/NumericTableRowTest.java @@ -34,6 +34,7 @@ import org.apache.iotdb.consensus.common.request.IoTConsensusRequest; import org.apache.iotdb.db.consensus.statemachine.dataregion.IoTConsensusDataRegionStateMachine; import org.apache.iotdb.db.pipe.event.common.tablet.parser.TabletInsertionEventTablePatternParser; import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletBatchReqV2; +import org.apache.iotdb.db.protocol.session.PreparedNumericRowInfo; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertNumericRowNode; import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis; @@ -53,8 +54,7 @@ import org.apache.iotdb.db.subscription.broker.consensus.ConsensusLogToTabletCon import org.apache.iotdb.db.utils.NumericRowValues; import org.apache.iotdb.db.utils.datastructure.AlignedTVList; import org.apache.iotdb.rpc.TSStatusCode; -import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq; -import org.apache.iotdb.session.util.SessionUtils; +import org.apache.iotdb.service.rpc.thrift.TSPrepareNumericRowReq; import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.ColumnCategory; @@ -62,7 +62,6 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.read.common.type.TypeFactory; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.ReadWriteIOUtils; -import org.apache.tsfile.write.UnSupportedDataTypeException; import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.MeasurementSchema; import org.junit.Test; @@ -81,10 +80,9 @@ import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; -/** Exercises one numeric row from the client codec through schema edits, WAL and memtable. */ +/** Exercises one numeric row from prepared binding through schema edits, WAL and memtable. */ public class NumericTableRowTest { private static final List<String> NAMES = Arrays.asList("b", "i", "l", "f", "d", "date", "ts"); private static final List<TSDataType> TYPES = @@ -100,30 +98,32 @@ public class NumericTableRowTest { 1, Integer.MIN_VALUE, Long.MIN_VALUE, 0x7fc01234, 0x7ff8000000001234L, 20260910, Long.MAX_VALUE }; - private TSInsertRecordReq request() throws Exception { - return new TSInsertRecordReq() - .setPrefixPath("numeric_table") - .setTimestamp(42) - .setMeasurements(NAMES) - .setValues( - SessionUtils.getValueBuffer( - TYPES, - BITS, - NAMES, - Collections.nCopies(NAMES.size(), ColumnCategory.FIELD), - new String[0])) - .setIsAligned(true) - .setIsWriteToTable(true) - .setIsNumericRow(true) - .setColumnCategoryies( - Collections.nCopies(NAMES.size(), (byte) ColumnCategory.FIELD.ordinal())); + private InsertNumericRowStatement preparedRow() { + TSPrepareNumericRowReq req = + new TSPrepareNumericRowReq() + .setTableName("numeric_table") + .setMeasurements(NAMES) + .setDataTypes(TYPES.stream().map(TSDataType::ordinal).toList()) + .setColumnCategories( + Collections.nCopies(NAMES.size(), (byte) ColumnCategory.FIELD.ordinal())); + ByteBuffer values = ByteBuffer.allocate(37); + values + .put((byte) BITS[0]) + .putInt((int) BITS[1]) + .putLong(BITS[2]) + .putInt((int) BITS[3]) + .putLong(BITS[4]) + .putInt((int) BITS[5]) + .putLong(BITS[6]) + .flip(); + return new PreparedNumericRowInfo("numeric_test", req) + .bind(42, ByteBuffer.allocate(0), values, null); } @Test - public void testRpcDecodeAndSchemaChangesRetainPrimitiveStorage() throws Exception { + public void testPreparedBindAndSchemaChangesRetainPrimitiveStorage() throws Exception { // Missing TAG insertion, column swaps and partial failure recovery must retain long[] values. - InsertNumericRowStatement statement = - (InsertNumericRowStatement) StatementGenerator.createStatement(request()); + InsertNumericRowStatement statement = preparedRow(); assertArrayEquals(BITS, statement.getNumericValues().getValues()); assertObjectStorageAbsent(InsertNumericRowStatement.class, statement); assertTrue(statement.isWriteToTable()); @@ -149,18 +149,6 @@ public class NumericTableRowTest { assertEquals("numeric_table", statement.getTableDeviceID().getTableName()); } - @Test - public void testRpcRejectsNumericTreeAndNonStringTagRequests() throws Exception { - // Raw RPC callers must use STRING TAGs and numeric FIELDs in the table model. - TSInsertRecordReq tree = request().setIsWriteToTable(false); - assertThrows(IllegalArgumentException.class, () -> StatementGenerator.createStatement(tree)); - TSInsertRecordReq tag = - request() - .setColumnCategoryies( - Collections.nCopies(NAMES.size(), (byte) ColumnCategory.TAG.ordinal())); - assertThrows(IllegalArgumentException.class, () -> StatementGenerator.createStatement(tag)); - } - @Test public void testStringTagsSurviveSchemaChanges() throws Exception { // The client places TAG metadata before FIELD metadata while retaining dense value arrays. @@ -170,24 +158,20 @@ public class NumericTableRowTest { List<ColumnCategory> categories = Arrays.asList( ColumnCategory.TAG, ColumnCategory.TAG, ColumnCategory.FIELD, ColumnCategory.FIELD); - TSInsertRecordReq req = - request() + TSPrepareNumericRowReq req = + new TSPrepareNumericRowReq() + .setTableName("numeric_table") .setMeasurements(names) - .setValues( - SessionUtils.getValueBuffer( - types, - new long[] {1, Long.MIN_VALUE}, - names, - categories, - new String[] {"设备😀", ""})) - .setColumnCategoryies( - Arrays.asList( - (byte) ColumnCategory.TAG.ordinal(), - (byte) ColumnCategory.TAG.ordinal(), - (byte) ColumnCategory.FIELD.ordinal(), - (byte) ColumnCategory.FIELD.ordinal())); + .setDataTypes(types.stream().map(TSDataType::ordinal).toList()) + .setColumnCategories(categories.stream().map(c -> (byte) c.ordinal()).toList()); + ByteBuffer tags = ByteBuffer.allocate(64); + ReadWriteIOUtils.write("设备😀", tags); + ReadWriteIOUtils.write("", tags); + tags.flip(); + ByteBuffer values = ByteBuffer.allocate(9); + values.put((byte) 1).putLong(Long.MIN_VALUE).flip(); InsertNumericRowStatement statement = - (InsertNumericRowStatement) StatementGenerator.createStatement(req); + new PreparedNumericRowInfo("numeric_test", req).bind(42, tags, values, null); assertEquals("设备😀", statement.getNumericValues().getTag(0)); assertEquals("", statement.getNumericValues().getTag(1)); assertArrayEquals(new long[] {1, Long.MIN_VALUE}, statement.getValues()); @@ -241,8 +225,7 @@ public class NumericTableRowTest { @Test public void testBulkTagExpansionKeepsValuesAndNulls() throws Exception { // A reversed schema mapping with two missing TAGs must preserve numeric bits and null slots. - InsertNumericRowStatement statement = - (InsertNumericRowStatement) StatementGenerator.createStatement(request()); + InsertNumericRowStatement statement = preparedRow(); statement.rebuildArraysAfterExpansion( new int[] {-1, -1, 6, 5, 4, 3, 2, 1, 0}, new String[] {"tag1", "tag2", "ts", "date", "d", "f", "l", "i", "b"}); @@ -384,8 +367,7 @@ public class NumericTableRowTest { @Test public void testNumericRowRedirectUsesSingleEndpoint() throws Exception { - InsertNumericRowStatement statement = - (InsertNumericRowStatement) StatementGenerator.createStatement(request()); + InsertNumericRowStatement statement = preparedRow(); Analysis analysis = new Analysis(new InsertNumericRow(statement, null), Collections.emptyMap()); TEndPoint local = new TEndPoint("127.0.0.1", 6667); TEndPoint leader = new TEndPoint("127.0.0.1", 6668); @@ -489,35 +471,6 @@ public class NumericTableRowTest { } } - @Test - public void testRpcRejectsNonNumericTypes() throws Exception { - // Raw RPC callers cannot smuggle unsupported values into numeric FIELD storage. - for (TSDataType type : - Arrays.asList( - TSDataType.TEXT, - TSDataType.BLOB, - TSDataType.OBJECT, - TSDataType.VECTOR, - TSDataType.UNKNOWN)) { - TSInsertRecordReq invalid = - request().setValues(ByteBuffer.wrap(new byte[] {type.serialize()})); - assertThrows( - UnSupportedDataTypeException.class, () -> StatementGenerator.createStatement(invalid)); - } - // STRING has a valid wire representation for TAGs, but must still be rejected as a FIELD. - ByteBuffer stringValue = ByteBuffer.allocate(32); - ReadWriteIOUtils.write(TSDataType.STRING, stringValue); - ReadWriteIOUtils.write("invalid", stringValue); - stringValue.flip(); - TSInsertRecordReq stringField = - request() - .setMeasurements(Collections.singletonList("s")) - .setColumnCategoryies(Collections.singletonList((byte) ColumnCategory.FIELD.ordinal())) - .setValues(stringValue); - assertThrows( - IllegalArgumentException.class, () -> StatementGenerator.createStatement(stringField)); - } - @Test public void testTabletAdaptersDoNotMaterializeObjectRows() throws Exception { // Pipe and subscription must preserve values and null TAGs without boxing the entire row. @@ -548,8 +501,7 @@ public class NumericTableRowTest { } private RelationalInsertNumericRowNode nodeWithNullTag() throws Exception { - InsertNumericRowStatement statement = - (InsertNumericRowStatement) StatementGenerator.createStatement(request()); + InsertNumericRowStatement statement = preparedRow(); statement.insertColumn( 0, new ColumnSchema( diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/client.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/client.thrift index edcf8834804..b01ef1a2946 100644 --- a/iotdb-protocol/thrift-datanode/src/main/thrift/client.thrift +++ b/iotdb-protocol/thrift-datanode/src/main/thrift/client.thrift @@ -248,11 +248,6 @@ struct TSInsertRecordReq { 6: optional bool isAligned 7: optional bool isWriteToTable 8: optional list<byte> columnCategoryies - // Numeric table rows retain primitive storage throughout the server write path. - 9: optional bool isNumericRow - // Numeric FIELDs only, in FIELD occurrence order: bit i (LSB first) means null. - // Exactly ceil(FIELD count / 8) bytes, unused high bits zero; null FIELD payloads are omitted. - 10: optional binary nullBitmap } struct TSInsertStringRecordReq { @@ -738,6 +733,7 @@ struct TSExecutePreparedNumericRowReq { 4: required binary values // Length-prefixed UTF-8 TAGs; length -1 represents a null TAG. 5: required binary tagValues - // Same FIELD-only bitmap and omitted-null-payload encoding as TSInsertRecordReq. + // FIELD-only bitmap: bit i (LSB first) marks FIELD i as null; null values omit payload bytes. + // Exactly ceil(FIELD count / 8) bytes, with unused high bits zero; absent means no nulls. 6: optional binary nullBitmap }
