[Fix] Preserve table insert errors for rejected fields (#18880)

* [Fix] Preserve table insert errors for rejected fields

* [Fix] Retain metadata errors for fully rejected table row batches

* [Fix] Preserve table timestamps when all fields are rejected

* [Test] Align rejected-field assertions and fix controlled flush fixtures

Author: hongzhi-gao

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();

integration-test/src/test/java/org/apache/iotdb/relational/it/session/IoTDBSessionRelationalIT.java

@@ -449,16 +449,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()) {
@@ -769,14 +772,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);
     }
   }
 
@@ -1512,9 +1524,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 {

iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java

@@ -3007,6 +3007,7 @@ private void insertRelationalTabletByGroup(Map<SessionConnection, Tablet> relati
             .collect(Collectors.toList());
 
     StringBuilder errMsgBuilder = new StringBuilder();
+    List<Throwable> errors = new ArrayList<>();
     for (CompletableFuture<Void> completableFuture : completableFutures) {
       try {
         completableFuture.join();
@@ -3016,6 +3017,7 @@ private void insertRelationalTabletByGroup(Map<SessionConnection, Tablet> 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<SessionConnection, Tablet> relati
       }
     }
     if (errMsgBuilder.length() > 0) {
-      throw new StatementExecutionException(errMsgBuilder.toString());
+      StatementExecutionException exception =
+          new StatementExecutionException(errMsgBuilder.toString());
+      errors.forEach(exception::addSuppressed);
+      throw exception;
     }
   }
 

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;
       }

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

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java

@@ -1563,7 +1563,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()) {
@@ -1613,7 +1613,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,
@@ -1821,7 +1821,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);
@@ -1850,7 +1850,7 @@ private List<InsertRowNode> insertToTsFileProcessors(
     Map<TsFileProcessor, InsertRowsNode> 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;

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;

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) {