protected void init(TableSchema schema) {
        if (!multipleSink) {
            validateTableStructure(schema);
        }
        String version = starrocksQueryVisitor.getStarRocksVersion();
        this.starrocksStreamLoadVisitor = new StarRocksStreamLoadVisitor(
                sinkOptions,
                null == schema ? new String[]{} : schema.getFieldNames(),
                version.length() > 0 && !version.trim().startsWith("1."));
    }
If a table writing throws exception, ignore it when receiving data later again
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/starrocks/src/main/java/org/apache/inlong/sort/starrocks/manager/StarRocksSinkManager.java