public static boolean changeSchema(RowType schema,
            String[] hiveColumns,
            DataType[] hiveTypes,
            String databaseName,
            String tableName,
            String hiveVersion) {
        boolean changed = false;

        Map<String, LogicalType> flinkTypeMap = new HashMap<>();
        for (RowField field : schema.getFields()) {
            // lowercase field name as Oracle field name is uppercase
            flinkTypeMap.put(field.getName().toLowerCase(), field.getType());
        }

        List<String> columnsFromData = new ArrayList<>(flinkTypeMap.keySet());
        List<String> columnsFromHive = Arrays.asList(hiveColumns);
        columnsFromData.removeAll(columnsFromHive);
        // add new field
        for (String fieldName : columnsFromData) {
            boolean result = alterTable(databaseName, tableName, fieldName, flinkTypeMap.get(fieldName), hiveVersion);
            if (result) {
                changed = true;
            }
        }
        // modify field type
        for (int i = 0; i < hiveColumns.length; i++) {
            if (!flinkTypeMap.containsKey(hiveColumns[i])) {
                // hive table has field which source table does not have, ignore it
                // so not support drop field dynamic now
                continue;
            }
            LogicalType flinkType = flinkTypeMap.get(hiveColumns[i]);
            if (hiveTypes[i].getLogicalType().getTypeRoot() != flinkType.getTypeRoot()) {
                // field type changed
                boolean result = alterTable(databaseName, tableName, hiveColumns[i], flinkType, hiveVersion);
                if (result) {
                    changed = true;
                }
            }
        }
        return changed;
    }
Utility class for list or create hive table
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/hive/src/main/java/org/apache/inlong/sort/hive/util/HiveTableUtil.java