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/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-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/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 1ac4d80b94fd3..1bdbe703318ac 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 @@ -1559,7 +1559,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()) { @@ -1609,7 +1609,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, @@ -1817,7 +1817,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); @@ -1846,7 +1846,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 697e15cd8f41a..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 @@ -24,6 +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; @@ -32,34 +35,180 @@ 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 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 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