Add aggregation UT

Author: Caideyipi

integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBShowReceiversIT.java

@@ -70,6 +70,20 @@ public void testInformationSchemaReceiversWithProtocolFilter() {
         "show_receivers_filter_pipe");
   }
 
+  @Test
+  public void testShowReceiversAggregatesMultiplePipesFromSameSender() {
+    final String pipeName1 = "show_receivers_aggregate_pipe_1";
+    final String pipeName2 = "show_receivers_aggregate_pipe_2";
+    createTwoWriteBackPipes("root.show_receivers_aggregate", pipeName1, pipeName2);
+
+    assertAggregatedShowReceivers("show receivers", BaseEnv.TREE_SQL_DIALECT, pipeName1, pipeName2);
+    assertAggregatedShowReceivers(
+        "select * from information_schema.receivers",
+        BaseEnv.TABLE_SQL_DIALECT,
+        pipeName1,
+        pipeName2);
+  }
+
   @Test
   public void testShowReceiversWithStoppedDataNode() throws Exception {
     Assert.assertTrue(env.getDataNodeWrapperList().size() >= 3);
@@ -152,6 +166,28 @@ private void createWriteBackPipe(final String database, final String pipeName) {
         null);
   }
 
+  private void createTwoWriteBackPipes(
+      final String database, final String firstPipeName, final String secondPipeName) {
+    TestUtils.executeNonQueries(
+        env,
+        Arrays.asList(
+            "create database " + database,
+            "create timeseries " + database + ".d1.s1 with datatype=INT32, encoding=PLAIN",
+            "create pipe "
+                + firstPipeName
+                + " with source ('pattern'='"
+                + database
+                + "') with sink ('sink'='write-back-sink')",
+            "create pipe "
+                + secondPipeName
+                + " with source ('pattern'='"
+                + database
+                + "') with sink ('sink'='write-back-sink')",
+            "insert into " + database + ".d1(time, s1) values (1, 1)",
+            "flush"),
+        null);
+  }
+
   private void assertShowReceivers(final String sql, final String sqlDialect) {
     assertShowReceivers(sql, sqlDialect, "show_receivers_pipe");
   }
@@ -183,6 +219,22 @@ private void assertShowReceiversAsUser(
                     hasExpectedReceiver(sql, sqlDialect, pipeName, userName, password)));
   }
 
+  private void assertAggregatedShowReceivers(
+      final String sql,
+      final String sqlDialect,
+      final String firstPipeName,
+      final String secondPipeName) {
+    Awaitility.await()
+        .pollInSameThread()
+        .pollDelay(1L, TimeUnit.SECONDS)
+        .pollInterval(1L, TimeUnit.SECONDS)
+        .atMost(60L, TimeUnit.SECONDS)
+        .untilAsserted(
+            () ->
+                Assert.assertTrue(
+                    hasAggregatedReceiver(sql, sqlDialect, firstPipeName, secondPipeName)));
+  }
+
   private void assertUserCannotSeeReceivers(
       final String sql, final String sqlDialect, final String userName, final String password) {
     Awaitility.await()
@@ -232,6 +284,33 @@ private boolean hasExpectedReceiver(
     }
   }
 
+  private boolean hasAggregatedReceiver(
+      final String sql,
+      final String sqlDialect,
+      final String firstPipeName,
+      final String secondPipeName)
+      throws SQLException {
+    try (final Connection connection =
+            env.getConnection(
+                SessionConfig.DEFAULT_USER, SessionConfig.DEFAULT_PASSWORD, sqlDialect);
+        final Statement statement = connection.createStatement();
+        final ResultSet resultSet = statement.executeQuery(sql)) {
+      while (resultSet.next()) {
+        final String pipeIds = resultSet.getString(8);
+        if ("DataNode".equals(resultSet.getString(1))
+            && "thrift".equals(resultSet.getString(3))
+            && resultSet.getInt(6) >= 1
+            && resultSet.getInt(7) >= 2
+            && pipeIds != null
+            && pipeIds.contains(firstPipeName + "@")
+            && pipeIds.contains(secondPipeName + "@")) {
+          return true;
+        }
+      }
+      return false;
+    }
+  }
+
   private boolean hasAnyReceiver(
       final String sql, final String sqlDialect, final String userName, final String password)
       throws SQLException {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/ShowReceiversOperatorTest.java

@@ -123,6 +123,48 @@ public void testSingleSenderSingleDataNodeSingleConnectionSinglePipeOutput() {
     assertEquals(DateTimeUtils.convertLongToDate(200, "ms"), getText(tsBlock, 11));
   }
 
+  @Test
+  public void testMultiplePipesFromSameSenderAreAggregatedInOutput() {
+    registry.registerOrUpdateSession(
+        "data-1",
+        PipeReceiverRuntimeRegistry.NODE_TYPE_DATA_NODE,
+        1,
+        PipeReceiverRuntimeRegistry.PROTOCOL_THRIFT,
+        "10.0.0.1",
+        9001,
+        "root",
+        "cluster-a",
+        "pipe-a",
+        1,
+        100);
+    registry.registerOrUpdateSession(
+        "data-2",
+        PipeReceiverRuntimeRegistry.NODE_TYPE_DATA_NODE,
+        1,
+        PipeReceiverRuntimeRegistry.PROTOCOL_THRIFT,
+        "10.0.0.1",
+        9002,
+        "root",
+        "cluster-a",
+        "pipe-b",
+        2,
+        200);
+
+    final ShowReceiversOperator operator =
+        new ShowReceiversOperator(null, new PlanNodeId("show-receivers"));
+
+    assertTrue(operator.hasNext());
+    final TsBlock tsBlock = operator.next();
+
+    assertEquals(1, tsBlock.getPositionCount());
+    assertEquals("9001,9002", getText(tsBlock, 4));
+    assertEquals(2, tsBlock.getColumn(5).getInt(0));
+    assertEquals(2, tsBlock.getColumn(6).getInt(0));
+    assertTrue(getText(tsBlock, 7).contains("pipe-a@"));
+    assertTrue(getText(tsBlock, 7).contains("pipe-b@"));
+    assertFalse(operator.hasNext());
+  }
+
   @Test
   public void testLastTransferTimeUsesHandshakeTimeBeforeTransfer() {
     registry.registerOrUpdateSession(