Address Onder's review

REST catalog options and retry mechanism:
- Rename RestCatalogConnectionInfo to RestCatalogOptions throughout
- Eliminate CurrentRetryServerName static; pass opts directly through
  HttpRetryFn callback (void *context + List *headers) so the 419
  token-expired handler can force-refresh and patch the Authorization
  header in-place
- Fix double-free in GetRestCatalogAccessToken: null out entry fields
  before calling FetchRestCatalogAccessToken

Transaction-scoped state:
- Reject transactions that touch tables from different REST catalog
  servers
- Replace per-table rest catalog opts  deep-copy with a single
  PgLakeXactRestCatalogOpts static, deep-copied into
  TopTransactionContext on first use
- This avoids syscache lookups at XACT_EVENT_COMMIT time, which are
  forbidden (AssertCouldGetRelation fires during TRANS_COMMIT state)

Server configuration enforcement:
- Require TYPE 'rest' on CREATE SERVER ... FOREIGN DATA WRAPPER
  iceberg_catalog (reject NULL or non-rest types)
- Make FDW option names and auth type values case-insensitive
  (pg_strcasecmp), while keeping server names case-sensitive
- Make reserved catalog name checks (postgres, object_store, rest)
  case-insensitive via IsCatalogOwnedByExtension
- Support location_prefix server option, overriding the GUC default
- Accept user-created iceberg_catalog servers in default_catalog GUC

Tests:
- Add test_reject_modify_different_rest_catalogs_in_single_transaction
- Add test_server_location_prefix_overrides_guc
- Add tests for TYPE enforcement, case-sensitive server names, and
  default_catalog GUC with user-created servers
- Remove obsolete no-TYPE-defaults-to-rest tests

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

Author: sfc-gh-npuka

pg_lake_engine/include/pg_lake/util/string_utils.h

@@ -34,6 +34,7 @@ extern PGDLLEXPORT char *ReverseStringSearch(const char *haystack, const char *n
 extern PGDLLEXPORT int32_t AdjustAnyCharTypmod(int32_t typmod, int32_t newLength);
 extern PGDLLEXPORT int32_t GetAnyCharLengthFrom(int32_t typmod);
 extern PGDLLEXPORT bool PgStrcasecmpNullable(const char *a, const char *b);
+extern PGDLLEXPORT char *PstrdupRemoveTrailingSlash(const char *input);
 
 #define RangeVarQuoteIdentifier(rv) \
 	(((rv)->schemaname != NULL) ?							  \

pg_lake_engine/src/utils/catalog_type.c

@@ -108,8 +108,10 @@ HasReadOnlyOption(List *options)
 
 
 /*
- * IsCatalogOwnedByExtension returns true if the catalog name is one of the
- * extension-owned literals: 'rest', 'object_store', or 'postgres'.
+ * IsCatalogOwnedByExtension returns true if the catalog name matches one of
+ * the extension-owned names: 'rest', 'object_store', or 'postgres'.
+ * Comparison is case-insensitive so that "Postgres", "REST", etc. are
+ * also recognized as reserved.
  */
 bool
 IsCatalogOwnedByExtension(const char *catalog)
@@ -123,7 +125,7 @@ IsCatalogOwnedByExtension(const char *catalog)
 /*
  * IsRestCatalog returns true if the catalog name identifies a REST catalog.
  * This includes the extension-owned 'rest' literal and any user-created
- * iceberg_catalog server whose TYPE is 'rest' (or omitted, defaulting to 'rest').
+ * iceberg_catalog server whose TYPE is 'rest'.
  */
 bool
 IsRestCatalog(const char *catalog)
@@ -143,13 +145,9 @@ IsRestCatalog(const char *catalog)
 
 	ForeignDataWrapper *fdw = GetForeignDataWrapper(server->fdwid);
 
-	if (strcmp(fdw->fdwname, "iceberg_catalog") != 0)
+	if (strcmp(fdw->fdwname, ICEBERG_CATALOG_FDW_NAME) != 0)
 		return false;
 
-	/* Check server TYPE if set */
-	if (server->servertype != NULL && *server->servertype != '\0')
-		return pg_strcasecmp(server->servertype, "rest") == 0;
-
-	/* No TYPE specified, assume rest */
-	return true;
+	Assert(server->servertype != NULL && *server->servertype != '\0');
+	return pg_strcasecmp(server->servertype, REST_CATALOG_NAME) == 0;
 }

pg_lake_engine/src/utils/string_utils.c

@@ -261,3 +261,23 @@ PgStrcasecmpNullable(const char *a, const char *b)
 
 	return pg_strcasecmp(a, b) == 0;
 }
