public void outputReadPhaseMetrics(MetricState metricState, ReadPhase readPhase) {
        ReadPhaseMetricData readPhaseMetricData = readPhaseMetricDataMap.get(readPhase);
        if (readPhaseMetricData == null) {
            // build read phase metric data
            String metricGroupLabels = getLabels().entrySet().stream()
                    .map(entry -> entry.getKey() + "=" + entry.getValue())
                    .collect(Collectors.joining(DELIMITER));

            MetricOption metricOption = MetricOption.builder()
                    .withInitReadPhase(metricState != null ? metricState.getMetricValue(readPhase.getPhase()) : 0L)
                    .withInlongLabels(metricGroupLabels + DELIMITER + READ_PHASE + "=" + readPhase.getCode())
                    .withRegisterMetric(RegisteredMetric.ALL)
                    .build();
            readPhaseMetricData = new ReadPhaseMetricData(metricOption, getMetricGroup());
            readPhaseMetricDataMap.put(readPhase, readPhaseMetricData);
        }
        readPhaseMetricData.outputMetrics();
    }
output read phase metric

@param metricState the metric state of record
@param readPhase the readPhase of record
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/sub/SourceTableMetricData.java