private SinkMetricData buildSubSinkMetricData(String[] schemaInfoArray, MetricState subMetricState,
            SinkMetricData sinkMetricData) {
        if (sinkMetricData == null || schemaInfoArray == null) {
            return null;
        }
        // build sub sink metric data
        Map<String, String> labels = sinkMetricData.getLabels();
        String metricGroupLabels = labels.entrySet().stream().map(entry -> entry.getKey() + "=" + entry.getValue())
                .collect(Collectors.joining(DELIMITER));
        StringBuilder labelBuilder = new StringBuilder(metricGroupLabels);
        if (schemaInfoArray.length == 2) {
            labelBuilder.append(DELIMITER).append(Constants.DATABASE_NAME).append("=").append(schemaInfoArray[0])
                    .append(DELIMITER).append(Constants.TABLE_NAME).append("=").append(schemaInfoArray[1]);
        } else if (schemaInfoArray.length == 3) {
            labelBuilder.append(DELIMITER).append(Constants.DATABASE_NAME).append("=").append(schemaInfoArray[0])
                    .append(DELIMITER).append(Constants.SCHEMA_NAME).append("=").append(schemaInfoArray[1])
                    .append(DELIMITER).append(Constants.TABLE_NAME).append("=").append(schemaInfoArray[2]);
        }

        MetricOption metricOption = MetricOption.builder()
                .withInitRecords(subMetricState != null ? subMetricState.getMetricValue(NUM_RECORDS_OUT) : 0L)
                .withInitBytes(subMetricState != null ? subMetricState.getMetricValue(NUM_BYTES_OUT) : 0L)
                .withInitDirtyRecords(subMetricState != null ? subMetricState.getMetricValue(DIRTY_RECORDS_OUT) : 0L)
                .withInitDirtyBytes(subMetricState != null ? subMetricState.getMetricValue(DIRTY_BYTES_OUT) : 0L)
                .withInlongLabels(labelBuilder.toString()).withRegisterMetric(RegisteredMetric.ALL).build();
        return new SinkTableMetricData(metricOption, sinkMetricData.getMetricGroup());
    }
build sub sink metric data

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