public void outputMetricsWithEstimate(String[] recordSchemaInfoArray, boolean isSnapshotRecord, Object data) {
        if (recordSchemaInfoArray == null) {
            outputMetricsWithEstimate(data);
            return;
        }
        String identify = String.join(Constants.SEMICOLON, recordSchemaInfoArray);
        SourceMetricData subSourceMetricData;
        if (subSourceMetricMap.containsKey(identify)) {
            subSourceMetricData = subSourceMetricMap.get(identify);
        } else {
            subSourceMetricData = buildSubSourceMetricData(recordSchemaInfoArray, this);
            subSourceMetricMap.put(identify, subSourceMetricData);
        }
        // source metric and sub source metric output metrics
        long rowCountSize = 1L;
        long rowDataSize = data.toString().getBytes(StandardCharsets.UTF_8).length;
        this.outputMetrics(rowCountSize, rowDataSize);
        subSourceMetricData.outputMetrics(rowCountSize, rowDataSize);

        // output read phase metric
        outputReadPhaseMetrics((isSnapshotRecord) ? ReadPhase.SNAPSHOT_PHASE : ReadPhase.INCREASE_PHASE);
    }
output metrics with estimate

@param recordSchemaInfoArray the schema info of record
@param isSnapshotRecord is it snapshot record
@param data the data of record
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/sub/SourceTableMetricData.java