Make read_only a server option; Support TYPE 'object_store' servers

Allow CREATE SERVER ... TYPE 'object_store' FOREIGN DATA WRAPPER
iceberg_catalog, mirroring the existing TYPE 'rest' pattern. Each
server can carry its own location_prefix and read_only options,
enabling shared writer/reader topologies where multiple Postgres
instances point at the same S3 path.

Key changes:
- ValidateIcebergCatalogServerDDL now accepts TYPE 'object_store'
  (TYPE 'postgres' remains rejected).
- Add IsObjectStoreCatalog() in catalog_type.c, paralleling
  IsRestCatalog(); update HasObjectStoreCatalogTableOption to use it
  so that tables with catalog='<named_server>' resolve correctly.
- Add ObjectStoreCatalogOptions struct and resolution function that
  reads server options with GUC fallback.
- Add read_only as a server-level option (valid for both REST and
  object_store servers). Server read_only propagates to all tables
  unless overridden; table read_only='false' on a read_only server
  is an error.
- Update create_table.c to resolve location_prefix from named
  object_store servers instead of requiring the GUC.
- Update IcebergDefaultCatalogCheckHook to accept named object_store
  servers.
- Update catalog export SPI query to include tables on named
  object_store servers.

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

Author: sfc-gh-npuka

pg_lake_engine/include/pg_lake/util/catalog_type.h

@@ -66,3 +66,4 @@ extern PGDLLEXPORT bool HasObjectStoreCatalogTableOption(List *options);
 extern PGDLLEXPORT bool HasReadOnlyOption(List *options);
 extern PGDLLEXPORT bool IsCatalogOwnedByExtension(const char *catalog);
 extern PGDLLEXPORT bool IsRestCatalog(const char *catalog);
+extern PGDLLEXPORT bool IsObjectStoreCatalog(const char *catalog);

pg_lake_engine/src/utils/catalog_type.c

@@ -82,15 +82,16 @@ HasRestCatalogTableOption(List *options)
 
 
 /*
- * HasObjectStoreCatalogTableOption returns true if the options contain
- * catalog='object_store'.
+ * HasObjectStoreCatalogTableOption returns true if the catalog option
+ * indicates an object store catalog: either the literal 'object_store'
+ * or a named iceberg_catalog server with TYPE 'object_store'.
  */
 bool
 HasObjectStoreCatalogTableOption(List *options)
 {
 	char	   *catalog = GetStringOption(options, "catalog", false);
 
-	return catalog ? pg_strcasecmp(catalog, OBJECT_STORE_CATALOG_NAME) == 0 : false;
+	return IsObjectStoreCatalog(catalog);
 }
 
 
@@ -148,6 +149,35 @@ IsRestCatalog(const char *catalog)
 	if (strcmp(fdw->fdwname, ICEBERG_CATALOG_FDW_NAME) != 0)
 		return false;
 
-	Assert(pg_strcasecmp(server->servertype, REST_CATALOG_NAME) == 0);
-	return true;
+	return pg_strcasecmp(server->servertype, REST_CATALOG_NAME) == 0;
+}
+
+
+/*
+ * IsObjectStoreCatalog returns true if the catalog name identifies an
+ * object store catalog.  This includes the built-in 'object_store'
+ * literal and any user-created iceberg_catalog server whose TYPE is
+ * 'object_store'.
+ */
+bool
+IsObjectStoreCatalog(const char *catalog)
+{
+	if (catalog == NULL)
+		return false;
+
+	if (pg_strcasecmp(catalog, OBJECT_STORE_CATALOG_NAME) == 0)
+		return true;
+
+	bool		missingOK = true;
+	ForeignServer *server = GetForeignServerByName(catalog, missingOK);
+
+	if (server == NULL)
+		return false;
+
+	ForeignDataWrapper *fdw = GetForeignDataWrapper(server->fdwid);
+
+	if (strcmp(fdw->fdwname, ICEBERG_CATALOG_FDW_NAME) != 0)
+		return false;
+
+	return pg_strcasecmp(server->servertype, OBJECT_STORE_CATALOG_NAME) == 0;
 }

