private boolean isSchemaUpdate(RecordWithSchema recordWithSchema) {
        TableIdentifier tableId = recordWithSchema.getTableId();
        recordWithSchema.replaceSchema();
        if (multipleSchemas.get(tableId) != null
                && multipleSchemas.get(tableId).sameSchema(recordWithSchema.getSchema())) {
            return false;
        }
        LOG.info("Schema evolution with table {}, old schema: {}, new Schema: {}",
                tableId, multipleSchemas.get(tableId), recordWithSchema.getSchema());
        multipleSchemas.put(recordWithSchema.getTableId(), recordWithSchema.getSchema());
        return true;
    }
Iceberg writer that can distinguish different sink tables and route and distribute data into different
IcebergStreamWriter.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergMultipleStreamWriter.java