public void outputMetricsWithEstimate(String database, String table, boolean isSnapshotRecord, Object data) {
        if (StringUtils.isBlank(database) || StringUtils.isBlank(table)) {
            outputMetricsWithEstimate(data);
            // output read phase metric
            outputReadPhaseMetrics((isSnapshotRecord) ? ReadPhase.SNAPSHOT_PHASE : ReadPhase.INCREASE_PHASE);
            return;
        }
        // output sub source metric
        outputMetricsWithEstimate(new String[]{database, table}, isSnapshotRecord, data);
    }
output metrics with estimate

@param database the database name of record
@param table the table name 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