pg_lake_iceberg/include/pg_lake/object_store_catalog/object_store_catalog.h

@@ -2,15 +2,28 @@
 
 #include "postgres.h"
 
-#pragma once
-
 /* crunchy_iceberg.enable_object_store_catalog setting */
 extern PGDLLEXPORT bool EnableObjectStoreCatalog;
 
 extern PGDLLEXPORT char *ObjectStoreCatalogLocationPrefix;
 extern PGDLLEXPORT char *ExternalObjectStorePrefix;
 extern PGDLLEXPORT char *InternalObjectStorePrefix;
 
+/*
+ * Resolved object store catalog options.  For the built-in 'object_store'
+ * catalog the fields come from GUC settings.  For user-created servers
+ * (CREATE SERVER ... TYPE 'object_store') the server options override the
+ * GUC defaults.
+ */
+typedef struct ObjectStoreCatalogOptions
+{
+	char	   *catalog;		/* server name or "object_store" */
+	char	   *locationPrefix; /* object store catalog location prefix */
+	bool		readOnly;		/* server-level read_only default */
+}			ObjectStoreCatalogOptions;
+
+extern PGDLLEXPORT ObjectStoreCatalogOptions * GetObjectStoreCatalogOptionsFromCatalog(const char *catalog);
+
 extern PGDLLEXPORT void InitObjectStoreCatalog(void);
 extern PGDLLEXPORT void ExportIcebergCatalogIfChanged(void);
 extern PGDLLEXPORT const char *GetObjectStoreDefaultLocationPrefix(void);

pg_lake_iceberg/pg_lake_iceberg--3.2--3.3.sql

@@ -18,21 +18,29 @@ CREATE OR REPLACE VIEW pg_catalog.iceberg_tables AS
 /*
  * iceberg_catalog foreign data wrapper: allows defining named catalog
  * configurations via CREATE SERVER so that users are not limited to a
- * single global REST catalog configured through GUC settings.
+ * single global catalog configured through GUC settings.
  *
- * Server options (non-secret): rest_endpoint, rest_auth_type,
- *   oauth_endpoint, scope, enable_vended_credentials, location_prefix,
- *   catalog_name.
- * User mapping options (credentials): client_id, client_secret, scope.
+ * Supported server types:
+ *   TYPE 'rest'          -- REST catalog (e.g. Polaris, Gravitino)
+ *   TYPE 'object_store'  -- Object store catalog (catalog.json in S3)
+ *
+ * Server options (all optional):
+ *   rest_endpoint, rest_auth_type, oauth_endpoint, scope,
+ *   enable_vended_credentials, location_prefix, catalog_name,
+ *   read_only (boolean).
+ *
+ * User mapping options (credentials, TYPE 'rest' only):
+ *   client_id, client_secret, scope.
  *
  * scope is accepted in both server and user mapping; user mapping wins.
+ * read_only on a server propagates to all tables unless overridden.
  *
- * Credential resolution order:
+ * Credential resolution order (TYPE 'rest'):
  *   1. CREATE USER MAPPING for the current user
  *   2. $PGDATA/catalogs.conf (platform-provided)
  *   3. GUC variables (backward compatibility)
  *
- * User-defined catalog example:
+ * REST catalog example:
  *   CREATE SERVER my_polaris TYPE 'rest'
  *     FOREIGN DATA WRAPPER iceberg_catalog
  *     OPTIONS (rest_endpoint 'https://polaris.example.com');
@@ -42,14 +50,19 @@ CREATE OR REPLACE VIEW pg_catalog.iceberg_tables AS
  *
  *   CREATE TABLE t (a int) USING iceberg WITH (catalog = 'my_polaris');
  *
- * Platform-provided catalog example:
- *   CREATE SERVER horizon TYPE 'rest'
+ * Object store catalog example (shared writer/reader):
+ *   -- Writer instance:
+ *   CREATE SERVER shared_catalog TYPE 'object_store'
  *     FOREIGN DATA WRAPPER iceberg_catalog
- *     OPTIONS (rest_endpoint 'https://horizon.example.com');
+ *     OPTIONS (location_prefix 's3://bucket/shared');
  *
- *   -- Credentials in $PGDATA/catalogs.conf:
- *   --   horizon.client_id = 'platform_id'
- *   --   horizon.client_secret = 'platform_secret'
+ *   CREATE TABLE t (a int) USING iceberg
+ *     WITH (catalog = 'shared_catalog');
+ *
+ *   -- Reader instance (same S3 path, read-only):
+ *   CREATE SERVER shared_catalog TYPE 'object_store'
+ *     FOREIGN DATA WRAPPER iceberg_catalog
+ *     OPTIONS (location_prefix 's3://bucket/shared', read_only 'true');
  */
 CREATE FUNCTION lake_iceberg.iceberg_catalog_validator(text[], oid)
 RETURNS void

