private void handleAlterOperation(String database, String table, byte[] originData,
            String originSchema, JsonNode data, AlterOperation operation) {
        if (operation.getAlterColumns() == null || operation.getAlterColumns().isEmpty()) {
            if (exceptionPolicy == SchemaUpdateExceptionPolicy.THROW_WITH_STOP) {
                throw new SchemaChangeHandleException(
                        String.format("Alter columns is empty, origin schema: %s", originSchema));
            }
            LOGGER.warn("Alter columns is empty, origin schema: {}", originSchema);
            return;
        }
        Map<SchemaChangeType, List<AlterColumn>> typeMap = new LinkedHashMap<>();
        for (AlterColumn alterColumn : operation.getAlterColumns()) {
            Set<SchemaChangeType> types = null;
            try {
                types = SchemaChangeUtils.extractSchemaChangeType(alterColumn);
                Preconditions.checkState(!types.isEmpty(), "Schema change types is empty");
            } catch (Exception e) {
                if (exceptionPolicy == SchemaUpdateExceptionPolicy.THROW_WITH_STOP) {
                    throw new SchemaChangeHandleException(
                            String.format("Extract schema change type failed, origin schema: %s", originSchema), e);
                }
                LOGGER.warn("Extract schema change type failed, origin schema: {}", originSchema, e);
            }
            if (types == null) {
                continue;
            }
            if (types.size() == 1) {
                SchemaChangeType type = types.stream().findFirst().get();
                typeMap.computeIfAbsent(type, k -> new ArrayList<>()).add(alterColumn);
            } else {
                // Handle change column, it only exists change column type and rename column in this scenario for now.
                for (SchemaChangeType type : types) {
                    SchemaChangePolicy policy = policyMap.get(type);
                    if (policy == SchemaChangePolicy.ENABLE) {
                        LOGGER.warn("Unsupported for {}: {}", type, originSchema);
                    } else {
                        doSchemaChangeBase(type, policy, originSchema);
                    }
                }
            }
        }
        if (!typeMap.isEmpty()) {
            doAlterOperation(database, table, originData, originSchema, data, typeMap);
        }
    }
Process schema change for Doris

@param data The origin data
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/doris/src/main/java/org/apache/inlong/sort/doris/schema/SchemaChangeHelper.java