private void doAlterOperation(String database, String table, byte[] originData, String originSchema, JsonNode data,
            Map<SchemaChangeType, List<AlterColumn>> typeMap) {
        StringJoiner joiner = new StringJoiner(",");
        for (Entry<SchemaChangeType, List<AlterColumn>> kv : typeMap.entrySet()) {
            SchemaChangePolicy policy = policyMap.get(kv.getKey());
            doSchemaChangeBase(kv.getKey(), policy, originSchema);
            if (policy == SchemaChangePolicy.ENABLE) {
                String alterStatement = null;
                try {
                    switch (kv.getKey()) {
                        case ADD_COLUMN:
                            alterStatement = doAddColumn(kv.getValue());
                            break;
                        case DROP_COLUMN:
                            alterStatement = doDropColumn(kv.getValue());
                            break;
                        case RENAME_COLUMN:
                            alterStatement = doRenameColumn(kv.getKey(), originSchema);
                            break;
                        case CHANGE_COLUMN_TYPE:
                            alterStatement = doChangeColumnType(kv.getKey(), originSchema);
                            break;
                        default:
                    }
                } catch (Exception e) {
                    if (exceptionPolicy == SchemaUpdateExceptionPolicy.THROW_WITH_STOP) {
                        throw new SchemaChangeHandleException(
                                String.format("Build alter statement failed, origin schema: %s", originSchema), e);
                    }
                    LOGGER.warn("Build alter statement failed, origin schema: {}", originSchema, e);
                }
                if (alterStatement != null) {
                    joiner.add(alterStatement);
                }
            }
        }
        String statement = joiner.toString();
        if (statement.length() != 0) {
            try {
                String alterStatementCommon = operationHelper.buildAlterStatementCommon(database, table);
                statement = alterStatementCommon + statement;
                // The checkLightSchemaChange is removed because most scenarios support it
                boolean result = executeStatement(database, statement);
                if (!result) {
                    LOGGER.error("Alter table failed,statement: {}", statement);
                    throw new SchemaChangeHandleException(String.format("Add column failed,statement: %s", statement));
                }
                LOGGER.info("Alter table success,statement: {}", statement);
                reportMetric(database, table, originData.length);
            } catch (Exception e) {
                if (exceptionPolicy == SchemaUpdateExceptionPolicy.THROW_WITH_STOP) {
                    throw new SchemaChangeHandleException(
                            String.format("Alter table failed, origin schema: %s", originSchema), e);
                }
                handleDirtyData(data, originData, database, table, DirtyType.HANDLE_ALTER_TABLE_ERROR, e);
            }
        }
    }
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