public static boolean alterTable(String databaseName, String tableName, String fieldName, LogicalType type,
            String hiveVersion) {
        HiveConf hiveConf = HiveTableInlongFactory.getHiveConf();
        try (HiveMetastoreClientWrapper client = HiveMetastoreClientFactory.create(hiveConf, hiveVersion)) {
            FieldSchema schema = new FieldSchema(fieldName, flinkType2HiveType(type), "");

            Table table = client.getTable(databaseName, tableName);
            ObjectIdentifier identifier = createObjectIdentifier(databaseName, tableName);
            List<FieldSchema> fieldSchemaList = getTableFields(client, identifier);
            // remove partition keys
            fieldSchemaList.removeAll(table.getPartitionKeys());
            boolean alter = false;
            boolean exist = false;
            for (FieldSchema fieldSchema : fieldSchemaList) {
                if (fieldSchema.getName().equals(fieldName)) {
                    exist = true;
                    if (!flinkType2HiveType(type).equalsIgnoreCase(fieldSchema.getType())) {
                        LOG.info("table {}.{} field {} change type from {} to {}", databaseName, tableName, fieldName,
                                fieldSchema.getType(), flinkType2HiveType(type));
                        fieldSchema.setType(flinkType2HiveType(type));
                        alter = true;
                    }
                    break;
                }
            }
            if (!exist) {
                LOG.info("table {}.{} add new field {} with type {}", databaseName, tableName, fieldName,
                        flinkType2HiveType(type));
                fieldSchemaList.add(schema);
                alter = true;
            }
            if (alter) {
                table.getSd().setCols(fieldSchemaList);
                IMetaStoreClient metaStoreClient = getMetaStoreClient(client);
                EnvironmentContext environmentContext = new EnvironmentContext();
                environmentContext.putToProperties("CASCADE", "true");
                metaStoreClient.alter_table_with_environmentContext(databaseName, tableName, table, environmentContext);
                LOG.info("alter table {}.{} success", databaseName, tableName);
                return true;
            }
        } catch (TException e) {
            throw new CatalogException(String.format("Failed to alter hive table %s.%s", databaseName, tableName), e);
        } catch (NoSuchFieldException | IllegalAccessException e) {
            throw new RuntimeException(e);
        }
        return false;
    }
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