More cleanup from Onder's latest review

Fail fast at statement time on cross-catalog DML. Rename
ValidateXactRestCatalog to BindRelationToXactRestCatalog and call it from
both FDW write paths (postgresBeginForeignModify and AddQueryResultToTable).
The function now binds the transaction to the relation's REST catalog on
the first DML and rejects any subsequent statement targeting a different
catalog, so the second INSERT errors out before any Parquet is written.
The pre-commit hook is kept as a belt-and-suspenders fallback for DDL
paths that reach the tracker without going through the new guard. The
regression test asserts the second INSERT (not COMMIT) raises.

Move ValidateIcebergCatalogServerDDL registration from pg_lake_table to
pg_lake_iceberg, where the handler is actually defined. Architecturally
this puts ownership of catalog-server DDL validation in the extension
that owns the catalog server abstraction.

Fix latent dangling-pointer in GetValidCatalogOptionsHint. The static
hint cache was allocated via initStringInfo in whatever short-lived
context the validator happened to be running in (typically MessageContext),
so the second invalid-option failure in the same backend returned freed
memory to errhint. Allocate in TopMemoryContext so the cache survives
transaction boundaries. The strengthened test_reject_unknown_server_option
now issues two failing CREATE SERVER statements on the same connection and
asserts the full option list appears in the hint on both attempts; the
previous version would not have caught this bug.

Also covers earlier review-driven hardening on the same branch: token
cache invalidation on ALTER SERVER credentials, ALTER SERVER rest_endpoint
blocking for all dependent REST iceberg tables (writable and read-only),
and DROP OWNED BY / concurrent DROP SERVER dependency tests.

Co-authored-by: Cursor <cursoragent@cursor.com>
Signed-off-by: sfc-gh-npuka <naisila.puka@snowflake.com>

Author: sfc-gh-npuka

pg_lake_iceberg/src/init.c

@@ -27,6 +27,7 @@
 #include "pg_lake/avro/avro_reader.h"
 #include "pg_lake/avro/avro_writer.h"
 #include "pg_lake/copy/copy_format.h"
+#include "pg_lake/ddl/utility_hook.h"
 #include "pg_lake/iceberg/api.h"
 #include "pg_lake/pgduck/numeric.h"
 #include "pg_lake/iceberg/catalog.h"
@@ -332,6 +333,8 @@ _PG_init(void)
 							 NULL, NULL, NULL);
 
 	AvroInit();
+
+	RegisterUtilityStatementHandler(ValidateIcebergCatalogServerDDL, NULL);
 }
 
 

pg_lake_iceberg/src/rest_catalog/rest_catalog.c

@@ -173,6 +173,12 @@ FindCatalogOptionDesc(const char *name)
 
 /*
  * Build the "Valid options are: ?" hint string.  Cached after first call.
+ *
+ * The cache must outlive any individual transaction: this helper is only
+ * called on validator error paths, and ereport(ERROR) immediately aborts
+ * the current transaction, freeing whatever short-lived context the
+ * validator was running under.  Allocate the buffer in TopMemoryContext
+ * so the static pointer remains valid for the lifetime of the backend.
  */
 static const char *
 GetValidCatalogOptionsHint(void)
@@ -181,8 +187,10 @@ GetValidCatalogOptionsHint(void)
 
 	if (hint == NULL)
 	{
+		MemoryContext oldcxt;
 		StringInfoData buf;
 
+		oldcxt = MemoryContextSwitchTo(TopMemoryContext);
 		initStringInfo(&buf);
 		appendStringInfoString(&buf, "Valid options are: ");
 		for (int i = 0; i < NUM_CATALOG_OPTIONS; i++)
@@ -193,6 +201,7 @@ GetValidCatalogOptionsHint(void)
 		}
 		appendStringInfoChar(&buf, '.');
 		hint = buf.data;
+		MemoryContextSwitchTo(oldcxt);
 	}
 
 	return hint;
@@ -334,13 +343,15 @@ iceberg_catalog_validator(PG_FUNCTION_ARGS)
 
 
 /*
- * ServerHasDependentWritableTable returns true if the given server
- * has at least one dependent writable iceberg table recorded in
- * pg_depend.  Used to block ALTER SERVER changes that would silently
- * break existing tables.
+ * ServerHasDependentRestIcebergTable returns true if the given server
+ * has at least one dependent REST-backed iceberg table (read-only or
+ * writable) recorded in pg_depend.  Used to block ALTER SERVER changes
+ * that would silently break existing tables, since both writable and
+ * read-only REST tables make REST API calls at runtime against the
+ * server's rest_endpoint.
  */
 static bool
