private void doCreateTable(byte[] originData, String database, String table, SchemaChangeType type,
            String originSchema, JsonNode data, CreateTableOperation operation) {
        SchemaChangePolicy policy = policyMap.get(type);
        if (policy == SchemaChangePolicy.ENABLE) {
            try {
                List<String> primaryKeys = dynamicSchemaFormat.extractPrimaryKeyNames(data);
                String stmt = operationHelper.buildCreateTableStatement(database, table, primaryKeys, operation);
                boolean result = executeStatement(database, stmt);
                if (!result) {
                    LOGGER.error("Create table failed,statement: {}", stmt);
                    throw new IOException(String.format("Create table failed,statement: %s", stmt));
                }
                reportMetric(database, table, originData.length);
                return;
            } catch (Exception e) {
                if (exceptionPolicy == SchemaUpdateExceptionPolicy.THROW_WITH_STOP) {
                    throw new SchemaChangeHandleException(
                            String.format("Drop column failed, origin schema: %s", originSchema), e);
                }
                handleDirtyData(data, originData, database, table, DirtyType.CREATE_TABLE_ERROR, e);
                return;
            }
        }
        doSchemaChangeBase(type, policy, 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