Pipe: Implemented schema batch for meta transfer (#17665)

* Schema

* fix

* fix

* Update PipeSchemaRegionSinkMetricsTest.java

* spt

* Address schema batch review comments

* Avoid extra copy in plan node airgap transfer

* Fix pipe plan node byte buffer test

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java

@@ -348,6 +348,12 @@ public void setTabletBatchSizeHistogram(Histogram tabletBatchSizeHistogram) {
     }
   }
 
+  public void setSchemaBatchSizeHistogram(Histogram schemaBatchSizeHistogram) {
+    if (outputPipeSink instanceof IoTDBSink) {
+      ((IoTDBSink) outputPipeSink).setSchemaBatchSizeHistogram(schemaBatchSizeHistogram);
+    }
+  }
+
   public void setTsFileBatchSizeHistogram(Histogram tsFileBatchSizeHistogram) {
     if (outputPipeSink instanceof IoTDBSink) {
       ((IoTDBSink) outputPipeSink).setTsFileBatchSizeHistogram(tsFileBatchSizeHistogram);
@@ -361,6 +367,13 @@ public void setTabletBatchTimeIntervalHistogram(Histogram tabletBatchTimeInterva
     }
   }
 
+  public void setSchemaBatchTimeIntervalHistogram(Histogram schemaBatchTimeIntervalHistogram) {
+    if (outputPipeSink instanceof IoTDBSink) {
+      ((IoTDBSink) outputPipeSink)
+          .setSchemaBatchTimeIntervalHistogram(schemaBatchTimeIntervalHistogram);
+    }
+  }
+
   public void setTsFileBatchTimeIntervalHistogram(Histogram tsFileBatchTimeIntervalHistogram) {
     if (outputPipeSink instanceof IoTDBSink) {
       ((IoTDBSink) outputPipeSink)

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/schema/PipeSchemaRegionSinkMetrics.java

@@ -25,6 +25,7 @@
 import org.apache.iotdb.db.pipe.agent.task.subtask.sink.PipeSinkSubtask;
 import org.apache.iotdb.metrics.AbstractMetricService;
 import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.type.Histogram;
 import org.apache.iotdb.metrics.type.Rate;
 import org.apache.iotdb.metrics.utils.MetricLevel;
 import org.apache.iotdb.metrics.utils.MetricType;
@@ -57,6 +58,7 @@ public void bindTo(final AbstractMetricService metricService) {
 
   private void createMetrics(final String taskID) {
     createRate(taskID);
+    createHistogram(taskID);
   }
 
   private void createRate(final String taskID) {
@@ -73,6 +75,38 @@ private void createRate(final String taskID) {
             String.valueOf(connector.getCreationTime())));
   }
 
+  private void createHistogram(final String taskID) {
+    final PipeSinkSubtask connector = connectorMap.get(taskID);
+
+    final Histogram schemaBatchSizeHistogram =
+        metricService.getOrCreateHistogram(
+            Metric.PIPE_SCHEMA_BATCH_SIZE.toString(),
+            MetricLevel.IMPORTANT,
+            Tag.NAME.toString(),
+            connector.getAttributeSortedString(),
+            Tag.CREATION_TIME.toString(),
+            String.valueOf(connector.getCreationTime()));
+    connector.setSchemaBatchSizeHistogram(schemaBatchSizeHistogram);
+
+    final Histogram schemaBatchTimeIntervalHistogram =
+        metricService.getOrCreateHistogram(
+            Metric.PIPE_SCHEMA_BATCH_TIME_COST.toString(),
+            MetricLevel.IMPORTANT,
+            Tag.NAME.toString(),
+            connector.getAttributeSortedString(),
+            Tag.CREATION_TIME.toString(),
+            String.valueOf(connector.getCreationTime()));
+    connector.setSchemaBatchTimeIntervalHistogram(schemaBatchTimeIntervalHistogram);
+
+    final Histogram schemaBatchEventSizeHistogram =
+        metricService.getOrCreateHistogram(
+            Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
+            MetricLevel.IMPORTANT,
+            Tag.NAME.toString(),
+            connector.getAttributeSortedString());
+    connector.setEventSizeHistogram(schemaBatchEventSizeHistogram);
+  }
+
   @Override
   public void unbindFrom(final AbstractMetricService metricService) {
     ImmutableSet.copyOf(connectorMap.keySet()).forEach(this::deregister);
@@ -83,6 +117,7 @@ public void unbindFrom(final AbstractMetricService metricService) {
 
   private void removeMetrics(final String taskID) {
     removeRate(taskID);
+    removeHistogram(taskID);
   }
 
   private void removeRate(final String taskID) {
@@ -98,6 +133,29 @@ private void removeRate(final String taskID) {
     schemaRateMap.remove(taskID);
   }
 
+  private void removeHistogram(final String taskID) {
+    final PipeSinkSubtask connector = connectorMap.get(taskID);
+    metricService.remove(
+        MetricType.HISTOGRAM,
+        Metric.PIPE_SCHEMA_BATCH_SIZE.toString(),
+        Tag.NAME.toString(),
+        connector.getAttributeSortedString(),
+        Tag.CREATION_TIME.toString(),
+        String.valueOf(connector.getCreationTime()));
+    metricService.remove(
+        MetricType.HISTOGRAM,
+        Metric.PIPE_SCHEMA_BATCH_TIME_COST.toString(),
+        Tag.NAME.toString(),
+        connector.getAttributeSortedString(),
+        Tag.CREATION_TIME.toString(),
+        String.valueOf(connector.getCreationTime()));
+    metricService.remove(
+        MetricType.HISTOGRAM,
+        Metric.PIPE_CONNECTOR_BATCH_SIZE.toString(),
+        Tag.NAME.toString(),
+        connector.getAttributeSortedString());
+  }
+
   //////////////////////////// Register & deregister (pipe integration) ////////////////////////////
 
   public void register(final PipeSinkSubtask pipeSinkSubtask) {