pg_lake_iceberg/src/init.c

@@ -386,14 +386,15 @@ IcebergDefaultCatalogCheckHook(char **newvalue, void **extra, GucSource source)
 
 	/*
 	 * When catalog access is available, also accept user-created
-	 * iceberg_catalog foreign servers with TYPE 'rest'.
+	 * iceberg_catalog foreign servers with TYPE 'rest' or 'object_store'.
 	 */
-	if (IsTransactionState() && IsRestCatalog(newCatalog))
+	if (IsTransactionState() &&
+		(IsRestCatalog(newCatalog) || IsObjectStoreCatalog(newCatalog)))
 		return true;
 
 	GUC_check_errdetail("pg_lake_iceberg: allowed iceberg catalog options are '" POSTGRES_CATALOG_NAME "', "
 						"'" REST_CATALOG_NAME "', '" OBJECT_STORE_CATALOG_NAME
-						"', or the name of a user-created iceberg_catalog server with TYPE 'rest'");
+						"', or the name of a user-created iceberg_catalog server");
 
 	return false;
 }

pg_lake_iceberg/src/object_store_catalog/object_store_catalog.c

@@ -3,6 +3,7 @@
 #include "miscadmin.h"
 
 #include "commands/dbcommands.h"
+#include "commands/defrem.h"
 #include "foreign/foreign.h"
 #include "utils/inval.h"
 #include "utils/snapmgr.h"
@@ -21,6 +22,7 @@
 #include "pg_lake/extensions/pg_lake_iceberg.h"
 #include "pg_lake/storage/local_storage.h"
 #include "pg_lake/util/string_utils.h"
+#include "pg_lake/util/catalog_type.h"
 
 char	   *ObjectStoreCatalogLocationPrefix = NULL;
 char	   *ExternalObjectStorePrefix = "fromsf";
@@ -321,6 +323,11 @@ PushMetadataLocationToObjectStoreCatalog(void)
 
 	StringInfo	fetchObjectStoreMetadata = makeStringInfo();
 
