diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index 1f8300c2807ed..0acc13f6cf53e 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -20,6 +20,8 @@ package org.apache.iotdb.db.i18n; public final class DataNodeQueryMessages { + public static final String MESSAGE_FAIL_TO_INSERT_MEASUREMENTS_ARG_CAUSED_BY_ARG_B9858E6B = + "Fail to insert measurements %s caused by %s"; private DataNodeQueryMessages() {} diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index 32e74a0e5b94f..72ea24db250f4 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -20,6 +20,8 @@ package org.apache.iotdb.db.i18n; public final class DataNodeQueryMessages { + public static final String MESSAGE_FAIL_TO_INSERT_MEASUREMENTS_ARG_CAUSED_BY_ARG_B9858E6B = + "测点 %s 插入失败，原因：%s"; private DataNodeQueryMessages() {} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java index c3d48812a462c..c393c275dd529 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java @@ -49,6 +49,7 @@ import org.apache.iotdb.db.exception.metadata.view.UnsupportedViewException; import org.apache.iotdb.db.exception.sql.SemanticException; import org.apache.iotdb.db.exception.sql.StatementAnalyzeException; +import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.protocol.client.ConfigNodeClient; import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager; import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; @@ -3120,16 +3121,43 @@ private void validateSchema( boolean hasFailedMeasurement = insertStatement.hasFailedMeasurements(); String partialInsertMessage; if (hasFailedMeasurement) { + final List failedMeasurements = new ArrayList<>(); + final List failedMessages = new ArrayList<>(); + collectDistinctFailedMeasurementMessages( + insertStatement.getFailedMeasurements(), + insertStatement.getFailedMessages(), + failedMeasurements, + failedMessages); partialInsertMessage = String.format( - "Fail to insert measurements %s caused by %s", - insertStatement.getFailedMeasurements(), insertStatement.getFailedMessages()); + DataNodeQueryMessages.MESSAGE_FAIL_TO_INSERT_MEASUREMENTS_ARG_CAUSED_BY_ARG_B9858E6B, + failedMeasurements, + failedMessages); logger.warn(partialInsertMessage); analysis.setFailStatus( RpcUtils.getStatus(TSStatusCode.METADATA_ERROR.getStatusCode(), partialInsertMessage)); } } + private static void collectDistinctFailedMeasurementMessages( + List rawFailedMeasurements, + List rawFailedMessages, + List failedMeasurements, + List failedMessages) { + final Set> distinctFailures = new LinkedHashSet<>(); + for (int i = 0; i < rawFailedMeasurements.size(); i++) { + final String failedMeasurement = rawFailedMeasurements.get(i); + final String failedMessage = i < rawFailedMessages.size() ? rawFailedMessages.get(i) : null; + final List failureKey = new ArrayList<>(2); + failureKey.add(failedMeasurement); + failureKey.add(failedMessage); + if (distinctFailures.add(failureKey)) { + failedMeasurements.add(failedMeasurement); + failedMessages.add(failedMessage); + } + } + } + private InsertBaseStatement removeLogicalView( Analysis analysis, InsertBaseStatement insertBaseStatement) { try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertMultiTabletsStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertMultiTabletsStatement.java index b300da7b26e95..a88bcf638690f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertMultiTabletsStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertMultiTabletsStatement.java @@ -21,7 +21,6 @@ import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.path.PartialPath; -import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaValidation; import org.apache.iotdb.db.queryengine.plan.statement.StatementType; import org.apache.iotdb.db.queryengine.plan.statement.StatementVisitor; @@ -31,6 +30,7 @@ import org.apache.tsfile.utils.RamUsageEstimator; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Objects; import java.util.stream.Collectors; @@ -104,12 +104,45 @@ public List getSchemaValidationList() { } @Override - public void updateAfterSchemaValidation(MPPQueryContext context) { - for (InsertTabletStatement insertTabletStatement : insertTabletStatementList) { - if (!this.hasFailedMeasurements() && insertTabletStatement.hasFailedMeasurements()) { - this.failedMeasurementIndex2Info = insertTabletStatement.failedMeasurementIndex2Info; - } - } + public boolean hasFailedMeasurements() { + return insertTabletStatementList != null + && insertTabletStatementList.stream().anyMatch(InsertBaseStatement::hasFailedMeasurements); + } + + @Override + public int getFailedMeasurementNumber() { + return insertTabletStatementList == null + ? 0 + : insertTabletStatementList.stream() + .mapToInt(InsertBaseStatement::getFailedMeasurementNumber) + .sum(); + } + + @Override + public List getFailedMeasurements() { + return insertTabletStatementList == null + ? Collections.emptyList() + : insertTabletStatementList.stream() + .flatMap(statement -> statement.getFailedMeasurements().stream()) + .collect(Collectors.toList()); + } + + @Override + public List getFailedExceptions() { + return insertTabletStatementList == null + ? Collections.emptyList() + : insertTabletStatementList.stream() + .flatMap(statement -> statement.getFailedExceptions().stream()) + .collect(Collectors.toList()); + } + + @Override + public List getFailedMessages() { + return insertTabletStatementList == null + ? Collections.emptyList() + : insertTabletStatementList.stream() + .flatMap(statement -> statement.getFailedMessages().stream()) + .collect(Collectors.toList()); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsOfOneDeviceStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsOfOneDeviceStatement.java index ef42489e87530..fe79867b062fa 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsOfOneDeviceStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsOfOneDeviceStatement.java @@ -33,6 +33,7 @@ import org.apache.tsfile.utils.RamUsageEstimator; import java.util.ArrayList; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Objects; @@ -127,12 +128,51 @@ public List getSchemaValidationList() { public void updateAfterSchemaValidation(MPPQueryContext context) throws QueryProcessException { for (InsertRowStatement insertRowStatement : insertRowStatementList) { insertRowStatement.updateAfterSchemaValidation(context); - if (!this.hasFailedMeasurements() && insertRowStatement.hasFailedMeasurements()) { - this.failedMeasurementIndex2Info = insertRowStatement.failedMeasurementIndex2Info; - } } } + @Override + public boolean hasFailedMeasurements() { + return insertRowStatementList != null + && insertRowStatementList.stream().anyMatch(InsertBaseStatement::hasFailedMeasurements); + } + + @Override + public int getFailedMeasurementNumber() { + return insertRowStatementList == null + ? 0 + : insertRowStatementList.stream() + .mapToInt(InsertBaseStatement::getFailedMeasurementNumber) + .sum(); + } + + @Override + public List getFailedMeasurements() { + return insertRowStatementList == null + ? Collections.emptyList() + : insertRowStatementList.stream() + .flatMap(statement -> statement.getFailedMeasurements().stream()) + .collect(Collectors.toList()); + } + + @Override + public List getFailedExceptions() { + return insertRowStatementList == null + ? Collections.emptyList() + : insertRowStatementList.stream() + .flatMap(statement -> statement.getFailedExceptions().stream()) + .collect(Collectors.toList()); + } + + @Override + public List getFailedMessages() { + return insertRowStatementList == null + ? Collections.emptyList() + : insertRowStatementList.stream() + .flatMap(statement -> statement.getFailedMessages().stream()) + .collect(Collectors.toList()); + } + @Override protected boolean checkAndCastDataType(int columnIndex, TSDataType dataType) { return false; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java index 0a4a84ad032da..99d20908b6832 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java @@ -34,6 +34,7 @@ import java.util.List; import java.util.Objects; import java.util.stream.Collectors; +import java.util.stream.Stream; public class InsertRowsStatement extends InsertBaseStatement { @@ -44,6 +45,8 @@ public class InsertRowsStatement extends InsertBaseStatement { /** the InsertRowsStatement list */ private List insertRowStatementList; + private List schemaValidatedRowStatementList; + public InsertRowsStatement() { super(); statementType = StatementType.BATCH_INSERT_ROWS; @@ -124,10 +127,51 @@ public List getSchemaValidationList() { public void updateAfterSchemaValidation(MPPQueryContext context) throws QueryProcessException { for (InsertRowStatement insertRowStatement : insertRowStatementList) { insertRowStatement.updateAfterSchemaValidation(context); - if (!this.hasFailedMeasurements() && insertRowStatement.hasFailedMeasurements()) { - this.failedMeasurementIndex2Info = insertRowStatement.failedMeasurementIndex2Info; - } } + // Table schema validation can remove rows without valid fields. Keep their failures for + // reporting. + schemaValidatedRowStatementList = insertRowStatementList; + } + + private Stream getRowsForFailureReporting() { + final List rows = + schemaValidatedRowStatementList == null + ? insertRowStatementList + : schemaValidatedRowStatementList; + return rows == null ? Stream.empty() : rows.stream(); + } + + @Override + public boolean hasFailedMeasurements() { + return getRowsForFailureReporting().anyMatch(InsertBaseStatement::hasFailedMeasurements); + } + + @Override + public int getFailedMeasurementNumber() { + return getRowsForFailureReporting() + .mapToInt(InsertBaseStatement::getFailedMeasurementNumber) + .sum(); + } + + @Override + public List getFailedMeasurements() { + return getRowsForFailureReporting() + .flatMap(statement -> statement.getFailedMeasurements().stream()) + .collect(Collectors.toList()); + } + + @Override + public List getFailedExceptions() { + return getRowsForFailureReporting() + .flatMap(statement -> statement.getFailedExceptions().stream()) + .collect(Collectors.toList()); + } + + @Override + public List getFailedMessages() { + return getRowsForFailureReporting() + .flatMap(statement -> statement.getFailedMessages().stream()) + .collect(Collectors.toList()); } @Override diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertStatementPartialInsertTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertStatementPartialInsertTest.java index d696745edf35f..056e72ebb0beb 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertStatementPartialInsertTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertStatementPartialInsertTest.java @@ -19,7 +19,17 @@ package org.apache.iotdb.db.queryengine.plan.statement.crud; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.db.exception.query.QueryProcessException; +import org.apache.iotdb.db.i18n.DataNodeQueryMessages; +import org.apache.iotdb.db.queryengine.common.MPPQueryContext; +import org.apache.iotdb.db.queryengine.common.QueryId; +import org.apache.iotdb.db.queryengine.common.SessionInfo; +import org.apache.iotdb.db.queryengine.plan.analyze.Analysis; +import org.apache.iotdb.db.queryengine.plan.analyze.Analyzer; +import org.apache.iotdb.db.queryengine.plan.analyze.FakePartitionFetcherImpl; +import org.apache.iotdb.db.queryengine.plan.analyze.FakeSchemaFetcherImpl; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.utils.Pair; @@ -27,6 +37,7 @@ import org.junit.Assert; import org.junit.Test; +import java.time.ZoneId; import java.util.Arrays; import java.util.List; import java.util.Map; @@ -157,6 +168,137 @@ public void testDeviceMeasurementMapSkipsMissingTabletColumns() throws Exception Assert.assertEquals("s1", result.get(statement.getDevicePath()).get(0).left); } + @Test + public void testInsertRowsStatementAggregatesFailedMeasurements() throws Exception { + final InsertRowStatement statement1 = createInsertRowStatement(); + final InsertRowStatement statement2 = createInsertRowStatement(); + statement1.markFailedMeasurement(0, new RuntimeException("failed1")); + statement2.markFailedMeasurement(1, new RuntimeException("failed2")); + + final InsertRowsStatement rowsStatement = new InsertRowsStatement(); + rowsStatement.setInsertRowStatementList(Arrays.asList(statement1, statement2)); + + Assert.assertTrue(rowsStatement.hasFailedMeasurements()); + Assert.assertEquals(2, rowsStatement.getFailedMeasurementNumber()); + Assert.assertEquals(Arrays.asList("s1", "s2"), rowsStatement.getFailedMeasurements()); + Assert.assertEquals(Arrays.asList("failed1", "failed2"), rowsStatement.getFailedMessages()); + } + + @Test + public void testInsertRowsStatementRetainsFailuresAfterRemovingInvalidRows() throws Exception { + final InsertRowStatement validRow = createInsertRowStatement(); + final InsertRowStatement invalidRow = createInsertRowStatement(); + final InsertRowStatement partiallyValidRow = createInsertRowStatement(); + final RuntimeException failure1 = new RuntimeException("failed1"); + final RuntimeException failure2 = new RuntimeException("failed2"); + final RuntimeException failure3 = new RuntimeException("failed3"); + invalidRow.markFailedMeasurement(0, failure1); + invalidRow.markFailedMeasurement(1, failure2); + partiallyValidRow.markFailedMeasurement(0, failure3); + + final List rowsAfterSchemaValidation = + Arrays.asList(validRow, partiallyValidRow); + final InsertRowsStatement rowsStatement = + new InsertRowsStatement() { + @Override + public void updateAfterSchemaValidation(MPPQueryContext context) + throws QueryProcessException { + super.updateAfterSchemaValidation(context); + setInsertRowStatementList(rowsAfterSchemaValidation); + } + }; + rowsStatement.setInsertRowStatementList(Arrays.asList(validRow, invalidRow, partiallyValidRow)); + + final Analysis analysis = analyzeRowsStatement(rowsStatement); + + Assert.assertEquals( + Arrays.asList(validRow, partiallyValidRow), rowsStatement.getInsertRowStatementList()); + Assert.assertTrue(rowsStatement.hasFailedMeasurements()); + Assert.assertEquals(3, rowsStatement.getFailedMeasurementNumber()); + Assert.assertEquals(Arrays.asList("s1", "s2", "s1"), rowsStatement.getFailedMeasurements()); + Assert.assertEquals( + Arrays.asList(failure1, failure2, failure3), rowsStatement.getFailedExceptions()); + Assert.assertEquals( + Arrays.asList("failed1", "failed2", "failed3"), rowsStatement.getFailedMessages()); + Assert.assertEquals( + String.format( + DataNodeQueryMessages.MESSAGE_FAIL_TO_INSERT_MEASUREMENTS_ARG_CAUSED_BY_ARG_B9858E6B, + "[s1, s2, s1]", + "[failed1, failed2, failed3]"), + analysis.getFailStatus().getMessage()); + } + + @Test + public void testValidateSchemaDeduplicatesFailedMeasurementMessage() throws Exception { + final InsertRowStatement statement1 = createInsertRowStatement(); + final InsertRowStatement statement2 = createInsertRowStatement(); + final InsertRowStatement statement3 = createInsertRowStatement(); + statement1.markFailedMeasurement(0, new RuntimeException("failed1")); + statement2.markFailedMeasurement(0, new RuntimeException("failed1")); + statement3.markFailedMeasurement(1, new RuntimeException("failed2")); + + final InsertRowsStatement rowsStatement = new InsertRowsStatement(); + rowsStatement.setInsertRowStatementList(Arrays.asList(statement1, statement2, statement3)); + + final Analysis analysis = analyzeRowsStatement(rowsStatement); + + Assert.assertEquals( + String.format( + DataNodeQueryMessages.MESSAGE_FAIL_TO_INSERT_MEASUREMENTS_ARG_CAUSED_BY_ARG_B9858E6B, + "[s1, s2]", + "[failed1, failed2]"), + analysis.getFailStatus().getMessage()); + } + + private static Analysis analyzeRowsStatement(InsertRowsStatement statement) { + return new Analyzer( + new MPPQueryContext( + "", + new QueryId("partial_insert_test"), + new SessionInfo(0, "test", ZoneId.systemDefault(), ""), + new TEndPoint(), + new TEndPoint()), + new FakePartitionFetcherImpl(), + new FakeSchemaFetcherImpl()) + .analyze(statement); + } + + @Test + public void testInsertRowsOfOneDeviceStatementAggregatesFailedMeasurements() throws Exception { + final InsertRowStatement statement1 = createInsertRowStatement(); + final InsertRowStatement statement2 = createInsertRowStatement(); + statement1.markFailedMeasurement(0, new RuntimeException("failed1")); + statement2.markFailedMeasurement(1, new RuntimeException("failed2")); + + final InsertRowsOfOneDeviceStatement rowsOfOneDeviceStatement = + new InsertRowsOfOneDeviceStatement(); + rowsOfOneDeviceStatement.setInsertRowStatementList(Arrays.asList(statement1, statement2)); + + Assert.assertTrue(rowsOfOneDeviceStatement.hasFailedMeasurements()); + Assert.assertEquals(2, rowsOfOneDeviceStatement.getFailedMeasurementNumber()); + Assert.assertEquals( + Arrays.asList("s1", "s2"), rowsOfOneDeviceStatement.getFailedMeasurements()); + Assert.assertEquals( + Arrays.asList("failed1", "failed2"), rowsOfOneDeviceStatement.getFailedMessages()); + } + + @Test + public void testInsertMultiTabletsStatementAggregatesFailedMeasurements() throws Exception { + final InsertTabletStatement statement1 = createInsertTabletStatement(); + final InsertTabletStatement statement2 = createInsertTabletStatement(); + statement1.markFailedMeasurement(0, new RuntimeException("failed1")); + statement2.markFailedMeasurement(1, new RuntimeException("failed2")); + + final InsertMultiTabletsStatement multiTabletsStatement = new InsertMultiTabletsStatement(); + multiTabletsStatement.setInsertTabletStatementList(Arrays.asList(statement1, statement2)); + + Assert.assertTrue(multiTabletsStatement.hasFailedMeasurements()); + Assert.assertEquals(2, multiTabletsStatement.getFailedMeasurementNumber()); + Assert.assertEquals(Arrays.asList("s1", "s2"), multiTabletsStatement.getFailedMeasurements()); + Assert.assertEquals( + Arrays.asList("failed1", "failed2"), multiTabletsStatement.getFailedMessages()); + } + private static InsertRowStatement createInsertRowStatement() throws Exception { final InsertRowStatement statement = new InsertRowStatement(); statement.setDevicePath(new PartialPath("root.sg.d1"));