public void process(byte[] originData, JsonNode data) {
        if (!schemaChange) {
            return;
        }
        String database;
        String table;
        try {
            database = dynamicSchemaFormat.parse(data, databasePattern);
            table = dynamicSchemaFormat.parse(data, tablePattern);
        } catch (Exception e) {
            if (exceptionPolicy == SchemaUpdateExceptionPolicy.THROW_WITH_STOP) {
                throw new SchemaChangeHandleException(
                        String.format("Parse database, table from origin data failed, origin data: %s",
                                new String(originData)),
                        e);
            }
            LOGGER.warn("Parse database, table from origin data failed, origin data: {}", new String(originData), e);
            if (exceptionPolicy == SchemaUpdateExceptionPolicy.LOG_WITH_IGNORE) {
                dirtySinkHelper.invoke(new String(originData), DirtyType.JSON_PROCESS_ERROR, e);
            }
            if (metricData != null) {
                metricData.invokeDirty(1, originData.length);
            }
            return;
        }
        Operation operation;
        try {
            JsonNode operationNode = Preconditions.checkNotNull(data.get("operation"),
                    "Operation node is null");
            operation = Preconditions.checkNotNull(
                    JsonDynamicSchemaFormat.OBJECT_MAPPER.convertValue(operationNode, new TypeReference<Operation>() {
                    }), "Operation is null");
        } catch (Exception e) {
            if (exceptionPolicy == SchemaUpdateExceptionPolicy.THROW_WITH_STOP) {
                throw new SchemaChangeHandleException(
                        String.format("Extract Operation from origin data failed,origin data: %s", data), e);
            }
            LOGGER.warn("Extract Operation from origin data failed,origin data: {}", data, e);
            handleDirtyData(data, originData, database, table, DirtyType.JSON_PROCESS_ERROR, e);
            return;
        }
        String originSchema = dynamicSchemaFormat.extractDDL(data);
        SchemaChangeType type = SchemaChangeUtils.extractSchemaChangeType(operation);
        if (type == null) {
            LOGGER.warn("Unsupported for schema-change: {}", originSchema);
            return;
        }
        switch (type) {
            case ALTER:
                handleAlterOperation(database, table, originData, originSchema, data, (AlterOperation) operation);
                break;
            case CREATE_TABLE:
                doCreateTable(originData, database, table, type, originSchema, data, (CreateTableOperation) operation);
                break;
            case DROP_TABLE:
                doDropTable(type, originSchema);
                break;
            case RENAME_TABLE:
                doRenameTable(type, originSchema);
                break;
            case TRUNCATE_TABLE:
                doTruncateTable(type, originSchema);
                break;
            default:
                LOGGER.warn("Unsupported for {}: {}", type, originSchema);
        }
    }
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