-ServerHasDependentWritableTable(Oid serverOid)
+ServerHasDependentRestIcebergTable(Oid serverOid)
 {
 	Relation	depRel;
 	ScanKeyData key[2];
@@ -369,7 +380,9 @@ ServerHasDependentWritableTable(Oid serverOid)
 		if (depForm->classid != RelationRelationId)
 			continue;
 
-		if (GetIcebergCatalogType(depForm->objid) == REST_CATALOG_READ_WRITE)
+		IcebergCatalogType type = GetIcebergCatalogType(depForm->objid);
+
+		if (type == REST_CATALOG_READ_WRITE || type == REST_CATALOG_READ_ONLY)
 		{
 			found = true;
 			break;
@@ -488,9 +501,9 @@ ValidateIcebergCatalogServerDDL(ProcessUtilityParams * processUtilityParams,
 			return false;
 
 		/*
-		 * Changing rest_endpoint on a server with dependent writable tables
-		 * would silently point them at a different REST catalog, breaking the
-		 * metadata chain.
+		 * Changing rest_endpoint on a server with dependent iceberg tables
+		 * (writable or read-only) would silently point them at a different
+		 * REST catalog, breaking the metadata chain.
 		 */
 		ListCell   *lc;
 
@@ -499,12 +512,12 @@ ValidateIcebergCatalogServerDDL(ProcessUtilityParams * processUtilityParams,
 			DefElem    *def = (DefElem *) lfirst(lc);
 
 			if (pg_strcasecmp(def->defname, "rest_endpoint") == 0 &&
-				ServerHasDependentWritableTable(server->serverid))
+				ServerHasDependentRestIcebergTable(server->serverid))
 			{
 				ereport(ERROR,
 						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 						 errmsg("cannot change \"rest_endpoint\" on server \"%s\" "
-								"because it has dependent writable iceberg tables",
+								"because it has dependent iceberg tables",
 								stmt->servername),
 						 errhint("Drop the dependent tables first, or create a "
 								 "new server with the desired endpoint.")));
@@ -524,14 +537,16 @@ ValidateIcebergCatalogServerDDL(ProcessUtilityParams * processUtilityParams,
 static void
 ApplyGUCDefaults(RestCatalogOptions * opts)
 {
+	char	   *defaultLocationPrefix = GetIcebergDefaultLocationPrefix();
+
 	opts->host = RestCatalogHost ? pstrdup(RestCatalogHost) : NULL;
 	opts->oauthHostPath = RestCatalogOauthHostPath ? pstrdup(RestCatalogOauthHostPath) : NULL;
 	opts->clientId = RestCatalogClientId ? pstrdup(RestCatalogClientId) : NULL;
 	opts->clientSecret = RestCatalogClientSecret ? pstrdup(RestCatalogClientSecret) : NULL;
 	opts->scope = RestCatalogScope ? pstrdup(RestCatalogScope) : NULL;
 	opts->authType = RestCatalogAuthType;
 	opts->enableVendedCredentials = RestCatalogEnableVendedCredentials;
-	opts->locationPrefix = GetIcebergDefaultLocationPrefix();
+	opts->locationPrefix = defaultLocationPrefix ? pstrdup(defaultLocationPrefix) : NULL;
 }
 
 

pg_lake_table/include/pg_lake/transaction/track_iceberg_metadata_changes.h

@@ -61,4 +61,4 @@ extern PGDLLEXPORT void ResetRestCatalogRequests(void);
 extern PGDLLEXPORT HTAB *GetTrackedIcebergMetadataOperations(void);
 extern PGDLLEXPORT bool HasAnyTrackedIcebergMetadataChanges(void);
 extern PGDLLEXPORT bool IsIcebergTableCreatedInCurrentTransaction(Oid relation);
-extern PGDLLEXPORT void ValidateXactRestCatalog(Oid relationId);
+extern PGDLLEXPORT void BindRelationToXactRestCatalog(Oid relationId);

pg_lake_table/src/fdw/pg_lake_table.c

@@ -2206,7 +2206,7 @@ postgresBeginForeignModify(ModifyTableState *mtstate,
 	if (eflags & EXEC_FLAG_EXPLAIN_ONLY)
 		return;
 
-	ValidateXactRestCatalog(RelationGetRelid(resultRelInfo->ri_RelationDesc));
+	BindRelationToXactRestCatalog(RelationGetRelid(resultRelInfo->ri_RelationDesc));
 
 	/* Construct an execution state. */
 	fmstate = create_foreign_modify(resultRelInfo->ri_RelationDesc,

pg_lake_table/src/fdw/writable_table.c

@@ -1095,7 +1095,7 @@ AddQueryResultToTable(Oid relationId, char *readQuery, TupleDesc queryTupleDesc,
 {
 	Assert(queryTupleDesc != NULL && queryTupleDesc->natts > 0);
 
-	ValidateXactRestCatalog(relationId);
+	BindRelationToXactRestCatalog(relationId);
 
 	int64		rowsProcessed = 0;
 	ForeignTable *foreignTable = GetForeignTable(relationId);

pg_lake_table/src/init.c

@@ -42,7 +42,6 @@
 #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 "pg_lake/transaction/track_iceberg_metadata_changes.h"
@@ -383,7 +382,6 @@ _PG_init(void)
 
 	MarkGUCPrefixReserved(PG_LAKE_TABLE);
 
-	RegisterUtilityStatementHandler(ValidateIcebergCatalogServerDDL, NULL);
 	RegisterUtilityStatementHandler(ProcessVacuumPgLakeTable, NULL);
 	RegisterUtilityStatementHandler(ProcessCreatePgLakeTable, NULL);
 	RegisterUtilityStatementHandler(ProcessCreateAsSelectPgLakeTable, NULL);

pg_lake_table/src/transaction/track_iceberg_metadata_changes.c

@@ -609,40 +609,68 @@ InitRestCatalogRequestsHashIfNeeded(void)
 
 
 /*
- * ValidateXactRestCatalog is a fail-fast guard that prevents cross-catalog
- * DML within a single transaction.  It resolves the relation's catalog
- * identifier and, if a different catalog was already locked in for this
- * transaction, errors out immediately — before any Parquet data is written
- * to S3.
+ * BindRelationToXactRestCatalog binds the current transaction to the REST
+ * catalog associated with `relationId`, failing fast if a *different* REST
+ * catalog was already locked in for this transaction.
  *
- * No-ops for relations that are not REST-backed writable iceberg tables,
- * or when no catalog has been locked in yet (first DML in the xact).
+ * Semantics:
+ *   - For relations that are not REST-backed writable iceberg tables: no-op.
+ *   - For the first REST-backed write of the transaction: pre-resolves the
+ *     full catalog options and stashes them in TopTransactionContext, so
+ *     subsequent calls within the same transaction can be checked without
+ *     touching pg_foreign_server again at XACT_EVENT_COMMIT.
+ *   - For any subsequent REST-backed write whose catalog differs from the
+ *     locked-in one: raises ERRCODE_FEATURE_NOT_SUPPORTED *before* any
+ *     Parquet data is written to S3.
+ *
+ * Called at the top of every DML entry point that can mutate REST-backed
+ * iceberg tables: postgresBeginForeignModify() for row-by-row DML, and
+ * AddQueryResultToTable() for the INSERT...SELECT and COPY..FROM pushdown
+ * paths.  DDL paths (CREATE TABLE / DROP TABLE) reach the same protection
+ * indirectly via RecordRestCatalogRequestInTx(), which is invoked
+ * synchronously at statement time from the utility hook.
  */
 void
-ValidateXactRestCatalog(Oid relationId)
+BindRelationToXactRestCatalog(Oid relationId)
 {
 	if (!IsPgLakeIcebergForeignTableById(relationId) ||
 		GetIcebergCatalogType(relationId) != REST_CATALOG_READ_WRITE)
 		return;
 
-	if (PgLakeXactRestCatalog == NULL ||
-		PgLakeXactRestCatalog->catalogOpts == NULL)
-		return;
+	/*
+	 * Resolve the relation's catalog options up front.  We need the full
+	 * resolved struct (host, credentials, ...), not just the user-facing
+	 * identifier, because XACT_EVENT_PRE_COMMIT reuses these fields when
+	 * issuing the REST API requests and is not allowed to do syscache lookups
+	 * by then.
+	 */
+	RestCatalogOptions *resolvedOpts = GetRestCatalogOptionsForRelation(relationId);
 
-	char	   *catalog = GetStringOption(GetForeignTable(relationId)->options,
-										  "catalog", false);
+	InitRestCatalogRequestsHashIfNeeded();
 
-	if (catalog == NULL)
+	if (PgLakeXactRestCatalog->catalogOpts == NULL)
+	{
+		PgLakeXactRestCatalog->catalogOpts =
+			CopyRestCatalogOptions(TopTransactionContext, resolvedOpts);
 		return;
+	}
 
-	if (pg_strcasecmp(PgLakeXactRestCatalog->catalogOpts->catalog, catalog) != 0)
+	/*
+	 * Both sides of the comparison are the canonical catalog name (lowercase
+	 * "rest" for the built-in catalog, server name as stored in
+	 * pg_foreign_server for user-defined ones).  pg_strcasecmp matches the
+	 * casing rules PostgreSQL applies to identifier resolution.
+	 */
+	if (pg_strcasecmp(PgLakeXactRestCatalog->catalogOpts->catalog,
+					  resolvedOpts->catalog) != 0)
 		ereport(ERROR,
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg("cannot modify tables from different REST catalogs "
 						"in the same transaction"),
 				 errdetail("This transaction already targets catalog \"%s\", "
 						   "but the current statement targets \"%s\".",
-						   PgLakeXactRestCatalog->catalogOpts->catalog, catalog)));
+						   PgLakeXactRestCatalog->catalogOpts->catalog,
+						   resolvedOpts->catalog)));
 }
 
 
@@ -668,17 +696,26 @@ RecordRestCatalogRequestInTx(Oid relationId, RestCatalogOperationType operationT
 		/* Resolve the options for this relation's REST catalog */
 		RestCatalogOptions *resolvedOpts = GetRestCatalogOptionsForRelation(relationId);
 
+		/*
+		 * DDL paths (CREATE TABLE / DROP TABLE) call us directly from the
+		 * utility hook and may be the very first thing to touch a REST
+		 * catalog in this transaction, so this branch is still genuinely
+		 * reached.  DML paths reach us only from XACT_EVENT_PRE_COMMIT, by
+		 * which time BindRelationToXactRestCatalog() has already populated
+		 * catalogOpts.
+		 */
 		if (PgLakeXactRestCatalog->catalogOpts == NULL)
 		{
 			PgLakeXactRestCatalog->catalogOpts =
 				CopyRestCatalogOptions(TopTransactionContext, resolvedOpts);
 		}
 
 		/*
-		 * Belt-and-suspenders check.  All DML and DDL entry points already
-		 * call ValidateXactRestCatalog() at statement time, so in practice we
-		 * should never reach here with a mismatched catalog.  Kept as a last
-		 * line of defense for any future code path that forgets to do so.
+		 * Belt-and-suspenders check.  All DML and DDL entry points either
+		 * bind through BindRelationToXactRestCatalog() at statement time or
+		 * populate catalogOpts via the branch above, so in practice we never
+		 * reach here with a mismatched catalog.  Kept as a last line of
+		 * defense for any future code path that forgets to do so.
 		 */
 		else if (pg_strcasecmp(PgLakeXactRestCatalog->catalogOpts->catalog, resolvedOpts->catalog) != 0)
 			ereport(ERROR,

pg_lake_table/tests/pytests/test_iceberg_catalog_server.py

@@ -111,19 +111,43 @@ def test_create_server_horizon_auth(superuser_conn, extension):
 
 
 def test_reject_unknown_server_option(superuser_conn, extension):
-    """Unknown options should be rejected by the validator."""
-    err = run_command(
-        """
-        CREATE SERVER test_bad_opt TYPE 'rest'
-            FOREIGN DATA WRAPPER iceberg_catalog
-            OPTIONS (rest_endpoint 'http://localhost:8181', bogus_option 'x')
-        """,
-        superuser_conn,
-        raise_error=False,
-    )
-    assert "invalid option" in str(err)
-    assert "bogus_option" in str(err)
-    superuser_conn.rollback()
+    """
+    Unknown options should be rejected by the validator.
+
+    Issued twice on the same connection because the validator caches the
+    hint string in a static; the second call must hit the cached path and
+    must still produce a well-formed hint (regression guard against the
+    hint being palloc'd in a per-statement memory context).
+    """
+    EXPECTED_OPTIONS = [
+        "rest_endpoint",
+        "rest_auth_type",
+        "oauth_endpoint",
+        "scope",
+        "enable_vended_credentials",
+        "location_prefix",
+        "catalog_name",
+        "client_id",
+        "client_secret",
+    ]
+
+    for typo in ("bogus_option", "another_typo"):
+        err = run_command(
+            f"""
+            CREATE SERVER test_bad_opt_{typo} TYPE 'rest'
+                FOREIGN DATA WRAPPER iceberg_catalog
+                OPTIONS (rest_endpoint 'http://localhost:8181', {typo} 'x')
+            """,
+            superuser_conn,
+            raise_error=False,
+        )
+        msg = str(err)
+        assert "invalid option" in msg
+        assert typo in msg
+        assert "Valid options are:" in msg
+        for opt in EXPECTED_OPTIONS:
+            assert opt in msg, f"hint missing {opt!r} on attempt {typo!r}: {msg}"
+        superuser_conn.rollback()
 
 
 def test_reject_invalid_auth_type(superuser_conn, extension):