public void registerSubMetricsGroup(MetricState metricState) {
        if (metricState == null) {
            return;
        }
        // register source read phase metric
        Stream.of(ReadPhase.values()).forEach(readPhase -> {
            Long readPhaseMetricValue = metricState.getMetricValue(readPhase.getPhase());
            if (readPhaseMetricValue > 0) {
                outputReadPhaseMetrics(metricState, readPhase);
            }
        });

        // register sub source metric data
        if (metricState.getSubMetricStateMap() == null) {
            return;
        }
        Map<String, MetricState> subMetricStateMap = metricState.getSubMetricStateMap();
        for (Entry<String, MetricState> subMetricStateEntry : subMetricStateMap.entrySet()) {
            String[] schemaInfoArray = subMetricStateEntry.getKey().split(Constants.SPILT_SEMICOLON);
            final MetricState subMetricState = subMetricStateEntry.getValue();
            SourceMetricData subSourceMetricData = buildSubSourceMetricData(schemaInfoArray,
                    subMetricState, this);
            subSourceMetricMap.put(subMetricStateEntry.getKey(), subSourceMetricData);
        }
        LOGGER.info("register subMetricsGroup from metricState,sub metric map size:{}", subSourceMetricMap.size());
    }
register sub source metrics group from metric state

@param metricState MetricState
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/sub/SourceTableMetricData.java