Address review part 1

- Block CREATE FOREIGN TABLE on iceberg_catalog servers. The
  iceberg_catalog FDW has no handler, so foreign tables created on it
  would fail at query time with "has no handler". The check is added to
  ErrorUnsupportedCreatePgLakeTableHandler in pg_lake_table, which
  already runs first for all CREATE FOREIGN TABLE statements.

- Block ALTER SERVER ... OWNER TO on extension-owned catalog servers
  (postgres, object_store, rest).

- Move ICEBERG_CATALOG_FDW_NAME from rest_catalog.h to catalog_type.h
  alongside the other catalog name constants, since it is referenced by
  both pg_lake_iceberg and pg_lake_table.

- Rename rest_auth_type value "default" to "oauth2" to better describe
  the standard OAuth2 client_credentials grant with Basic auth. "default"
  value is also kept.

- Rename ProtectExtensionCatalogServersHandler to
  BlockDDLOnExtensionCatalogs for clarity.

- Move BlockDDLOnExtensionCatalogs registration from pg_lake_iceberg
  init to pg_lake_table init, where all other ProcessUtility hooks are
  registered.

Signed-off-by: sfc-gh-npuka <naisila.puka@snowflake.com>

Author: sfc-gh-npuka

pg_lake_engine/include/pg_lake/util/catalog_type.h

@@ -17,6 +17,9 @@
 
 #pragma once
 
+/* FDW name for iceberg_catalog servers */
+#define ICEBERG_CATALOG_FDW_NAME "iceberg_catalog"
+
 /*
 * The allowed values for IcebergDefaultCatalog, case insensitive.
 */

pg_lake_iceberg/include/pg_lake/rest_catalog/rest_catalog.h

@@ -24,7 +24,7 @@
 #include "pg_lake/parquet/field.h"
 #include "pg_lake/iceberg/api/snapshot.h"
 
-#define REST_CATALOG_AUTH_TYPE_DEFAULT (0)
+#define REST_CATALOG_AUTH_TYPE_OAUTH2 (0)
 #define REST_CATALOG_AUTH_TYPE_HORIZON (1)
 
 extern PGDLLEXPORT char *RestCatalogHost;
@@ -99,9 +99,6 @@ typedef struct RestCatalogRequest
 extern PGDLLEXPORT RestCatalogConnectionInfo * GetRestCatalogConnectionFromServer(const char *serverName);
 extern PGDLLEXPORT RestCatalogConnectionInfo * GetRestCatalogConnectionForRelation(Oid relationId);
 
-/* FDW name for iceberg_catalog servers */
-#define ICEBERG_CATALOG_FDW_NAME "iceberg_catalog"
-
 extern PGDLLEXPORT void RegisterNamespaceToRestCatalog(RestCatalogConnectionInfo * conn, const char *catalogName, const char *namespaceName);
 extern PGDLLEXPORT void StartStageRestCatalogIcebergTableCreate(Oid relationId);
 extern PGDLLEXPORT char *FinishStageRestCatalogIcebergTableCreateRestRequest(Oid relationId, DataFileSchema * dataFileSchema, List *partitionSpecs);
@@ -126,4 +123,4 @@ extern PGDLLEXPORT RestCatalogRequest * GetSetPartitionDefaultIdCatalogRequest(O
 extern PGDLLEXPORT RestCatalogRequest * GetRemoveSnapshotCatalogRequest(List *removedSnapshotIds, Oid relationId);
 
 /* ProcessUtility handler: protects extension-owned catalog servers */
-extern PGDLLEXPORT bool ProtectExtensionCatalogServersHandler(ProcessUtilityParams *processUtilityParams, void *arg);
+extern PGDLLEXPORT bool BlockDDLOnExtensionCatalogs(ProcessUtilityParams *processUtilityParams, void *arg);

pg_lake_iceberg/src/init.c

@@ -59,7 +59,8 @@ void		_PG_init(void);
 
 /* pg_lake_iceberg.rest_catalog_auth_type */
 static const struct config_enum_entry RestCatalogAuthTypeOptions[] = {
-	{"default", REST_CATALOG_AUTH_TYPE_DEFAULT, false},
+	{"oauth2", REST_CATALOG_AUTH_TYPE_OAUTH2, false},
+	{"default", REST_CATALOG_AUTH_TYPE_OAUTH2, false},
 	{"horizon", REST_CATALOG_AUTH_TYPE_HORIZON, false},
 	{NULL, 0, false},
 };
@@ -256,7 +257,7 @@ _PG_init(void)
 							 gettext_noop("Determines the format for the initial OAuth token requests."),
 							 NULL,
 							 &RestCatalogAuthType,
-							 REST_CATALOG_AUTH_TYPE_DEFAULT,
+							 REST_CATALOG_AUTH_TYPE_OAUTH2,
 							 RestCatalogAuthTypeOptions,
 							 PGC_SUSET,
 							 GUC_SUPERUSER_ONLY | GUC_NO_SHOW_ALL | GUC_NOT_IN_SAMPLE,
@@ -329,8 +330,6 @@ _PG_init(void)
 							 NULL, NULL, NULL);
 
 	AvroInit();
