This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new e492e444a8d [Fix] Preserve table insert errors for rejected fields
(#18880)
e492e444a8d is described below
commit e492e444a8da398d6ecf5c7f4882f8221432a929
Author: Hongzhi Gao <[email protected]>
AuthorDate: Fri Oct 16 10:29:56 2026 +0800
[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
---
.../relational/it/db/it/IoTDBInsertTableIT.java | 62 +++++++++
.../it/session/IoTDBSessionRelationalIT.java | 48 +++++--
.../java/org/apache/iotdb/session/Session.java | 7 +-
.../iotdb/session/SessionCacheLeaderTest.java | 79 ++++++++++-
.../plan/planner/plan/node/write/InsertNode.java | 5 +
.../db/storageengine/dataregion/DataRegion.java | 8 +-
.../dataregion/memtable/AbstractMemTable.java | 4 +-
.../wal/recover/file/TsFilePlanRedoer.java | 4 +-
.../queryengine/plan/analyze/AnalyzeUtilsTest.java | 149 +++++++++++++++++++++
.../wal/recover/file/TsFilePlanRedoerTest.java | 52 +++++++
10 files changed, 394 insertions(+), 24 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 3abe7497130..70eab82727c 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 class IoTDBInsertTableIT {
}
}
+ @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/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 63a5548ee90..d6f7aef603d 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
@@ -449,16 +449,19 @@ public class IoTDBSessionRelationalIT {
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 class IoTDBSessionRelationalIT {
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 @@ public class IoTDBSessionRelationalIT {
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-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 fe9fb38f6db..6d31f81885a 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 @@ public class Session implements ISession {
.collect(Collectors.toList());
StringBuilder errMsgBuilder = new StringBuilder();
+ List<Throwable> errors = new ArrayList<>();
for (CompletableFuture<Void> completableFuture : completableFutures) {
try {
completableFuture.join();
@@ -3016,6 +3017,7 @@ public class Session implements ISession {
if (cause instanceof IoTDBConnectionException) {
throw (IoTDBConnectionException) cause;
} else {
+ errors.add(cause);
if (errMsgBuilder.length() > 0) {
errMsgBuilder.append(";");
}
@@ -3024,7 +3026,10 @@ public class Session implements ISession {
}
}
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 40268a54cd3..46ec3ee1f5d 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 org.junit.Test;
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 class SessionCacheLeaderTest {
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 @@ public class SessionCacheLeaderTest {
@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/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 289866c6882..0f2bc190120 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 abstract class InsertNode extends SearchNode {
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 2dcb699807d..b7284abd0e3 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
@@ -1563,7 +1563,7 @@ public class DataRegion implements IDataRegionForQuery {
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 @@ public class DataRegion implements IDataRegionForQuery {
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 @@ public class DataRegion implements IDataRegionForQuery {
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 @@ public class DataRegion implements IDataRegionForQuery {
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;
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 5fd3d5f9224..382063359c1 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 abstract class AbstractMemTable implements IMemTable
{
dataTypes.add(schema.getType());
writableValues[i] = value;
}
- if (schemaList.isEmpty()) {
+ if (schemaList.isEmpty() && insertRowNode.getColumnCategories() == null) {
return 0;
}
memSize +=
@@ -497,7 +497,7 @@ public abstract class AbstractMemTable implements IMemTable
{
}
}
}
- 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 8e191d475ba..c4550898a05 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 @@ public class TsFilePlanRedoer {
}
void redoInsert(InsertNode node) throws WriteProcessException {
- if (!node.hasValidMeasurements()) {
+ if (node.shouldSkipWrite()) {
return;
}
boolean isSingleDevice = !(node instanceof RelationalInsertTabletNode);
@@ -155,7 +155,7 @@ public class TsFilePlanRedoer {
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 697e15cd8f4..e5a1fb23355 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,9 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
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;
@@ -32,34 +35,180 @@ import
org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.QualifiedNam
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.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;
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;
+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;
public class AnalyzeUtilsTest {
+ @Test
+ 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);
+ assertRejectedFieldStillLooksUpPartition(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);
+ assertRejectedFieldStillLooksUpPartition(tablet);
+ }
+
+ private static void
assertRejectedFieldStillLooksUpPartition(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"));
+ 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) -> {
+ partitionQueried.set(true);
+ return partition;
+ },
+ AnalyzeUtils::computeTableDataPartitionParams,
+ analysis,
+ false);
+
+ 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 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) -> {
+ partitionQueried.set(true);
+ return partition;
+ },
+ AnalyzeUtils::computeTableDataPartitionParams,
+ analysis,
+ false);
+
+ 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 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<InsertRowStatement> 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");
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 238606b073d..a58171017ac 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.MeasurementPath;
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 org.junit.Test;
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 class TsFilePlanRedoerTest {
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