+
+
+/*
+ * PstrdupRemoveTrailingSlash returns a pstrdup'd copy of input with a
+ * trailing '/' removed, if present. Returns NULL when input is NULL.
+ */
+char *
+PstrdupRemoveTrailingSlash(const char *input)
+{
+	if (input == NULL)
+		return NULL;
+
+	char	   *result = pstrdup(input);
+	size_t		len = strlen(result);
+
+	if (len > 0 && result[len - 1] == '/')
+		result[len - 1] = '\0';
+
+	return result;
+}

pg_lake_iceberg/include/pg_lake/http/http_client.h

@@ -46,7 +46,7 @@ typedef struct
 extern bool HttpClientTraceTraffic;
 
 /* Callback function to determine if a request should be retried */
-typedef bool (*HttpRetryFn) (long status, int maxRetry, int retryNo);
+typedef bool (*HttpRetryFn) (long status, int maxRetry, int retryNo, void *context, List *headers);
 
 /* plain C API (no PostgreSQL types) */
 extern PGDLLEXPORT HttpResult HttpGet(const char *url, List *headers);
@@ -55,5 +55,6 @@ extern PGDLLEXPORT HttpResult HttpPost(const char *url, const char *body, List *
 extern PGDLLEXPORT HttpResult HttpDelete(const char *url, List *headers);
 extern PGDLLEXPORT HttpResult HttpPut(const char *url, const char *body, List *headers);
 extern PGDLLEXPORT HttpResult SendHttpRequestWithRetry(HttpMethod method, const char *url, const char *body,
-													   List *headers, HttpRetryFn retryFn, int maxRetry);
+													   List *headers, HttpRetryFn retryFn, int maxRetry,
+													   void *retryContext);
 extern PGDLLEXPORT int LinearBackoffSleepMs(int baseMs, int retryNo);

pg_lake_iceberg/include/pg_lake/rest_catalog/rest_catalog.h

@@ -36,21 +36,22 @@ extern int	RestCatalogAuthType;
 extern bool RestCatalogEnableVendedCredentials;
 
 /*
- * Holds per-server REST catalog connection settings. Populated from the
- * server options of an iceberg_catalog ForeignServer, with GUC fallback
- * for any option not explicitly set on the server.
+ * Holds per-server REST catalog options. Populated from the server options
+ * of an iceberg_catalog ForeignServer, with GUC fallback for any option
+ * not explicitly set on the server.
  */
-typedef struct RestCatalogConnectionInfo
+typedef struct RestCatalogOptions
 {
 	char	   *serverName;		/* server name, used for token cache keying */
 	char	   *host;
 	char	   *oauthHostPath;
 	char	   *clientId;
 	char	   *clientSecret;
 	char	   *scope;
+	char	   *locationPrefix;
 	int			authType;
 	bool		enableVendedCredentials;
-}			RestCatalogConnectionInfo;
+}			RestCatalogOptions;
 
 #define REST_CATALOG_AUTH_TOKEN_PATH "%s/api/catalog/v1/oauth/tokens"
 
@@ -96,31 +97,32 @@ typedef struct RestCatalogRequest
 #define GET_REST_CATALOG_METADATA_LOCATION "%s/api/catalog/v1/%s/namespaces/%s/tables/%s"
 
 /* Connection info resolution */
-extern PGDLLEXPORT RestCatalogConnectionInfo * GetRestCatalogConnectionFromServer(const char *serverName);
-extern PGDLLEXPORT RestCatalogConnectionInfo * GetRestCatalogConnectionForRelation(Oid relationId);
+extern PGDLLEXPORT RestCatalogOptions * GetRestCatalogOptionsFromServer(const char *serverName);
+extern PGDLLEXPORT RestCatalogOptions * GetRestCatalogOptionsForRelation(Oid relationId);
 
