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 ed0e2661d3e3f30c832aa560133a3ae688f17027 Author: Tian Jiang <[email protected]> AuthorDate: Thu Sep 17 14:22:22 2026 +0800 retain compatible table schema --- .../it/session/IoTDBTableModelSessionIT.java | 60 ++++++++++++++-- .../protocol/session/PreparedNumericRowInfo.java | 49 +++++++++++--- .../protocol/thrift/impl/ClientRPCServiceImpl.java | 9 ++- .../plan/relational/sql/ast/InsertNumericRow.java | 29 +++++++- .../statement/crud/InsertNumericRowStatement.java | 19 ++++++ .../session/PreparedNumericRowInfoTest.java | 79 ++++++++++++++++++++-- 6 files changed, 223 insertions(+), 22 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBTableModelSessionIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBTableModelSessionIT.java index 9ac67ad9653..59169cb8554 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 @@ -60,6 +60,7 @@ import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.showTa 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; import static org.junit.Assert.fail; @@ -196,7 +197,8 @@ public class IoTDBTableModelSessionIT { @Test public void testPreparedNumericRowSchemaVersionChange() throws Exception { - // Reuse a prepared row before and after adding a TAG, including a flush/read round trip. + // Refresh compatible FIELD additions, validate TAG padding, then verify a flush/read round + // trip. try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) { session.executeNonQueryStatement("CREATE DATABASE numeric_version"); session.executeNonQueryStatement("USE numeric_version"); @@ -207,8 +209,11 @@ public class IoTDBTableModelSessionIT { Arrays.asList("tag", "v"), Arrays.asList(TSDataType.STRING, TSDataType.DOUBLE), Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD)); - for (int time = 1; time <= 3; time++) { - if (time == 3) { + for (int time = 1; time <= 5; time++) { + if (time == 2) { + session.executeNonQueryStatement("ALTER TABLE readings ADD COLUMN extra INT64 FIELD"); + } + if (time == 4) { session.executeNonQueryStatement("ALTER TABLE readings ADD COLUMN tag2 STRING TAG"); } session.executePreparedInsertNumericRow( @@ -220,7 +225,7 @@ public class IoTDBTableModelSessionIT { session.executeNonQueryStatement("FLUSH"); try (SessionDataSet dataSet = session.executeQueryStatement("SELECT time,tag,tag2,v FROM readings ORDER BY time")) { - for (int time = 1; time <= 3; time++) { + for (int time = 1; time <= 5; time++) { assertTrue(dataSet.hasNext()); RowRecord row = dataSet.next(); assertEquals(time, row.getFields().get(0).getLongV()); @@ -289,6 +294,53 @@ public class IoTDBTableModelSessionIT { } } + // Failed schema validation closes only its handle; fixing the table still requires a new prepare. + @Test + public void testPreparedNumericRowClosesOnSchemaFailure() throws Exception { + try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) { + session.executeNonQueryStatement("CREATE DATABASE numeric_invalid"); + session.executeNonQueryStatement("USE numeric_invalid"); + session.executeNonQueryStatement("CREATE TABLE readings(tag STRING TAG, v DOUBLE FIELD)"); + List<String> names = Arrays.asList("tag", "v"); + List<TSDataType> types = Arrays.asList(TSDataType.STRING, TSDataType.DOUBLE); + List<ColumnCategory> categories = Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD); + long[] values = {Double.doubleToRawLongBits(2.5)}; + String[] tags = {"device"}; + try (NumericRowPreparedStatement prepared = + session.prepareInsertNumericRow("readings", names, types, categories); + NumericRowPreparedStatement other = + session.prepareInsertNumericRow("readings", names, types, categories)) { + session.executeNonQueryStatement("DROP TABLE readings"); + session.executeNonQueryStatement("CREATE TABLE readings(tag STRING TAG, v BOOLEAN FIELD)"); + StatementExecutionException schemaError = + assertThrows( + StatementExecutionException.class, + () -> session.executePreparedInsertNumericRow(prepared, 1, values, tags)); + assertTrue(schemaError.getMessage().contains("v")); + session.executeNonQueryStatement("DROP TABLE readings"); + session.executeNonQueryStatement("CREATE TABLE readings(tag STRING TAG, v DOUBLE FIELD)"); + StatementExecutionException closedError = + assertThrows( + StatementExecutionException.class, + () -> session.executePreparedInsertNumericRow(prepared, 2, values, tags)); + assertTrue(closedError.getMessage().contains("closed")); + session.executePreparedInsertNumericRow(other, 3, values, tags); + try (NumericRowPreparedStatement replacement = + session.prepareInsertNumericRow("readings", names, types, categories)) { + session.executePreparedInsertNumericRow(replacement, 4, values, tags); + } + try (SessionDataSet dataSet = + session.executeQueryStatement("SELECT v FROM readings ORDER BY time")) { + for (int i = 0; i < 2; i++) { + assertTrue(dataSet.hasNext()); + assertEquals(2.5, dataSet.next().getFields().get(0).getDoubleV(), 0); + } + assertFalse(dataSet.hasNext()); + } + } + } + } + private void assertNumericNullRows(ITableSession session) throws Exception { try (SessionDataSet dataSet = session.executeQueryStatement( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfo.java index 938afbdc852..7d3003276ee 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfo.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/PreparedNumericRowInfo.java @@ -60,7 +60,7 @@ public final class PreparedNumericRowInfo { private final TsTableColumnCategory[] categories; private final int tagCount; private final int valueBytes; - private final Pair<Long, Long> tableVersion; + private volatile Pair<Long, Long> tableVersion; private final MeasurementSchema[] measurementSchemas; public PreparedNumericRowInfo(String database, TSPrepareNumericRowReq req) { @@ -178,20 +178,13 @@ public final class PreparedNumericRowInfo { } private MeasurementSchema[] prepareMeasurementSchemas(TsTable currentTable) { - // Views and requests needing TAG padding/reordering or type conversion use normal analysis. - if (currentTable == null - || currentTable.getClass() != TsTable.class - || DataNodeTreeViewSchemaUtils.isTreeViewTable(currentTable) - || currentTable.getTagColumnSchemaList().size() != tagCount) { + if (!matchesTableSchema(currentTable)) { return null; } MeasurementSchema[] schemas = new MeasurementSchema[measurements.length]; for (int i = 0; i < measurements.length; i++) { TsTableColumnSchema column = currentTable.getColumnSchema(measurements[i]); - if (column == null - || column.getDataType() != types[i] - || column.getColumnCategory() != categories[i] - || (i < tagCount && currentTable.getTagColumnOrdinal(measurements[i]) != i)) { + if (column == null) { return null; } schemas[i] = @@ -204,6 +197,41 @@ public final class PreparedNumericRowInfo { return schemas; } + private boolean matchesTableSchema(TsTable currentTable) { + // Views and requests needing TAG padding/reordering or type conversion use normal analysis. + if (currentTable == null + || currentTable.getClass() != TsTable.class + || DataNodeTreeViewSchemaUtils.isTreeViewTable(currentTable) + || currentTable.getTagColumnSchemaList().size() != tagCount) { + return false; + } + for (int i = 0; i < measurements.length; i++) { + TsTableColumnSchema column = currentTable.getColumnSchema(measurements[i]); + if (column == null + || column.getDataType() != types[i] + || column.getColumnCategory() != categories[i] + || (i < tagCount && currentTable.getTagColumnOrdinal(measurements[i]) != i)) { + return false; + } + } + return true; + } + + /** Returns a stable compatible version, retaining the cached schemas and wire layout. */ + public Pair<Long, Long> refreshTableVersionIfCompatible(TsTable currentTable) { + if (measurementSchemas == null || currentTable == null) { + return null; + } + Pair<Long, Long> version = currentTable.getInstanceVersion(); + if (!matchesTableSchema(currentTable) || !currentTable.matchesInstanceVersion(version)) { + return null; + } + // Publish only a compatible candidate. Each execution rechecks it under the schema lock; + // retaining the old candidate on failure also keeps the session's memory reservation stable. + tableVersion = version; + return version; + } + public InsertNumericRowStatement bind(long timestamp, ByteBuffer tags, ByteBuffer values) { return bind(timestamp, tags, values, null); } @@ -287,6 +315,7 @@ public final class PreparedNumericRowInfo { statement.setTime(timestamp); statement.setNumericValues(row); if (measurementSchemas != null) { + statement.setPreparedNumericRowInfo(this); statement.setPreparedTableVersion(tableVersion); statement.setMeasurementSchemas(measurementSchemas.clone()); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java index f74ed0b8a58..7435e901d2f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java @@ -115,6 +115,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering; 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.InsertRowsOfOneDeviceStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; @@ -2612,9 +2613,10 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { public TSStatus executePreparedNumericRow(TSExecutePreparedNumericRowReq req) { IClientSession session = SESSION_MANAGER.getCurrSessionAndUpdateIdleTime(); if (!SESSION_MANAGER.checkLogin(session)) return getNotLoggedInStatus(); + InsertNumericRowStatement statement = null; try { if (req.getSessionId() != session.getId()) return RpcUtils.getStatus(TSStatusCode.NOT_LOGIN); - InsertBaseStatement statement = + statement = session .getPreparedNumericRow(req.getPreparedId()) .bind( @@ -2636,6 +2638,11 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { } catch (Exception e) { return onQueryException( e, OperationType.INSERT_RECORD.getName(), TSStatusCode.EXECUTE_STATEMENT_ERROR); + } finally { + if (statement != null && statement.isTableSchemaValidationFailed()) { + // Remove only this session's invalid handle, releasing its reservation exactly once. + session.removePreparedNumericRow(req.getPreparedId()); + } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertNumericRow.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertNumericRow.java index 4510973f753..8145485a96e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertNumericRow.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/InsertNumericRow.java @@ -28,6 +28,8 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertNumericRowStatement; import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.tsfile.utils.Pair; + import java.util.Arrays; import java.util.Collections; import java.util.List; @@ -50,6 +52,19 @@ public class InsertNumericRow extends WrappedInsertStatement { @Override public void validateTableSchema(Metadata metadata, MPPQueryContext context) { + try { + validatePreparedTableSchema(metadata, context); + if (getInnerTreeStatement().hasFailedMeasurements()) { + getInnerTreeStatement().markTableSchemaValidationFailed(); + } + } catch (RuntimeException e) { + // The coordinator may convert exceptions to statuses; retain the failure for RPC cleanup. + getInnerTreeStatement().markTableSchemaValidationFailed(); + throw e; + } + } + + private void validatePreparedTableSchema(Metadata metadata, MPPQueryContext context) { InsertNumericRowStatement statement = getInnerTreeStatement(); if (statement.getPreparedTableVersion() != null) { // Check at analysis time under the same deletion lock and metadata lease as normal inserts. @@ -57,8 +72,18 @@ public class InsertNumericRow extends WrappedInsertStatement { .takeReadLock(context, SchemaLockType.VALIDATE_VS_DELETION_TABLE); TsTable currentTable = DataNodeTableCache.getInstance().getTableInWrite(getDatabase(), getTableName()); - if (currentTable != null - && currentTable.matchesInstanceVersion(statement.getPreparedTableVersion())) { + boolean compatible = + currentTable != null + && currentTable.matchesInstanceVersion(statement.getPreparedTableVersion()); + if (!compatible && statement.getPreparedNumericRowInfo() != null) { + Pair<Long, Long> version = + statement.getPreparedNumericRowInfo().refreshTableVersionIfCompatible(currentTable); + if (version != null) { + statement.setPreparedTableVersion(version); + compatible = true; + } + } + if (compatible) { statement.setAttributeColumnsPresent(false); // Cached schema names must not be mutated by later normalization. statement.setToLowerCaseApplied(true); 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 98e255de911..62970448eb4 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 @@ -29,6 +29,7 @@ import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.commons.utils.TimePartitionUtils; import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.pipe.resource.memory.InsertNodeMemoryEstimator; +import org.apache.iotdb.db.protocol.session.PreparedNumericRowInfo; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaValidation; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.InsertNumericRow; @@ -58,6 +59,8 @@ public class InsertNumericRowStatement extends InsertBaseStatement { private IDeviceID deviceID; // A candidate snapshot, checked again under the schema analysis lock before reuse. private Pair<Long, Long> preparedTableVersion; + private PreparedNumericRowInfo preparedNumericRowInfo; + private volatile boolean tableSchemaValidationFailed; public InsertNumericRowStatement() { statementType = StatementType.INSERT_NUMERIC_ROW; @@ -73,6 +76,22 @@ public class InsertNumericRowStatement extends InsertBaseStatement { this.preparedTableVersion = preparedTableVersion; } + public PreparedNumericRowInfo getPreparedNumericRowInfo() { + return preparedNumericRowInfo; + } + + public void setPreparedNumericRowInfo(PreparedNumericRowInfo preparedNumericRowInfo) { + this.preparedNumericRowInfo = preparedNumericRowInfo; + } + + public boolean isTableSchemaValidationFailed() { + return tableSchemaValidationFailed; + } + + public void markTableSchemaValidationFailed() { + tableSchemaValidationFailed = true; + } + public long getTime() { return time; } 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 e8e6ef31153..45cb991615f 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 @@ -65,6 +65,7 @@ import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -142,20 +143,88 @@ public class PreparedNumericRowInfoTest { assertArrayEquals(new String[] {"tag", "v"}, bind(info, 1).getMeasurements()); } - // A replacement table with the same name/revision must not reuse the old instance's schema. + // A replacement instance must pass compatibility checks even when its revision is unchanged. @Test - public void testRecreatedTableUsesFullValidation() { - InsertNumericRowStatement statement = - bind(new PreparedNumericRowInfo(DATABASE, request("tag", "v")), 1); + public void testCompatibleRecreatedTableRefreshesVersion() { + PreparedNumericRowInfo info = new PreparedNumericRowInfo(DATABASE, request("tag", "v")); + InsertNumericRowStatement statement = bind(info, 1); TsTable replacement = new TsTable("t"); replacement.addColumnSchema(new TagColumnSchema("tag", TSDataType.STRING)); replacement.addColumnSchema(new FieldColumnSchema("v", TSDataType.DOUBLE)); assertEquals(table.getInstanceVersion().right, replacement.getInstanceVersion().right); install(replacement); - validateFallback(statement); + new InsertNumericRow(statement, context).validateTableSchema(metadata, context); + assertEquals(replacement.getInstanceVersion(), statement.getPreparedTableVersion()); + assertEquals(replacement.getInstanceVersion(), bind(info, 1).getPreparedTableVersion()); + verify(metadata, never()) + .validateInsertNodeMeasurements(anyString(), any(), any(), anyBoolean(), any(), any()); assertNotNull(statement.getMeasurementSchemas()[1]); } + // An unrelated FIELD addition refreshes only the version, preserving schemas and memory usage. + @Test + public void testCompatibleVersionChangeRetainsFastPath() { + PreparedNumericRowInfo info = new PreparedNumericRowInfo(DATABASE, request("tag", "v")); + long reservedBytes = info.ramBytesUsed(); + InsertNumericRowStatement statement = bind(info, 1); + Object schemas = statement.getMeasurementSchemas(); + table.addColumnSchema(new FieldColumnSchema("extra", TSDataType.INT64)); + new InsertNumericRow(statement, context).validateTableSchema(metadata, context); + assertSame(schemas, statement.getMeasurementSchemas()); + assertFalse(statement.isTableSchemaValidationFailed()); + assertEquals(table.getInstanceVersion(), statement.getPreparedTableVersion()); + InsertNumericRowStatement next = bind(info, 1); + assertEquals(table.getInstanceVersion(), next.getPreparedTableVersion()); + assertSame(statement.getMeasurementSchemas()[1], next.getMeasurementSchemas()[1]); + new InsertNumericRow(next, context).validateTableSchema(metadata, context); + verify(metadata, never()) + .validateInsertNodeMeasurements(anyString(), any(), any(), anyBoolean(), any(), any()); + assertEquals(reservedBytes, info.ramBytesUsed()); + } + + // Changed FIELD types must use normal validation and preserve its concrete error. + @Test + public void testIncompatibleVersionUsesValidationError() { + PreparedNumericRowInfo info = new PreparedNumericRowInfo(DATABASE, request("tag", "v")); + InsertNumericRowStatement statement = bind(info, 1); + TsTable replacement = new TsTable("t"); + replacement.addColumnSchema(new TagColumnSchema("tag", TSDataType.STRING)); + replacement.addColumnSchema(new FieldColumnSchema("v", TSDataType.BOOLEAN)); + install(replacement); + SemanticException failure = new SemanticException("column v type mismatch"); + doThrow(failure) + .when(metadata) + .validateInsertNodeMeasurements(anyString(), any(), any(), anyBoolean(), any(), any()); + assertSame( + failure, + assertThrows( + SemanticException.class, + () -> new InsertNumericRow(statement, context).validateTableSchema(metadata, context))); + verify(metadata) + .validateInsertNodeMeasurements(eq(DATABASE), any(), eq(context), eq(true), any(), any()); + assertNull(statement.getPreparedTableVersion()); + assertNull(statement.getMeasurementSchemas()); + assertTrue(statement.isTableSchemaValidationFailed()); + assertFalse(replacement.matchesInstanceVersion(bind(info, 1).getPreparedTableVersion())); + } + + // Partial insert errors also invalidate the handle even when validation returns normally. + @Test + public void testPartialSchemaFailureMarksPreparedRowInvalid() { + PreparedNumericRowInfo info = new PreparedNumericRowInfo(DATABASE, request("tag", "v")); + InsertNumericRowStatement statement = bind(info, 1); + table.addColumnSchema(new TagColumnSchema("tag2", TSDataType.STRING)); + doAnswer( + invocation -> { + statement.markFailedMeasurement(1, new SemanticException("column v rejected")); + return null; + }) + .when(metadata) + .validateInsertNodeMeasurements(anyString(), any(), any(), anyBoolean(), any(), any()); + validateFallback(statement); + assertTrue(statement.isTableSchemaValidationFailed()); + } + // Missing columns/table, views and type conversions remain on normal creation/validation. @Test public void testIneligibleMetadataHasNoCachedSchema() {
