private static void validateFlinkTable(CatalogBaseTable table) {
        Preconditions.checkArgument(
                table instanceof CatalogTable, "The Table should be a CatalogTable.");

        TableSchema schema = table.getSchema();
        schema
                .getTableColumns()
                .forEach(
                        column -> {
                            if (!FlinkCompatibilityUtil.isPhysicalColumn(column)) {
                                throw new UnsupportedOperationException(
                                        "Creating table with computed columns is not supported yet.");
                            }
                        });

        if (!schema.getWatermarkSpecs().isEmpty()) {
            throw new UnsupportedOperationException(
                    "Creating table with watermark specs is not supported yet.");
        }
    }
Append a new level to the base namespace
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkCatalog.java