-extern PGDLLEXPORT void RegisterNamespaceToRestCatalog(RestCatalogConnectionInfo * conn, const char *catalogName, const char *namespaceName);
+extern PGDLLEXPORT void RegisterNamespaceToRestCatalog(RestCatalogOptions * opts, const char *catalogName, const char *namespaceName);
 extern PGDLLEXPORT void StartStageRestCatalogIcebergTableCreate(Oid relationId);
 extern PGDLLEXPORT char *FinishStageRestCatalogIcebergTableCreateRestRequest(Oid relationId, DataFileSchema * dataFileSchema, List *partitionSpecs);
-extern PGDLLEXPORT void ErrorIfRestNamespaceDoesNotExist(RestCatalogConnectionInfo * conn, const char *catalogName, const char *namespaceName);
+extern PGDLLEXPORT void ErrorIfRestNamespaceDoesNotExist(RestCatalogOptions * opts, const char *catalogName, const char *namespaceName);
 extern PGDLLEXPORT char *GetRestCatalogName(Oid relationId);
 extern PGDLLEXPORT char *GetRestCatalogNamespace(Oid relationId);
 extern PGDLLEXPORT char *GetRestCatalogTableName(Oid relationId);
 extern PGDLLEXPORT bool IsReadOnlyRestCatalogIcebergTable(Oid relationId);