+	/*
+	 * Match tables whose catalog option is either the built-in
+	 * 'object_store' or any user-created iceberg_catalog server with
+	 * TYPE 'object_store'.
+	 */
 	appendStringInfo(fetchObjectStoreMetadata,
 					 "  SELECT "
 					 "  c.metadata_location,"
@@ -331,7 +338,17 @@ PushMetadataLocationToObjectStoreCatalog(void)
 					 "  WHERE EXISTS ("
 					 "		SELECT 1"
 					 "		FROM unnest(f.ftoptions) opt"
-					 "		WHERE LOWER(opt) = LOWER('catalog=" OBJECT_STORE_CATALOG_NAME "')"
+					 "		WHERE opt ~* '^catalog='"
+					 "		  AND ("
+					 "			LOWER(opt) = LOWER('catalog=" OBJECT_STORE_CATALOG_NAME "')"
+					 "			OR LOWER(SUBSTRING(opt FROM 9)) IN ("
+					 "				SELECT LOWER(srvname)"
+					 "				FROM pg_foreign_server s"
+					 "				JOIN pg_foreign_data_wrapper w ON s.srvfdw = w.oid"
+					 "				WHERE w.fdwname = '" ICEBERG_CATALOG_FDW_NAME "'"
+					 "				  AND LOWER(s.srvtype) = LOWER('" OBJECT_STORE_CATALOG_NAME "')"
+					 "			)"
+					 "		  )"
 					 "		)");
 
 	/*
@@ -569,3 +586,47 @@ GetObjectStoreDefaultLocationPrefix(void)
 
 	return StripTrailingSlash(ObjectStoreCatalogLocationPrefix, inPlace);
 }
+
+
+/*
+ * GetObjectStoreCatalogOptionsFromCatalog returns resolved object store
+ * catalog options.  For the built-in 'object_store' literal the GUCs are
+ * used directly.  For user-created servers the server options override
+ * the GUC defaults.
+ */
+ObjectStoreCatalogOptions *
+GetObjectStoreCatalogOptionsFromCatalog(const char *catalog)
+{
+	ObjectStoreCatalogOptions *opts = palloc0(sizeof(ObjectStoreCatalogOptions));
+
+	if (pg_strcasecmp(catalog, OBJECT_STORE_CATALOG_NAME) == 0)
+		opts->catalog = pstrdup(OBJECT_STORE_CATALOG_NAME);
+	else
+		opts->catalog = pstrdup(catalog);
+
+	/* GUC defaults */
+	opts->locationPrefix = (char *) GetObjectStoreDefaultLocationPrefix();
+	opts->readOnly = false;
+
+	if (pg_strcasecmp(catalog, OBJECT_STORE_CATALOG_NAME) != 0)
+	{
+		ForeignServer *server = GetForeignServerByName(catalog, false);
+		ListCell   *lc;
+
+		foreach(lc, server->options)
+		{
+			DefElem    *def = (DefElem *) lfirst(lc);
+
+			if (pg_strcasecmp(def->defname, "location_prefix") == 0)
+			{
+				bool		inPlace = false;
+
+				opts->locationPrefix = StripTrailingSlash(defGetString(def), inPlace);
+			}
+			else if (pg_strcasecmp(def->defname, "read_only") == 0)
+				opts->readOnly = defGetBoolean(def);
+		}
+	}
+
+	return opts;
+}

pg_lake_iceberg/src/rest_catalog/rest_catalog.c

@@ -118,6 +118,7 @@ static const char *iceberg_catalog_server_options[] = {
 	"enable_vended_credentials",
 	"location_prefix",
 	"catalog_name",
+	"read_only",
 	NULL
 };
 
@@ -146,7 +147,7 @@ is_valid_option_in_list(const char *keyword, const char *const *options)
  * iceberg_catalog_validator validates options for the iceberg_catalog FDW.
  *
  * Server options: rest_endpoint, scope, rest_auth_type, oauth_endpoint,
- *   enable_vended_credentials, location_prefix, catalog_name.
+ *   enable_vended_credentials, location_prefix, catalog_name, read_only.
  * User mapping options: client_id, client_secret, scope.
  *
  * scope is accepted in both places; user mapping scope takes priority.
@@ -199,7 +200,7 @@ iceberg_catalog_validator(PG_FUNCTION_ARGS)
 					 errmsg("invalid option \"%s\" for iceberg_catalog server", def->defname),
 					 errhint("Valid options are: rest_endpoint, rest_auth_type, "
 							 "oauth_endpoint, scope, enable_vended_credentials, "
-							 "location_prefix, catalog_name.")));
+							 "location_prefix, catalog_name, read_only.")));
 		}
 
 		if (pg_strcasecmp(def->defname, "rest_auth_type") == 0)
@@ -214,7 +215,8 @@ iceberg_catalog_validator(PG_FUNCTION_ARGS)
 						 errmsg("invalid rest_auth_type option: \"%s\"", authType),
 						 errhint("Valid values are \"oauth2\" and \"horizon\".")));
 		}
