public void outputDirtyMetricsWithEstimate(String database, String schema, String table, long rowCount,
            long rowSize) {
        if (StringUtils.isBlank(database) || StringUtils.isBlank(schema) || StringUtils.isBlank(table)) {
            invokeDirty(rowCount, rowSize);
            return;
        }
        String identify = buildSchemaIdentify(database, schema, table);
        SinkMetricData subSinkMetricData;
        if (subSinkMetricMap.containsKey(identify)) {
            subSinkMetricData = subSinkMetricMap.get(identify);
        } else {
            subSinkMetricData = buildSubSinkMetricData(new String[]{database, schema, table}, this);
            subSinkMetricMap.put(identify, subSinkMetricData);
        }
        // sink metric and sub sink metric output metrics
        this.invokeDirty(rowCount, rowSize);
        subSinkMetricData.invokeDirty(rowCount, rowSize);
    }
output dirty metrics with estimate

@param database the database name of record
@param schema the schema name of record
@param table the table name of record
@param rowCount the row count of records
@param rowSize the row size of records
inlong-sort/sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/sub/SinkTableMetricData.java