-extern PGDLLEXPORT char *GetMetadataLocationFromRestCatalog(RestCatalogConnectionInfo * conn, const char *restCatalogName, const char *namespaceName,
+extern PGDLLEXPORT char *GetMetadataLocationFromRestCatalog(RestCatalogOptions * opts, const char *restCatalogName, const char *namespaceName,
 															const char *relationName);
 extern PGDLLEXPORT char *GetMetadataLocationForRestCatalogForIcebergTable(Oid relationId);
 extern PGDLLEXPORT void ReportHTTPError(HttpResult httpResult, int level);
-extern PGDLLEXPORT List *PostHeadersWithAuth(RestCatalogConnectionInfo * conn);
-extern PGDLLEXPORT List *DeleteHeadersWithAuth(RestCatalogConnectionInfo * conn);
-extern PGDLLEXPORT bool ShouldRetryRequestToRestCatalog(long status, int maxRetry, int retryNo);
-extern PGDLLEXPORT HttpResult SendRequestToRestCatalog(HttpMethod method, const char *url, const char *body, List *headers, const char *serverName);
+extern PGDLLEXPORT List *PostHeadersWithAuth(RestCatalogOptions * opts);
+extern PGDLLEXPORT List *DeleteHeadersWithAuth(RestCatalogOptions * opts);
+extern PGDLLEXPORT bool ShouldRetryRequestToRestCatalog(long status, int maxRetry, int retryNo, void *context, List *headers);
+extern PGDLLEXPORT HttpResult SendRequestToRestCatalog(HttpMethod method, const char *url, const char *body, List *headers, RestCatalogOptions * opts);
 extern PGDLLEXPORT RestCatalogRequest * GetAddSnapshotCatalogRequest(IcebergSnapshot * newSnapshot, Oid relationId);
 extern PGDLLEXPORT RestCatalogRequest * GetAddSchemaCatalogRequest(Oid relationId, DataFileSchema * dataFileSchema);
 extern PGDLLEXPORT RestCatalogRequest * GetSetCurrentSchemaCatalogRequest(Oid relationId, int32_t schemaId);
 extern PGDLLEXPORT RestCatalogRequest * GetAddPartitionCatalogRequest(Oid relationId, List *partitionSpec);
 extern PGDLLEXPORT RestCatalogRequest * GetSetPartitionDefaultIdCatalogRequest(Oid relationId, int specId);
 extern PGDLLEXPORT RestCatalogRequest * GetRemoveSnapshotCatalogRequest(List *removedSnapshotIds, Oid relationId);
 
-/* ProcessUtility handler: protects extension-owned catalog servers */
-extern PGDLLEXPORT bool BlockDDLOnExtensionCatalogs(ProcessUtilityParams *processUtilityParams, void *arg);
+/* ProcessUtility handlers for iceberg_catalog servers */
+extern PGDLLEXPORT bool BlockDDLOnExtensionCatalogs(ProcessUtilityParams * processUtilityParams, void *arg);
+extern PGDLLEXPORT bool RequireRestTypeForIcebergCatalogServer(ProcessUtilityParams * processUtilityParams, void *arg);

pg_lake_iceberg/src/http/http_client.c

@@ -277,7 +277,8 @@ CurlReturnError(CURL * curl, struct curl_slist *headerList,
  */
 HttpResult
 SendHttpRequestWithRetry(HttpMethod method, const char *url, const char *body,
-						 List *headers, HttpRetryFn retryFn, int maxRetry)
+						 List *headers, HttpRetryFn retryFn, int maxRetry,
+						 void *retryContext)
 {
 	Assert(maxRetry > 0);
 
@@ -287,7 +288,7 @@ SendHttpRequestWithRetry(HttpMethod method, const char *url, const char *body,
 	{
 		result = SendHttpRequest(method, url, body, headers);
 
-		if (retryFn != NULL && retryFn(result.status, maxRetry, retryNo))
+		if (retryFn != NULL && retryFn(result.status, maxRetry, retryNo, retryContext, headers))
 			continue;
 		else
 			break;

pg_lake_iceberg/src/iceberg/catalog.c

@@ -23,6 +23,7 @@
 #include "pg_lake/object_store_catalog/object_store_catalog.h"
 #include "pg_lake/rest_catalog/rest_catalog.h"
 #include "pg_lake/util/rel_utils.h"
+#include "pg_lake/util/string_utils.h"
 #include "pg_extension_base/spi_helpers.h"
 #include "catalog/namespace.h"
 #include "commands/dbcommands.h"
@@ -715,24 +716,7 @@ UpdateAllInternalIcebergTablesToReadOnly(void)
 char *
 GetIcebergDefaultLocationPrefix(void)
 {
-	if (IcebergDefaultLocationPrefix == NULL)
-	{
-		return NULL;
-	}
-
-	size_t		len = strlen(IcebergDefaultLocationPrefix);
-
-	if (len > 0 && IcebergDefaultLocationPrefix[len - 1] == '/')
-	{
-		/* remove trailing "/" */
-		char	   *locationPrefixRemovedTrailingSlash = pstrdup(IcebergDefaultLocationPrefix);
-
-		locationPrefixRemovedTrailingSlash[len - 1] = '\0';
-
-		return locationPrefixRemovedTrailingSlash;
-	}
-
-	return IcebergDefaultLocationPrefix;
+	return PstrdupRemoveTrailingSlash(IcebergDefaultLocationPrefix);
 }
 
 

pg_lake_iceberg/src/init.c

@@ -35,6 +35,8 @@
 #include "pg_lake/iceberg/operations/vacuum.h"
 #include "pg_lake/object_store_catalog/object_store_catalog.h"
 #include "pg_lake/rest_catalog/rest_catalog.h"
+#include "pg_lake/util/catalog_type.h"
+#include "access/xact.h"
 
 #define GUC_STANDARD 0
 
@@ -372,8 +374,16 @@ IcebergDefaultCatalogCheckHook(char **newvalue, void **extra, GucSource source)
 		pg_strcasecmp(newCatalog, OBJECT_STORE_CATALOG_NAME) == 0)
 		return true;
 
+	/*
+	 * When catalog access is available, also accept user-created
+	 * iceberg_catalog foreign servers with TYPE 'rest'.
+	 */
+	if (IsTransactionState() && IsRestCatalog(newCatalog))
+		return true;
+
 	GUC_check_errdetail("pg_lake_iceberg: allowed iceberg catalog options are '" POSTGRES_CATALOG_NAME "', "
-						" '" REST_CATALOG_NAME "' and '" OBJECT_STORE_CATALOG_NAME "'");
+						"'" REST_CATALOG_NAME "', '" OBJECT_STORE_CATALOG_NAME
+						"', or the name of a user-created iceberg_catalog server with TYPE 'rest'");
 
 	return false;
 }