public static Set<SchemaChangeType> extractSchemaChangeTypes(Operation operation) {
        Set<SchemaChangeType> types = new HashSet<>();
        switch (operation.getOperationType()) {
            case ALTER:
                AlterOperation alterOperation = (AlterOperation) operation;
                Preconditions.checkState(alterOperation.getAlterColumns() != null
                        && !alterOperation.getAlterColumns().isEmpty(), "alter columns is empty");
                for (AlterColumn alterColumn : alterOperation.getAlterColumns()) {
                    extractSchemaChangeType(alterColumn, types);
                }
                break;
            case CREATE:
                types.add(SchemaChangeType.CREATE_TABLE);
                break;
            case TRUNCATE:
                types.add(SchemaChangeType.TRUNCATE_TABLE);
                break;
            case RENAME:
                types.add(SchemaChangeType.RENAME_TABLE);
                break;
            case DROP:
                types.add(SchemaChangeType.DROP_TABLE);
            default:
        }
        return types;
    }
Extract the schema change types from {@link Operation}

@param operation The operation
@return Set of {@link SchemaChangeType}
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/util/SchemaChangeUtils.java