-		else if (pg_strcasecmp(def->defname, "enable_vended_credentials") == 0)
+		else if (pg_strcasecmp(def->defname, "enable_vended_credentials") == 0 ||
+				 pg_strcasecmp(def->defname, "read_only") == 0)
 		{
 			(void) defGetBoolean(def);
 		}
@@ -349,8 +351,8 @@ RedactRestCatalogUserMappingSecrets(ProcessUtilityParams * processUtilityParams,
  * ValidateIcebergCatalogServerDDL validates DDL on iceberg_catalog servers:
  *
  *  - CREATE SERVER: rejects reserved names ('postgres', 'object_store',
- *    'rest'), rejects TYPE 'postgres'/'object_store', and requires
- *    TYPE 'rest'.
+ *    'rest'), rejects TYPE 'postgres', and requires TYPE 'rest' or
+ *    'object_store'.
  *  - ALTER SERVER RENAME TO: rejects renaming to a reserved name.
  *
  * ALTER/DROP/OWNER on reserved names will fail naturally because no
@@ -381,21 +383,20 @@ ValidateIcebergCatalogServerDDL(ProcessUtilityParams * processUtilityParams,
 					 errhint("Choose a different server name.")));
 
 		if (stmt->servertype != NULL &&
-			(pg_strcasecmp(stmt->servertype, POSTGRES_CATALOG_NAME) == 0 ||
-			 pg_strcasecmp(stmt->servertype, OBJECT_STORE_CATALOG_NAME) == 0))
+			pg_strcasecmp(stmt->servertype, POSTGRES_CATALOG_NAME) == 0)
 			ereport(ERROR,
 					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 					 errmsg("cannot create iceberg_catalog server with TYPE '%s'",
 							stmt->servertype),
-					 errhint("Use the built-in \"%s\" or \"%s\" catalogs, "
-							 "or create a server of type 'rest'.",
-							 POSTGRES_CATALOG_NAME, OBJECT_STORE_CATALOG_NAME)));
+					 errhint("Use the built-in \"%s\" catalog.",
+							 POSTGRES_CATALOG_NAME)));
 
 		if (stmt->servertype == NULL ||
-			pg_strcasecmp(stmt->servertype, REST_CATALOG_NAME) != 0)
+			(pg_strcasecmp(stmt->servertype, REST_CATALOG_NAME) != 0 &&
+			 pg_strcasecmp(stmt->servertype, OBJECT_STORE_CATALOG_NAME) != 0))
 			ereport(ERROR,
 					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
-					 errmsg("iceberg_catalog server requires TYPE 'rest'")));
+					 errmsg("iceberg_catalog server requires TYPE 'rest' or 'object_store'")));
 	}
 	else if (IsA(parsetree, RenameStmt))
 	{

pg_lake_table/src/ddl/create_table.c

@@ -865,12 +865,23 @@ ProcessCreateIcebergTableFromForeignTableStmt(ProcessUtilityParams * params)
 
 	if (hasObjectStoreCatalogOption)
 	{
-		const char *objectStoreCatalogLocationPrefix = GetObjectStoreDefaultLocationPrefix();
+		char	   *catalogOptionValue =
+			GetStringOption(createStmt->options, "catalog", false);
+		ObjectStoreCatalogOptions *osOpts =
+			GetObjectStoreCatalogOptionsFromCatalog(catalogOptionValue);
+
+		/*
+		 * Named servers may carry their own location_prefix; fall back to
+		 * the GUC-based prefix for the built-in 'object_store' catalog.
+		 */
+		if (osOpts->locationPrefix != NULL)
+			defaultLocationPrefix = (char *) osOpts->locationPrefix;
 
-		if (objectStoreCatalogLocationPrefix == NULL)
+		if (defaultLocationPrefix == NULL)
 		{
 			ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-							errmsg(OBJECT_STORE_CATALOG_NAME " catalog iceberg tables require "
+							errmsg("object_store catalog iceberg tables require "
+								   "a location_prefix server option or "
 								   "pg_lake_iceberg.object_store_catalog_location_prefix "
 								   "to be set")));
 		}