From 047ef89af44c0ba5ad84aedbb98afbdaf750373b Mon Sep 17 00:00:00 2001 From: 761417898 <761417898@qq.com> Date: Thu, 15 Oct 2026 11:18:45 +0800 Subject: [PATCH 1/4] [Fix] Preserve table insert errors for rejected fields --- .../org/apache/iotdb/session/Session.java | 7 +- .../iotdb/session/SessionCacheLeaderTest.java | 79 ++++++++++++++++++- .../plan/analyze/AnalyzeUtils.java | 8 ++ .../plan/analyze/AnalyzeUtilsTest.java | 53 +++++++++++++ 4 files changed, 145 insertions(+), 2 deletions(-) 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 fe9fb38f6dbbe..6d31f81885a47 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 @@ -3007,6 +3007,7 @@ private void insertRelationalTabletByGroup(Map relati .collect(Collectors.toList()); StringBuilder errMsgBuilder = new StringBuilder(); + List errors = new ArrayList<>(); for (CompletableFuture completableFuture : completableFutures) { try { completableFuture.join(); @@ -3016,6 +3017,7 @@ private void insertRelationalTabletByGroup(Map relati if (cause instanceof IoTDBConnectionException) { throw (IoTDBConnectionException) cause; } else { + errors.add(cause); if (errMsgBuilder.length() > 0) { errMsgBuilder.append(";"); } @@ -3024,7 +3026,10 @@ private void insertRelationalTabletByGroup(Map relati } } if (errMsgBuilder.length() > 0) { - throw new StatementExecutionException(errMsgBuilder.toString()); + StatementExecutionException exception = + new StatementExecutionException(errMsgBuilder.toString()); + errors.forEach(exception::addSuppressed); + throw exception; } } diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionCacheLeaderTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionCacheLeaderTest.java index 40268a54cd340..46ec3ee1f5d21 100644 --- a/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionCacheLeaderTest.java +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/SessionCacheLeaderTest.java @@ -20,10 +20,12 @@ package org.apache.iotdb.session; import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.rpc.RedirectException; import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq; import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsOfOneDeviceReq; import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsReq; @@ -44,10 +46,12 @@ import java.time.ZoneId; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Random; +import java.util.concurrent.atomic.AtomicBoolean; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; @@ -705,6 +709,79 @@ public void testInsertRelationalTablet() session.close(); } + @Test + public void testGroupedRelationalTabletPreservesStatementFailures() throws Exception { + AtomicBoolean failWrites = new AtomicBoolean(); + session = + new MockSession("127.0.0.1", 55560, true, "table") { + @Override + public SessionConnection constructSessionConnection( + Session owner, TEndPoint endpoint, ZoneId zoneId) { + return new MockSessionConnection(owner, endpoint, zoneId) { + @Override + protected void insertTablet(TSInsertTabletReq request) + throws RedirectException, IoTDBConnectionException, StatementExecutionException { + if (failWrites.get()) { + TSStatusCode code = + endpoint.getPort() == 55560 + ? TSStatusCode.METADATA_ERROR + : TSStatusCode.DATA_TYPE_MISMATCH; + throw new StatementExecutionException( + new TSStatus(code.getStatusCode()).setMessage("incompatible field")); + } + super.insertTablet(request); + } + }; + } + }; + session.open(); + try { + Tablet tablet = + new Tablet( + "table1", + Arrays.asList("id", "field"), + Arrays.asList(TSDataType.STRING, TSDataType.INT64), + Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD), + 50); + for (int i = 0; i < 50; i++) { + int row = tablet.getRowSize(); + tablet.addTimestamp(row, row); + tablet.addValue("id", row, "id" + (row % 4)); + tablet.addValue("field", row, 1L); + } + session.insertRelationalTablet(tablet); + assertEquals(4, session.tableModelDeviceIdToEndpoint.size()); + + failWrites.set(true); + tablet.reset(); + for (int i = 0; i < 4; i++) { + int row = tablet.getRowSize(); + tablet.addTimestamp(row, row); + tablet.addValue("id", row, "id" + row); + tablet.addValue("field", row, 1L); + } + StatementExecutionException failure = + Assert.assertThrows( + StatementExecutionException.class, () -> session.insertRelationalTablet(tablet)); + assertEquals(4, failure.getSuppressed().length); + long metadataFailures = 0; + long typeFailures = 0; + for (Throwable cause : failure.getSuppressed()) { + Assert.assertTrue(cause instanceof StatementExecutionException); + int code = ((StatementExecutionException) cause).getStatusCode(); + if (code == TSStatusCode.METADATA_ERROR.getStatusCode()) { + metadataFailures++; + } else if (code == TSStatusCode.DATA_TYPE_MISMATCH.getStatusCode()) { + typeFailures++; + } + } + assertEquals(1, metadataFailures); + assertEquals(3, typeFailures); + } finally { + session.close(); + } + } + @Test public void testInsertRecordsWithSessionBroken() throws StatementExecutionException { // without leader cache @@ -1281,7 +1358,7 @@ protected void insertRecordsOfOneDevice(TSInsertRecordsOfOneDeviceReq request) @Override protected void insertTablet(TSInsertTabletReq request) - throws RedirectException, IoTDBConnectionException { + throws RedirectException, IoTDBConnectionException, StatementExecutionException { if (isConnectionBroken()) { throw ioTDBConnectionException; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java index 5da7306656778..4b65b3489cf83 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java @@ -173,6 +173,14 @@ public static InsertBaseStatement analyzeInsert( return realStatement; } analysis.setRealStatement(realStatement); + // A failed row or tablet may retain TAG columns after every FIELD column was rejected. + if (analysis.isFailed() + && (realStatement instanceof InsertRowStatement + || realStatement instanceof InsertTabletStatement) + && !realStatement.hasValidMeasurements()) { + analysis.setFinishQueryAfterAnalyze(true); + return realStatement; + } analyzeDataPartition( analysis, diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java index 697e15cd8f41a..47cfa35284687 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java @@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.ComparisonExpression; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Identifier; @@ -32,18 +33,23 @@ import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Table; import org.apache.iotdb.commons.schema.table.TsTable; import org.apache.iotdb.commons.schema.table.column.TimeColumnSchema; +import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeReq; import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeResp; import org.apache.iotdb.db.protocol.client.ConfigNodeClient; import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Delete; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.read.common.TimeRange; +import org.apache.tsfile.write.schema.MeasurementSchema; import org.junit.Test; import org.mockito.ArgumentCaptor; import org.mockito.Mockito; @@ -60,6 +66,53 @@ public class AnalyzeUtilsTest { + @Test + public void testRejectedOnlyFieldStopsRowAndTabletBeforePartitionLookup() throws Exception { + final InsertRowStatement row = new InsertRowStatement(); + row.setDevicePath(new PartialPath("root.sg.d1")); + row.setMeasurements(new String[] {"tag", "field"}); + row.setDataTypes(new TSDataType[] {TSDataType.STRING, TSDataType.INT64}); + row.setValues(new Object[] {"device", 1L}); + row.setTime(1L); + assertRejectedFieldStopsBeforePartitionLookup(row); + + final InsertTabletStatement tablet = new InsertTabletStatement(); + tablet.setDevicePath(new PartialPath("root.sg.d1")); + tablet.setMeasurements(new String[] {"tag", "field"}); + tablet.setDataTypes(new TSDataType[] {TSDataType.STRING, TSDataType.INT64}); + tablet.setColumns(new Object[] {new String[] {"device"}, new long[] {1L}}); + tablet.setTimes(new long[] {1L}); + tablet.setRowCount(1); + assertRejectedFieldStopsBeforePartitionLookup(tablet); + } + + private static void assertRejectedFieldStopsBeforePartitionLookup(InsertBaseStatement statement) { + statement.setColumnCategories( + new TsTableColumnCategory[] {TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD}); + statement.setMeasurementSchemas( + new MeasurementSchema[] { + new MeasurementSchema("tag", TSDataType.STRING), + new MeasurementSchema("field", TSDataType.INT64) + }); + final Analysis analysis = new Analysis(); + final MPPQueryContext context = new MPPQueryContext(new QueryId("rejected_field")); + + AnalyzeUtils.analyzeInsert( + context, + statement, + () -> statement.markFailedMeasurement(1, new IllegalArgumentException("type mismatch")), + (params, userName) -> { + throw new AssertionError("Partition lookup must not run after every field fails"); + }, + AnalyzeUtils::computeTableDataPartitionParams, + analysis, + false); + + assertTrue(analysis.isFinishQueryAfterAnalyze()); + assertEquals(TSStatusCode.METADATA_ERROR.getStatusCode(), analysis.getFailStatus().getCode()); + assertTrue(analysis.getFailStatus().getMessage().contains("field")); + } + @Test public void testParseDeletePredicateWithRenamedTimeColumn() { TsTable table = new TsTable("table1"); From 8d57461c8cc98c89b5cf3e422867eb6faad91350 Mon Sep 17 00:00:00 2001 From: 761417898 <761417898@qq.com> Date: Thu, 15 Oct 2026 11:38:52 +0800 Subject: [PATCH 2/4] [Fix] Retain metadata errors for fully rejected table row batches --- .../plan/analyze/AnalyzeUtils.java | 17 ++-- .../plan/analyze/AnalyzeUtilsTest.java | 85 +++++++++++++++++++ 2 files changed, 97 insertions(+), 5 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java index 4b65b3489cf83..4ec0a39c08bfe 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java @@ -173,11 +173,8 @@ public static InsertBaseStatement analyzeInsert( return realStatement; } analysis.setRealStatement(realStatement); - // A failed row or tablet may retain TAG columns after every FIELD column was rejected. - if (analysis.isFailed() - && (realStatement instanceof InsertRowStatement - || realStatement instanceof InsertTabletStatement) - && !realStatement.hasValidMeasurements()) { + // A failed insert may retain TAG columns after every FIELD column was rejected. + if (analysis.isFailed() && hasNoValidMeasurements(realStatement)) { analysis.setFinishQueryAfterAnalyze(true); return realStatement; } @@ -190,6 +187,16 @@ public static InsertBaseStatement analyzeInsert( return realStatement; } + private static boolean hasNoValidMeasurements(InsertBaseStatement statement) { + if (statement instanceof InsertRowsStatement) { + return ((InsertRowsStatement) statement) + .getInsertRowStatementList().stream() + .noneMatch(InsertBaseStatement::hasValidMeasurements); + } + return (statement instanceof InsertRowStatement || statement instanceof InsertTabletStatement) + && !statement.hasValidMeasurements(); + } + public static String getDatabaseName( final InsertBaseStatement statement, final MPPQueryContext context) { if (statement.getDatabaseName().isPresent()) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java index 47cfa35284687..d1d347d71ae12 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java @@ -24,7 +24,9 @@ import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.partition.DataPartition; import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.common.SessionInfo; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.ComparisonExpression; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression; import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Identifier; @@ -42,6 +44,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Delete; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertBaseStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement; +import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement; import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; @@ -54,13 +57,17 @@ import org.mockito.ArgumentCaptor; import org.mockito.Mockito; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; @@ -113,6 +120,84 @@ private static void assertRejectedFieldStopsBeforePartitionLookup(InsertBaseStat assertTrue(analysis.getFailStatus().getMessage().contains("field")); } + @Test + public void testRejectedFieldsInEveryRowStopBeforePartitionLookup() throws Exception { + final InsertRowsStatement rows = createRowsWithRejectedFields(2); + final Analysis analysis = new Analysis(); + final MPPQueryContext context = new MPPQueryContext(new QueryId("all_rows_rejected")); + context.setSession(Mockito.mock(SessionInfo.class)); + + AnalyzeUtils.analyzeInsert( + context, + rows, + () -> {}, + (params, userName) -> { + throw new AssertionError("Partition lookup must not run after every field fails"); + }, + AnalyzeUtils::computeTableDataPartitionParams, + analysis, + false); + + assertTrue(analysis.isFinishQueryAfterAnalyze()); + assertEquals(TSStatusCode.METADATA_ERROR.getStatusCode(), analysis.getFailStatus().getCode()); + assertTrue(analysis.getFailStatus().getMessage().contains("field")); + } + + @Test + public void testValidFieldInAnotherRowContinuesPartitionLookup() throws Exception { + final InsertRowsStatement rows = createRowsWithRejectedFields(1); + final Analysis analysis = new Analysis(); + final MPPQueryContext context = new MPPQueryContext(new QueryId("one_valid_row")); + context.setSession(Mockito.mock(SessionInfo.class)); + final DataPartition partition = Mockito.mock(DataPartition.class); + final AtomicBoolean partitionQueried = new AtomicBoolean(); + + AnalyzeUtils.analyzeInsert( + context, + rows, + () -> {}, + (params, userName) -> { + partitionQueried.set(true); + return partition; + }, + AnalyzeUtils::computeTableDataPartitionParams, + analysis, + false); + + assertTrue(partitionQueried.get()); + assertFalse(analysis.isFinishQueryAfterAnalyze()); + assertEquals(TSStatusCode.METADATA_ERROR.getStatusCode(), analysis.getFailStatus().getCode()); + assertSame(partition, analysis.getDataPartitionInfo()); + } + + private static InsertRowsStatement createRowsWithRejectedFields(int rejectedRows) + throws Exception { + final List rowStatements = new ArrayList<>(); + for (int i = 0; i < 2; i++) { + final InsertRowStatement row = new InsertRowStatement(); + row.setDevicePath(new PartialPath("root.sg.d1")); + row.setMeasurements(new String[] {"tag", "field"}); + row.setDataTypes(new TSDataType[] {TSDataType.STRING, TSDataType.INT64}); + row.setColumnCategories( + new TsTableColumnCategory[] {TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD}); + row.setMeasurementSchemas( + new MeasurementSchema[] { + new MeasurementSchema("tag", TSDataType.STRING), + new MeasurementSchema("field", TSDataType.INT64) + }); + row.setValues(new Object[] {"device" + i, 1L}); + row.setTime(i); + if (i < rejectedRows) { + row.markFailedMeasurement(1, new IllegalArgumentException("type mismatch")); + } + rowStatements.add(row); + } + final InsertRowsStatement rows = new InsertRowsStatement(); + rows.setInsertRowStatementList(rowStatements); + rows.updateAfterSchemaValidation(null); + return rows; + } + @Test public void testParseDeletePredicateWithRenamedTimeColumn() { TsTable table = new TsTable("table1"); From 9a7fb10ff705722d54cedd5ff6ea440f80970c4a Mon Sep 17 00:00:00 2001 From: 761417898 <761417898@qq.com> Date: Thu, 15 Oct 2026 16:05:16 +0800 Subject: [PATCH 3/4] [Fix] Preserve table timestamps when all fields are rejected --- .../it/db/it/IoTDBInsertTableIT.java | 62 +++++++++++++++++++ .../plan/analyze/AnalyzeUtils.java | 15 ----- .../planner/plan/node/write/InsertNode.java | 5 ++ .../storageengine/dataregion/DataRegion.java | 8 +-- .../dataregion/memtable/AbstractMemTable.java | 4 +- .../wal/recover/file/TsFilePlanRedoer.java | 4 +- .../plan/analyze/AnalyzeUtilsTest.java | 29 ++++++--- .../recover/file/TsFilePlanRedoerTest.java | 52 ++++++++++++++++ 8 files changed, 147 insertions(+), 32 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java index 3abe7497130a5..70eab82727cfb 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBInsertTableIT.java @@ -1379,6 +1379,68 @@ public void testInsertAllNullRow() throws SQLException { } } + @Test + public void testRejectedOnlyFieldStillWritesTime() throws Exception { + try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) { + session.executeNonQueryStatement("use \"test\""); + session.executeNonQueryStatement( + "create table rejected_only_field_time (tag1 string tag, s1 int64 field)"); + + Tablet tablet = + new Tablet( + "rejected_only_field_time", + Arrays.asList("tag1", "s1"), + Arrays.asList(TSDataType.STRING, TSDataType.BOOLEAN), + Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD), + 2); + for (int i = 0; i < 2; i++) { + int row = tablet.getRowSize(); + tablet.addTimestamp(row, i + 1); + tablet.addValue("tag1", row, "d" + i); + tablet.addValue("s1", row, true); + } + + StatementExecutionException error = + assertThrows(StatementExecutionException.class, () -> session.insert(tablet)); + assertTrue(error.getMessage().contains("s1")); + + Tablet withoutTag = + new Tablet( + "rejected_only_field_time", + Arrays.asList("s1"), + Arrays.asList(TSDataType.BOOLEAN), + Arrays.asList(ColumnCategory.FIELD), + 1); + withoutTag.addTimestamp(0, 3); + withoutTag.addValue("s1", 0, true); + error = assertThrows(StatementExecutionException.class, () -> session.insert(withoutTag)); + assertTrue(error.getMessage().contains("s1")); + assertRejectedFieldTimeRows(session); + + session.executeNonQueryStatement("flush"); + assertRejectedFieldTimeRows(session); + } + } + + private static void assertRejectedFieldTimeRows(ITableSession session) + throws IoTDBConnectionException, StatementExecutionException { + SessionDataSet dataSet = + session.executeQueryStatement( + "select time, tag1, s1 from rejected_only_field_time order by time"); + for (int i = 0; i < 3; i++) { + assertTrue(dataSet.hasNext()); + RowRecord row = dataSet.next(); + assertEquals(i + 1, row.getFields().get(0).getLongV()); + if (i < 2) { + assertEquals("d" + i, row.getFields().get(1).getStringValue()); + } else { + assertNull(row.getFields().get(1).getObjectValue(TSDataType.STRING)); + } + assertNull(row.getFields().get(2).getObjectValue(TSDataType.INT64)); + } + assertFalse(dataSet.hasNext()); + } + @Test public void testInsertWithChangedSchema() throws SQLException, IOException { SimpleEnv simpleEnv = new SimpleEnv(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java index 4ec0a39c08bfe..5da7306656778 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java @@ -173,11 +173,6 @@ public static InsertBaseStatement analyzeInsert( return realStatement; } analysis.setRealStatement(realStatement); - // A failed insert may retain TAG columns after every FIELD column was rejected. - if (analysis.isFailed() && hasNoValidMeasurements(realStatement)) { - analysis.setFinishQueryAfterAnalyze(true); - return realStatement; - } analyzeDataPartition( analysis, @@ -187,16 +182,6 @@ public static InsertBaseStatement analyzeInsert( return realStatement; } - private static boolean hasNoValidMeasurements(InsertBaseStatement statement) { - if (statement instanceof InsertRowsStatement) { - return ((InsertRowsStatement) statement) - .getInsertRowStatementList().stream() - .noneMatch(InsertBaseStatement::hasValidMeasurements); - } - return (statement instanceof InsertRowStatement || statement instanceof InsertTabletStatement) - && !statement.hasValidMeasurements(); - } - public static String getDatabaseName( final InsertBaseStatement statement, final MPPQueryContext context) { if (statement.getDatabaseName().isPresent()) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java index 289866c6882f2..0f2bc190120f2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java @@ -446,6 +446,11 @@ public boolean allMeasurementFailed() { return measurements == null || !hasValidMeasurements(); } + public boolean shouldSkipWrite() { + // Table rows remain writable through their time column after every FIELD is rejected. + return allMeasurementFailed() && columnCategories == null; + } + // endregion // region progress index diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 46952ee87fabc..b312f563e4da4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -1551,7 +1551,7 @@ private int findLastInsertTabletIndexToMark(final InsertMultiTabletsNode insertM for (int i = insertMultiTabletsNode.getInsertTabletNodeList().size() - 1; i >= 0; i--) { final InsertTabletNode insertTabletNode = insertMultiTabletsNode.getInsertTabletNodeList().get(i); - if (insertTabletNode.getRowCount() <= 0 || insertTabletNode.allMeasurementFailed()) { + if (insertTabletNode.getRowCount() <= 0 || insertTabletNode.shouldSkipWrite()) { continue; } if (!insertTabletNode.shouldCheckTTL()) { @@ -1601,7 +1601,7 @@ private boolean insertTabletToTsFileProcessor( boolean noFailure, long[] infoForMetrics) throws DataTypeInconsistentException { - if (insertTabletNode.allMeasurementFailed()) { + if (insertTabletNode.shouldSkipWrite()) { if (logger.isDebugEnabled()) { logger.debug( StorageEngineMessages.STORAGE_LOG_WON_T_INSERT_TABLET_BECAUSE_C2DC8032, @@ -1809,7 +1809,7 @@ private void tryToUpdateInsertTabletLastCache( private TsFileProcessor insertToTsFileProcessor( InsertRowNode insertRowNode, boolean sequence, long timePartitionId) throws WriteProcessException { - if (insertRowNode.allMeasurementFailed()) { + if (insertRowNode.shouldSkipWrite()) { return null; } TsFileProcessor tsFileProcessor = getOrCreateTsFileProcessor(timePartitionId, sequence); @@ -1838,7 +1838,7 @@ private List insertToTsFileProcessors( Map tsFileProcessorMap = new HashMap<>(); for (int i = 0; i < areSequence.length; i++) { InsertRowNode insertRowNode = insertRowsNode.getInsertRowNodeList().get(i); - if (insertRowNode.allMeasurementFailed()) { + if (insertRowNode.shouldSkipWrite()) { continue; } final TsFileProcessor tsFileProcessor; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java index 5fd3d5f9224a6..382063359c186 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java @@ -266,7 +266,7 @@ public int insertAlignedRow(InsertRowNode insertRowNode) { dataTypes.add(schema.getType()); writableValues[i] = value; } - if (schemaList.isEmpty()) { + if (schemaList.isEmpty() && insertRowNode.getColumnCategories() == null) { return 0; } memSize += @@ -497,7 +497,7 @@ public void writeAlignedTablet( } } } - if (schemaList.isEmpty()) { + if (schemaList.isEmpty() && insertTabletNode.getColumnCategories() == null) { return; } int splitStart = start; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java index 8e191d475ba5e..c4550898a0529 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java @@ -92,7 +92,7 @@ void redoDelete(RelationalDeleteDataNode node) throws IOException { } void redoInsert(InsertNode node) throws WriteProcessException { - if (!node.hasValidMeasurements()) { + if (node.shouldSkipWrite()) { return; } boolean isSingleDevice = !(node instanceof RelationalInsertTabletNode); @@ -155,7 +155,7 @@ void redoInsert(InsertNode node) throws WriteProcessException { void redoInsertRows(InsertRowsNode insertRowsNode) { int pointsInserted = 0; for (InsertRowNode node : insertRowsNode.getInsertRowNodeList()) { - if (!node.hasValidMeasurements()) { + if (node.shouldSkipWrite()) { continue; } if (tsFileResource != null) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java index d1d347d71ae12..e5a1fb2335519 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtilsTest.java @@ -74,14 +74,14 @@ public class AnalyzeUtilsTest { @Test - public void testRejectedOnlyFieldStopsRowAndTabletBeforePartitionLookup() throws Exception { + public void testRejectedOnlyFieldStillLooksUpRowAndTabletPartition() throws Exception { final InsertRowStatement row = new InsertRowStatement(); row.setDevicePath(new PartialPath("root.sg.d1")); row.setMeasurements(new String[] {"tag", "field"}); row.setDataTypes(new TSDataType[] {TSDataType.STRING, TSDataType.INT64}); row.setValues(new Object[] {"device", 1L}); row.setTime(1L); - assertRejectedFieldStopsBeforePartitionLookup(row); + assertRejectedFieldStillLooksUpPartition(row); final InsertTabletStatement tablet = new InsertTabletStatement(); tablet.setDevicePath(new PartialPath("root.sg.d1")); @@ -90,10 +90,10 @@ public void testRejectedOnlyFieldStopsRowAndTabletBeforePartitionLookup() throws tablet.setColumns(new Object[] {new String[] {"device"}, new long[] {1L}}); tablet.setTimes(new long[] {1L}); tablet.setRowCount(1); - assertRejectedFieldStopsBeforePartitionLookup(tablet); + assertRejectedFieldStillLooksUpPartition(tablet); } - private static void assertRejectedFieldStopsBeforePartitionLookup(InsertBaseStatement statement) { + private static void assertRejectedFieldStillLooksUpPartition(InsertBaseStatement statement) { statement.setColumnCategories( new TsTableColumnCategory[] {TsTableColumnCategory.TAG, TsTableColumnCategory.FIELD}); statement.setMeasurementSchemas( @@ -103,44 +103,55 @@ private static void assertRejectedFieldStopsBeforePartitionLookup(InsertBaseStat }); final Analysis analysis = new Analysis(); final MPPQueryContext context = new MPPQueryContext(new QueryId("rejected_field")); + context.setSession(Mockito.mock(SessionInfo.class)); + final DataPartition partition = Mockito.mock(DataPartition.class); + final AtomicBoolean partitionQueried = new AtomicBoolean(); AnalyzeUtils.analyzeInsert( context, statement, () -> statement.markFailedMeasurement(1, new IllegalArgumentException("type mismatch")), (params, userName) -> { - throw new AssertionError("Partition lookup must not run after every field fails"); + partitionQueried.set(true); + return partition; }, AnalyzeUtils::computeTableDataPartitionParams, analysis, false); - assertTrue(analysis.isFinishQueryAfterAnalyze()); + assertTrue(partitionQueried.get()); + assertFalse(analysis.isFinishQueryAfterAnalyze()); assertEquals(TSStatusCode.METADATA_ERROR.getStatusCode(), analysis.getFailStatus().getCode()); assertTrue(analysis.getFailStatus().getMessage().contains("field")); + assertSame(partition, analysis.getDataPartitionInfo()); } @Test - public void testRejectedFieldsInEveryRowStopBeforePartitionLookup() throws Exception { + public void testRejectedFieldsInEveryRowStillLookUpPartition() throws Exception { final InsertRowsStatement rows = createRowsWithRejectedFields(2); final Analysis analysis = new Analysis(); final MPPQueryContext context = new MPPQueryContext(new QueryId("all_rows_rejected")); context.setSession(Mockito.mock(SessionInfo.class)); + final DataPartition partition = Mockito.mock(DataPartition.class); + final AtomicBoolean partitionQueried = new AtomicBoolean(); AnalyzeUtils.analyzeInsert( context, rows, () -> {}, (params, userName) -> { - throw new AssertionError("Partition lookup must not run after every field fails"); + partitionQueried.set(true); + return partition; }, AnalyzeUtils::computeTableDataPartitionParams, analysis, false); - assertTrue(analysis.isFinishQueryAfterAnalyze()); + assertTrue(partitionQueried.get()); + assertFalse(analysis.isFinishQueryAfterAnalyze()); assertEquals(TSStatusCode.METADATA_ERROR.getStatusCode(), analysis.getFailStatus().getCode()); assertTrue(analysis.getFailStatus().getMessage().contains("field")); + assertSame(partition, analysis.getDataPartitionInfo()); } @Test diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java index 238606b073dfa..a58171017ac1b 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java @@ -23,18 +23,21 @@ import org.apache.iotdb.commons.path.NonAlignedFullPath; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk; import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.TsFileUtilsForRecoverTest; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.tsfile.common.conf.TSFileConfig; @@ -66,6 +69,7 @@ import java.io.File; import java.io.IOException; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -691,6 +695,54 @@ public void testRedoRelationalInsertTablet() Assert.assertFalse(recoveryMemTable.getMemTableMap().containsKey(existDevice)); } + @Test + public void testRedoRelationalInsertWithoutFieldValues() throws Exception { + File file = new File(FILE_NAME); + generateCompleteFile(file); + tsFileResource = new TsFileResource(file); + + RelationalInsertRowNode row = + new RelationalInsertRowNode( + new PlanNodeId("row"), + new PartialPath("table1", false), + true, + new String[] {null}, + new TSDataType[] {null}, + new MeasurementSchema[] {null}, + 101, + new Object[] {null}, + false, + new TsTableColumnCategory[] {TsTableColumnCategory.FIELD}); + RelationalInsertTabletNode tablet = + new RelationalInsertTabletNode( + new PlanNodeId("tablet"), + new PartialPath("table1", false), + true, + new String[] {null}, + new TSDataType[] {null}, + new MeasurementSchema[] {null}, + new long[] {102, 103}, + null, + new Object[] {null}, + 2, + new TsTableColumnCategory[] {TsTableColumnCategory.FIELD}); + + TsFilePlanRedoer redoer = new TsFilePlanRedoer(tsFileResource); + ByteBuffer rowBuffer = ByteBuffer.allocate(row.serializedSize()); + row.serializeToWAL(new WALByteBufferForTest(rowBuffer)); + rowBuffer.flip(); + redoer.redoInsert((RelationalInsertRowNode) PlanNodeType.deserializeFromWAL(rowBuffer)); + + ByteBuffer tabletBuffer = ByteBuffer.allocate(tablet.serializedSize()); + tablet.serializeToWAL(new WALByteBufferForTest(tabletBuffer)); + tabletBuffer.flip(); + redoer.redoInsert((RelationalInsertTabletNode) PlanNodeType.deserializeFromWAL(tabletBuffer)); + + assertEquals(1, redoer.getRecoveryMemTable().getMemTableMap().size()); + assertEquals( + 3, redoer.getRecoveryMemTable().getMemTableMap().values().iterator().next().count(false)); + } + @Test public void testRedoAlignedInsertAfterDeleteTimeseries() throws Exception { // generate .tsfile and update resource in memory From bfd715de89d341860a19eb68625900d2877d6959 Mon Sep 17 00:00:00 2001 From: 761417898 <761417898@qq.com> Date: Thu, 15 Oct 2026 17:54:23 +0800 Subject: [PATCH 4/4] [Test] Align rejected-field assertions and fix controlled flush fixtures --- .../it/session/IoTDBSessionRelationalIT.java | 48 +++++++++++++------ .../memtable/TsFileProcessorTest.java | 9 ++-- 2 files changed, 40 insertions(+), 17 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBSessionRelationalIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBSessionRelationalIT.java index 134a7a2bea0fd..fbe62729c3fd7 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBSessionRelationalIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBSessionRelationalIT.java @@ -437,16 +437,19 @@ public void partialInsertSQLTest() throws IoTDBConnectionException, StatementExe SessionDataSet dataSet = session.executeQueryStatement("select * from partial_insert order by time"); - long[] timestamps = new long[] {10000, 20000, 40000, 50000}; - Boolean[] values = new Boolean[] {true, false, false, true}; + // Partial insertion retains timestamps even when the only FIELD value is rejected. + long[] timestamps = + new long[] {10000, 20000, 30000, 35000, 40000, 50000, 60000, 70000, 80000, 90000}; + Boolean[] values = + new Boolean[] {true, false, null, null, false, true, null, null, null, null}; int cnt = 0; while (dataSet.hasNext()) { RowRecord rec = dataSet.next(); assertEquals(timestamps[cnt], rec.getFields().get(0).getLongV()); - assertEquals(values[cnt], rec.getFields().get(1).getBoolV()); + assertEquals(values[cnt], rec.getFields().get(1).getObjectValue(TSDataType.BOOLEAN)); cnt++; } - assertEquals(4, cnt); + assertEquals(timestamps.length, cnt); } finally { try (ISession session = EnvFactory.getEnv().getSessionConnection()) { @@ -757,14 +760,23 @@ public void insertAllFieldDataTypeMismatchTest() e.getMessage()); } tablet.reset(); - int cnt = 0; - SessionDataSet dataSet = - session.executeQueryStatement("select * from field_wrong_type order by time"); - while (dataSet.hasNext()) { - dataSet.next(); - cnt++; + for (int phase = 0; phase < 2; phase++) { + try (SessionDataSet dataSet = + session.executeQueryStatement("select * from field_wrong_type order by time")) { + for (int row = 0; row < 10; row++) { + assertTrue(dataSet.hasNext()); + RowRecord record = dataSet.next(); + assertEquals(row, record.getFields().get(0).getLongV()); + assertEquals("tag:" + row, record.getFields().get(1).getStringValue()); + assertNull(record.getFields().get(2).getDataType()); + assertNull(record.getFields().get(3).getDataType()); + } + assertFalse(dataSet.hasNext()); + } + if (phase == 0) { + session.executeNonQueryStatement("flush"); + } } - assertEquals(0, cnt); } } @@ -1500,9 +1512,17 @@ private void testOneCastWithTablet( e.getMessage()); } // time, tag1, m1 - SessionDataSet dataSet = - session.executeQueryStatement("select * from table" + testNum + " order by time"); - assertFalse(dataSet.hasNext()); + try (SessionDataSet dataSet = + session.executeQueryStatement("select * from table" + testNum + " order by time")) { + for (int row = 0; row < 2; row++) { + assertTrue(dataSet.hasNext()); + RowRecord record = dataSet.next(); + assertEquals(row, record.getFields().get(0).getLongV()); + assertEquals("d1", record.getFields().get(1).getStringValue()); + assertNull(record.getFields().get(2).getDataType()); + } + assertFalse(dataSet.hasNext()); + } } else { // cannot cast, expect an exception try { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index 324bf78a37e09..eed99aadc5734 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -236,8 +236,9 @@ public void testDeletionDuringFlushCompletionIsPersisted() throws Exception { TSRecord record = new TSRecord(deviceId, 1); record.addTuple(DataPoint.getDataPoint(dataType, measurementId, "1")); processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]); + // Queue an ordinary flush while keeping its execution under test control. processor.setManagedByFlushManager(true); - processor.asyncClose(); + processor.asyncFlush(); Assert.assertNull(processor.getWorkMemTable()); MeasurementPath path = new MeasurementPath(deviceId + "." + measurementId); @@ -284,6 +285,7 @@ public void testDeletionDuringFlushCompletionIsPersisted() throws Exception { } finally { try { if (processor != null) { + processor.asyncCloseForFullFlush().get(); processor.syncClose(); } } finally { @@ -330,10 +332,10 @@ private void testDeletionWALWithoutWorkingMemTable(boolean tableModel) throws Ex long memTableId = processor.getWorkMemTable().getMemTableId(); logDeletionForTest(tableModel, 2); - // Use the production close path, while keeping flush execution under test control. + // Queue an ordinary flush while keeping its execution under test control. processor.setManagedByFlushManager(true); try { - processor.asyncClose(); + processor.asyncFlush(); Assert.assertNull(processor.getWorkMemTable()); assertEquals(1, processor.getFlushingMemTableSize()); processor.flushOneMemTable(); @@ -383,6 +385,7 @@ private void testDeletionWALWithoutWorkingMemTable(boolean tableModel) throws Ex } finally { try { if (processor != null) { + processor.asyncCloseForFullFlush().get(); processor.syncClose(); } } finally {