-
-	RegisterUtilityStatementHandler(ProtectExtensionCatalogServersHandler, NULL);
 }
 
 

pg_lake_iceberg/src/rest_catalog/rest_catalog.c

@@ -56,7 +56,7 @@ char	   *RestCatalogOauthHostPath = "";
 char	   *RestCatalogClientId = NULL;
 char	   *RestCatalogClientSecret = NULL;
 char	   *RestCatalogScope = "PRINCIPAL_ROLE:ALL";
-int			RestCatalogAuthType = REST_CATALOG_AUTH_TYPE_DEFAULT;
+int			RestCatalogAuthType = REST_CATALOG_AUTH_TYPE_OAUTH2;
 bool		RestCatalogEnableVendedCredentials = true;
 
 /*
@@ -157,11 +157,13 @@ iceberg_catalog_validator(PG_FUNCTION_ARGS)
 		{
 			char	   *authType = defGetString(def);
 
-			if (strcmp(authType, "default") != 0 && strcmp(authType, "horizon") != 0)
+			if (strcmp(authType, "oauth2") != 0 &&
+				strcmp(authType, "default") != 0 &&
+				strcmp(authType, "horizon") != 0)
 				ereport(ERROR,
 						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 						 errmsg("invalid rest_auth_type option: \"%s\"", authType),
-						 errhint("Valid values are \"default\" and \"horizon\".")));
+						 errhint("Valid values are \"oauth2\" and \"horizon\".")));
 		}
 		else if (strcmp(def->defname, "enable_vended_credentials") == 0)
 		{
@@ -192,7 +194,7 @@ IsIcebergCatalogServer(const char *serverName)
 
 
 /*
- * ProtectExtensionCatalogServersHandler guards the extension-owned
+ * BlockDDLOnExtensionCatalogs guards the extension-owned
  * iceberg_catalog servers (postgres, object_store, rest) against
  * unauthorized DDL.
  *
@@ -202,9 +204,10 @@ IsIcebergCatalogServer(const char *serverName)
  *  - ALTER SERVER on 'rest' is allowed (users may set options).
  *  - DROP SERVER on 'postgres', 'object_store', or 'rest' is blocked.
  *  - ALTER ... RENAME on 'postgres', 'object_store', or 'rest' is blocked.
+ *  - ALTER ... OWNER TO on 'postgres', 'object_store', or 'rest' is blocked.
  */
 bool
-ProtectExtensionCatalogServersHandler(ProcessUtilityParams *processUtilityParams,
+BlockDDLOnExtensionCatalogs(ProcessUtilityParams *processUtilityParams,
 									  void *arg)
 {
 	Node	   *parsetree = processUtilityParams->plannedStmt->utilityStmt;
@@ -286,6 +289,24 @@ ProtectExtensionCatalogServersHandler(ProcessUtilityParams *processUtilityParams
 					 errmsg("cannot rename the extension-owned \"%s\" catalog server",
 							serverName)));
 	}
+	else if (IsA(parsetree, AlterOwnerStmt))
+	{
+		AlterOwnerStmt *stmt = (AlterOwnerStmt *) parsetree;
+
+		if (stmt->objectType != OBJECT_FOREIGN_SERVER)
+			return false;
+
+		char	   *serverName = strVal(stmt->object);
+
+		if (!IsIcebergCatalogServer(serverName))
+			return false;
+
+		if (IsCatalogOwnedByExtension(serverName))
+			ereport(ERROR,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("cannot change owner of the extension-owned \"%s\" catalog server",
+							serverName)));
+	}
 
 	return false;
 }
@@ -347,7 +368,7 @@ GetRestCatalogConnectionFromServer(const char *serverName)
 
 			conn->authType = (strcmp(authType, "horizon") == 0)
 				? REST_CATALOG_AUTH_TYPE_HORIZON
-				: REST_CATALOG_AUTH_TYPE_DEFAULT;
+				: REST_CATALOG_AUTH_TYPE_OAUTH2;
 		}
 		else if (strcmp(def->defname, "oauth_endpoint") == 0)
 			conn->oauthHostPath = defGetString(def);

pg_lake_table/src/ddl/create_table.c

@@ -90,6 +90,7 @@ static bool IsJsonOrCSVBackedTable(PgLakeTableType tableType, List *options);
 static void ErrorIfUnsupportedColumnTypeForJsonOrCSVTables(List *columnDefList);
 static void ErrorIfUsingGeometryWithoutSpatialAnalytics(List *columnDefList);
 static void ErrorIfUnsupportedLakeTable(CreateForeignTableStmt *createStmt);
