[Pipe] Prevent receiver statement exception log floods

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -1318,8 +1318,6 @@ public final class DataNodePipeMessages {
       "Receiver id = {}: Failure status encountered while executing statement {}: {}";
   public static final String RECEIVER_ID_EXCEPTION_WHILE_EXECUTING_STATEMENT =
       "Receiver id = {}: Exception encountered while executing statement {}: ";
-  public static final String RECEIVER_ID_STATEMENT_EXCEPTION_MESSAGE =
-      "Receiver id = {}, statement = {}, exception = {}, message = {}";
   public static final String UNKNOWN_PIPEREQUESTTYPE = "Unknown PipeRequestType %s.";
   public static final String EXCEPTION_ENCOUNTERED_WHILE_HANDLING_REQUEST =
       "Exception %s encountered while handling request %s.";

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -1237,8 +1237,6 @@ public final class DataNodePipeMessages {
       "Receiver id = {}: 执行 statement {} 时遇到失败状态：{}";
   public static final String RECEIVER_ID_EXCEPTION_WHILE_EXECUTING_STATEMENT =
       "Receiver id = {}: 执行 statement {} 时遇到异常：";
-  public static final String RECEIVER_ID_STATEMENT_EXCEPTION_MESSAGE =
-      "Receiver id = {}，statement = {}，exception = {}，message = {}";
   public static final String UNKNOWN_PIPEREQUESTTYPE = "未知 PipeRequestType %s。";
   public static final String EXCEPTION_ENCOUNTERED_WHILE_HANDLING_REQUEST =
       "遇到异常 %s，处理请求 %s 时。";

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java

@@ -1080,7 +1080,7 @@ private TSStatus executeStatementAndClassifyExceptions(
   }
 
   private void logStatementExceptionIfNecessary(final Statement statement, final Exception e) {
-    if (shouldLogStatementException(receiverId.get(), statement, e)) {
+    if (shouldLogStatementException(statement, e)) {
       PipeLogger.log(
           LOGGER::warn,
           e,
@@ -1090,16 +1090,25 @@ private void logStatementExceptionIfNecessary(final Statement statement, final E
     }
   }
 
-  static boolean shouldLogStatementException(
-      final long receiverId, final Statement statement, final Exception e) {
-    // Use the reducer cache as a gate. The actual stack trace is logged only when it passes.
-    return LoggerPeriodicalLogReducer.log(
-        message -> {},
-        DataNodePipeMessages.RECEIVER_ID_STATEMENT_EXCEPTION_MESSAGE,
-        receiverId,
-        Objects.isNull(statement) ? null : statement.getPipeLoggingString(),
-        e.getClass().getName(),
-        e.getMessage());
+  static boolean shouldLogStatementException(final Statement statement, final Exception e) {
+    final Throwable rootCause = getRootCause(e);
+    final StackTraceElement[] rootCauseStackTrace = rootCause.getStackTrace();
+    final StackTraceElement rootCauseLocation =
+        rootCauseStackTrace.length > 0 ? rootCauseStackTrace[0] : null;
+
+    // Reduce exceptions raised from the same code location regardless of receiver, statement
+    // content, or dynamic exception message. Different statement types and failure locations are
+    // still logged independently.
+    return LoggerPeriodicalLogReducer.shouldLog(
+        IoTDBDataNodeReceiver.class.getName()
+            + '|'
+            + (Objects.isNull(statement)
+                ? Statement.class.getName()
+                : statement.getClass().getName())
+            + '|'
+            + rootCause.getClass().getName()
+            + '|'
+            + rootCauseLocation);
   }
 
   private TSStatus executeStatementWithPermissionCheckAndRetryOnDataTypeMismatch(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java

@@ -246,6 +246,24 @@ protected void subRemoveAttributeColumns(List<Integer> columnsToKeep) {
     insertRowStatementList.forEach(InsertBaseStatement::removeAttributeColumns);
   }
 
+  @Override
+  public String getPipeLoggingString() {
+    if (Objects.isNull(insertRowStatementList) || insertRowStatementList.isEmpty()) {
+      return "InsertRowsStatement{rowCount=0}";
+    }
+
+    final int rowCount = insertRowStatementList.size();
+    return "InsertRowsStatement{"
+        + "rowCount="
+        + rowCount
+        + ", firstRow="
+        + insertRowStatementList.get(0).getPipeLoggingString()
+        + (rowCount > 1
+            ? ", lastRow=" + insertRowStatementList.get(rowCount - 1).getPipeLoggingString()
+            : "")
+        + '}';
+  }
+
   @Override
   public String toString() {
     return "InsertRowsStatement{" + "insertRowStatementList=" + insertRowStatementList + '}';

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java

@@ -95,25 +95,72 @@ public void testLoadTsFileSyncStatementKeepsDefaultDatabaseLevelWhenDatabaseName
   }
 
   @Test
-  public void testRepeatedStatementExceptionLogIsReduced() throws Exception {
-    final Path tsFile = Files.createTempFile("pipe-load-log-reducer", ".tsfile");
-    try {
-      final LoadTsFileStatement statement =
-          IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync(
-              "root.test.sg_0", tsFile.toString(), true, true);
-      final long receiverId = System.nanoTime();
-      final Exception exception = new RuntimeException("repeated receiver exception " + receiverId);
-
-      Assert.assertTrue(
-          IoTDBDataNodeReceiver.shouldLogStatementException(receiverId, statement, exception));
-      Assert.assertFalse(
-          IoTDBDataNodeReceiver.shouldLogStatementException(receiverId, statement, exception));
-      Assert.assertTrue(
-          IoTDBDataNodeReceiver.shouldLogStatementException(
-              receiverId, statement, new RuntimeException("another receiver exception")));
-    } finally {
-      Files.deleteIfExists(tsFile);
-    }
+  public void testStatementExceptionLogIsReducedByFailureLocation() {
+    final InsertRowStatement firstStatement = new InsertRowStatement();
+    firstStatement.setTime(1);
+    firstStatement.setValues(new Object[] {"first statement value"});
+    final InsertRowStatement secondStatement = new InsertRowStatement();
+    secondStatement.setTime(2);
+    secondStatement.setValues(new Object[] {"second statement value"});
+
+    final String testId = Long.toString(System.nanoTime());
+    final String sameFailureLocation = "sameFailureLocation" + testId;
+    Assert.assertTrue(
+        IoTDBDataNodeReceiver.shouldLogStatementException(
+            firstStatement, newStatementException("first message", sameFailureLocation)));
+    Assert.assertFalse(
+        IoTDBDataNodeReceiver.shouldLogStatementException(
+            secondStatement, newStatementException("second message", sameFailureLocation)));
+    Assert.assertTrue(
+        IoTDBDataNodeReceiver.shouldLogStatementException(
+            secondStatement,
+            newStatementException("second message", "differentFailureLocation" + testId)));
+    Assert.assertTrue(
+        IoTDBDataNodeReceiver.shouldLogStatementException(
+            new InsertRowsStatement(),
+            newStatementException("second message", sameFailureLocation)));
+  }
+
+  @Test
+  public void testInsertRowsPipeLoggingStringIsCompact() {
+    final InsertRowStatement firstStatement = new InsertRowStatement();
+    firstStatement.setTime(1);
+    firstStatement.setValues(new Object[] {"first secret value"});
+    final InsertRowStatement middleStatement = new InsertRowStatement();
+    middleStatement.setTime(2);
+    middleStatement.setValues(new Object[] {"middle secret value"});
+    final InsertRowStatement lastStatement = new InsertRowStatement();
+    lastStatement.setTime(3);
+    lastStatement.setValues(new Object[] {"last secret value"});
+
+    final InsertRowsStatement statement = new InsertRowsStatement();
+    statement.setInsertRowStatementList(
+        Arrays.asList(firstStatement, middleStatement, lastStatement));
+
+    final String pipeLoggingString = statement.getPipeLoggingString();
+    Assert.assertTrue(pipeLoggingString.contains("rowCount=3"));
+    Assert.assertTrue(pipeLoggingString.contains("firstRow="));
+    Assert.assertTrue(pipeLoggingString.contains("time=1"));
+    Assert.assertTrue(pipeLoggingString.contains("lastRow="));
+    Assert.assertTrue(pipeLoggingString.contains("time=3"));
+    Assert.assertFalse(pipeLoggingString.contains("time=2"));
+    Assert.assertFalse(pipeLoggingString.contains("first secret value"));
+    Assert.assertFalse(pipeLoggingString.contains("middle secret value"));
+    Assert.assertFalse(pipeLoggingString.contains("last secret value"));
+  }
+
+  private static Exception newStatementException(
+      final String message, final String failureLocation) {
+    final NullPointerException rootCause = new NullPointerException(message);
+    rootCause.setStackTrace(
+        new StackTraceElement[] {
+          new StackTraceElement(
+              IoTDBDataNodeReceiverTest.class.getName(),
+              failureLocation,
+              "IoTDBDataNodeReceiverTest.java",
+              1)
+        });
+    return new RuntimeException("wrapper " + message, rootCause);
   }
 
   @Test