private SourceMetricData buildSubSourceMetricData(String[] schemaInfoArray, MetricState subMetricState,
            SourceMetricData sourceMetricData) {
        if (sourceMetricData == null || schemaInfoArray == null) {
            return null;
        }
        // build sub source metric data
        Map<String, String> labels = sourceMetricData.getLabels();
        String metricGroupLabels = labels.entrySet().stream().map(entry -> entry.getKey() + "=" + entry.getValue())
                .collect(Collectors.joining(DELIMITER));
        StringBuilder labelBuilder = new StringBuilder(metricGroupLabels);
        for (int i = 0; i < tableMetricLabelList.size(); i++) {
            labelBuilder.append(DELIMITER).append(tableMetricLabelList.get(i)).append("=").append(schemaInfoArray[i]);
        }
        MetricOption metricOption = MetricOption.builder()
                .withInitRecords(subMetricState != null ? subMetricState.getMetricValue(NUM_RECORDS_IN) : 0L)
                .withInitBytes(subMetricState != null ? subMetricState.getMetricValue(NUM_BYTES_IN) : 0L)
                .withInlongLabels(labelBuilder.toString())
                .withRegisterMetric(RegisteredMetric.ALL)
                .build();
        return new SourceTableMetricData(metricOption, sourceMetricData.getMetricGroup(), tableMetricLabelList);
    }
build sub source metric data

@param schemaInfoArray the schema info array of record
@param subMetricState sub metric state
@param sourceMetricData source metric data
@return sub source metric data
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/sub/SourceTableMetricData.java