+static void ErrorIfCreateForeignTableOnIcebergCatalog(CreateForeignTableStmt *createStmt);
 static void ErrorIfWritableTableWithReservedColumnName(List *columnDefList, PgLakeTableType tableType);
 static void ErrorIfInvalidFilenameColumn(List *columnDefList);
 static bool IsConflictingColumnNameForReadParquet(const char *columnName);
@@ -324,6 +325,9 @@ ErrorIfUsingGeometryWithoutSpatialAnalytics(List *columnDefList)
 *
 * We check for unsupported features in the table definition, such as unsupported URLs or unsupported
 * combinations such as writable tables without column definitions.
+*
+* Also blocks CREATE FOREIGN TABLE on iceberg_catalog servers, which have no
+* handler. Tables should be created via CREATE TABLE ... USING iceberg instead.
 */
 bool
 ErrorUnsupportedCreatePgLakeTableHandler(ProcessUtilityParams * params, void *arg)
@@ -339,6 +343,8 @@ ErrorUnsupportedCreatePgLakeTableHandler(ProcessUtilityParams * params, void *ar
 	CreateForeignTableStmt *createStmt =
 		(CreateForeignTableStmt *) plannedStmt->utilityStmt;
 
+	ErrorIfCreateForeignTableOnIcebergCatalog(createStmt);
+
 	if (!IsCreateLakeTable(createStmt))
 	{
 		/* not a lake table */
@@ -351,6 +357,31 @@ ErrorUnsupportedCreatePgLakeTableHandler(ProcessUtilityParams * params, void *ar
 }
 
 
+/*
+ * ErrorIfCreateForeignTableOnIcebergCatalog blocks CREATE FOREIGN TABLE
+ * when the target server uses the iceberg_catalog FDW, which has no handler.
+ */
+static void
+ErrorIfCreateForeignTableOnIcebergCatalog(CreateForeignTableStmt *createStmt)
+{
+	ForeignServer *server =
+		GetForeignServerByName(createStmt->servername, true);
+
+	if (server == NULL)
+		return;
+
+	ForeignDataWrapper *fdw = GetForeignDataWrapper(server->fdwid);
+
+	if (strcmp(fdw->fdwname, ICEBERG_CATALOG_FDW_NAME) == 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("cannot create foreign tables on iceberg_catalog server \"%s\"",
+						createStmt->servername),
+				 errhint("Use CREATE TABLE ... USING iceberg WITH (catalog = '%s') instead.",
+						 createStmt->servername)));
+}
+
+
 /*
 * ErrorIfUnsupportedLakeTable is a helper function for checking unsupported features
 * in CREATE FOREIGN TABLE statements that are pg_lake tables.

pg_lake_table/src/init.c

@@ -42,6 +42,7 @@
 #include "pg_lake/planner/insert_select.h"
 #include "pg_lake/planner/query_pushdown.h"
 #include "pg_lake/util/s3_file_utils.h"
+#include "pg_lake/rest_catalog/rest_catalog.h"
 #include "pg_lake/test/hide_lake_objects.h"
 #include "pg_lake/transaction/transaction_hooks.h"
 #include "utils/guc.h"
@@ -350,6 +351,7 @@ _PG_init(void)
 
 	MarkGUCPrefixReserved(PG_LAKE_TABLE);
 
+	RegisterUtilityStatementHandler(BlockDDLOnExtensionCatalogs, NULL);
 	RegisterUtilityStatementHandler(ProcessVacuumPgLakeTable, NULL);
 	RegisterUtilityStatementHandler(ProcessCreatePgLakeTable, NULL);
 	RegisterUtilityStatementHandler(ProcessCreateAsSelectPgLakeTable, NULL);

pg_lake_table/tests/pytests/test_iceberg_catalog_server.py

@@ -57,7 +57,7 @@ def test_create_rest_server_with_all_options(superuser_conn, extension):
             FOREIGN DATA WRAPPER iceberg_catalog
             OPTIONS (
                 rest_endpoint 'http://localhost:8181',
-                rest_auth_type 'default',
+                rest_auth_type 'oauth2',
                 oauth_endpoint 'http://localhost:8181/oauth/tokens',
                 scope 'PRINCIPAL_ROLE:ALL',
                 enable_vended_credentials 'true',
@@ -135,12 +135,12 @@ def test_reject_unknown_server_option(superuser_conn, extension):
 
 
 def test_reject_invalid_auth_type(superuser_conn, extension):
-    """Only 'default' and 'horizon' are valid for rest_auth_type."""
+    """Only 'oauth2', 'default', and 'horizon' are valid for rest_auth_type."""
     err = run_command(
         """
         CREATE SERVER test_bad_auth TYPE 'rest'
             FOREIGN DATA WRAPPER iceberg_catalog
-            OPTIONS (rest_endpoint 'http://localhost:8181', rest_auth_type 'oauth2')
+            OPTIONS (rest_endpoint 'http://localhost:8181', rest_auth_type 'basic')
         """,
         superuser_conn,
         raise_error=False,
@@ -177,38 +177,23 @@ def test_reject_options_on_non_server(superuser_conn, extension):
     superuser_conn.rollback()
 
 
-# ── Creating foreign tables on iceberg_catalog should fail ─────────────────
+# ── CREATE FOREIGN TABLE on iceberg_catalog servers is blocked ──────────────
 
 
-def test_cannot_query_foreign_table_on_catalog_server(superuser_conn, extension):
-    """iceberg_catalog has no handler, so querying a foreign table should fail.
-
-    PostgreSQL allows CREATE FOREIGN TABLE on a handler-less FDW; the error
-    only surfaces at query time when GetFdwRoutineByServerId() is called.
-    """
-    run_command(
-        """
-        CREATE SERVER test_ft_server TYPE 'rest'
-            FOREIGN DATA WRAPPER iceberg_catalog
-            OPTIONS (rest_endpoint 'http://localhost:8181')
-        """,
-        superuser_conn,
-    )
-
-    run_command(
+def test_reject_create_foreign_table_on_iceberg_catalog_server(
+    superuser_conn, extension
+):
+    """CREATE FOREIGN TABLE on an iceberg_catalog server is blocked."""
+    err = run_command(
         """
-        CREATE FOREIGN TABLE test_ft_table (id int)
-            SERVER test_ft_server
+        CREATE FOREIGN TABLE test_ft_pg (id int)
+            SERVER postgres
         """,
         superuser_conn,
-    )
-
-    err = run_command(
-        "SELECT * FROM test_ft_table",
-        superuser_conn,
         raise_error=False,
     )
-    assert "has no handler" in str(err)
+    assert err is not None
+    assert "cannot create foreign tables on iceberg_catalog server" in str(err)
     superuser_conn.rollback()
 
 
@@ -631,6 +616,50 @@ def test_reject_rename_rest_server(superuser_conn, extension):
     superuser_conn.rollback()
 
 
+def test_reject_owner_change_postgres_server(superuser_conn, extension):
+    """ALTER SERVER ... OWNER TO on the extension-owned 'postgres' server is blocked."""
+    err = run_command(
+        "ALTER SERVER postgres OWNER TO CURRENT_USER",
+        superuser_conn,
+        raise_error=False,
+    )
+    assert err is not None
+    assert (
+        'cannot change owner of the extension-owned "postgres" catalog server'
+        in str(err)
+    )
+    superuser_conn.rollback()
+
+
+def test_reject_owner_change_object_store_server(superuser_conn, extension):
+    """ALTER SERVER ... OWNER TO on the extension-owned 'object_store' server is blocked."""
+    err = run_command(
+        "ALTER SERVER object_store OWNER TO CURRENT_USER",
+        superuser_conn,
+        raise_error=False,
+    )
+    assert err is not None
+    assert (
+        'cannot change owner of the extension-owned "object_store" catalog server'
+        in str(err)
+    )
+    superuser_conn.rollback()
+
+
+def test_reject_owner_change_rest_server(superuser_conn, extension):
+    """ALTER SERVER ... OWNER TO on the extension-owned 'rest' server is blocked."""
+    err = run_command(
+        "ALTER SERVER rest OWNER TO CURRENT_USER",
+        superuser_conn,
+        raise_error=False,
+    )
+    assert err is not None
+    assert 'cannot change owner of the extension-owned "rest" catalog server' in str(
+        err
+    )
+    superuser_conn.rollback()
+
+
 def test_allow_drop_user_created_server(superuser_conn, extension):
     """DROP SERVER on a user-created server should work fine."""
     run_command(
@@ -659,3 +688,20 @@ def test_allow_rename_user_created_server(superuser_conn, extension):
         "ALTER SERVER user_rename_srv RENAME TO user_renamed_srv", superuser_conn
     )
     superuser_conn.rollback()
+
+
+def test_allow_owner_change_user_created_server(superuser_conn, extension):
+    """ALTER SERVER ... OWNER TO on a user-created server should work fine."""
+    run_command(
+        """
+        CREATE SERVER user_owner_srv TYPE 'rest'
+            FOREIGN DATA WRAPPER iceberg_catalog
+            OPTIONS (rest_endpoint 'http://localhost:8181')
+        """,
+        superuser_conn,
+    )
+    run_command(
+        "ALTER SERVER user_owner_srv OWNER TO CURRENT_USER",
+        superuser_conn,
+    )
+    superuser_conn.rollback()