@Override
    public void schemaEvolution(TaskWriterFactory<T> schema) throws IOException {
        emit(writer.complete());

        taskWriterFactory = (RowDataTaskWriterFactory) schema;
        taskWriterFactory.initialize(subTaskId, attemptId);
        writer = taskWriterFactory.create();
    }
check if the data is incremental phase
by checking the incremental field
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/multiple/IcebergSingleStreamWriter.java