Actually create the namespace on table creation

Signed-off-by: Onder KALACI <[email protected]>

Author: sfc-gh-okalaci

pg_lake_iceberg/include/pg_lake/rest_catalog/rest_catalog.h

@@ -38,6 +38,7 @@ extern PGDLLEXPORT void RegisterNamespaceToRestCatalog(const char *catalogName,
 													   bool hasRestCatalogReadOnlyOption);
 extern PGDLLEXPORT void ErrorIfRestNamespaceDoesNotExist(const char *catalogName, const char *namespaceName);
 extern PGDLLEXPORT IcebergCatalogType GetIcebergCatalogType(Oid relationId);
+extern PGDLLEXPORT char *GetRestCatalogName(Oid relationId);
 extern PGDLLEXPORT char *GetRestCatalogNamespace(Oid relationId);
 extern PGDLLEXPORT char *GetRestCatalogTableName(Oid relationId);
 extern PGDLLEXPORT bool HasRestCatalogTableOption(List *options);

pg_lake_iceberg/src/rest_catalog/rest_catalog.c

@@ -44,7 +44,6 @@ char	   *RestCatalogClientSecret = NULL;
 static void CreateNamespaceOnRestCatalog(const char *catalogName, const char *namespaceName);
 static char *EncodeBasicAuth(const char *clientId, const char *clientSecret);
 static char *JsonbGetStringByPath(const char *jsonb_text, int nkeys,...);
-static char *GetRestCatalogName(Oid relationId);
 static List *PostHeadersWithAuth(void);
 static List *GetHeadersWithAuth(void);
 static void ReportHTTPError(HttpResult httpResult, int level);
@@ -582,7 +581,7 @@ HasReadOnlyOption(List *options)
 * as the catalog name in the external catalog. Writable rest catalog tables
 * use the current database name as the catalog name.
 */
-static char *
+char *
 GetRestCatalogName(Oid relationId)
 {
 	IcebergCatalogType catalogType = GetIcebergCatalogType(relationId);

pg_lake_table/src/ddl/create_table.c

@@ -659,6 +659,8 @@ ProcessCreateIcebergTableFromForeignTableStmt(ProcessUtilityParams * params)
 
 	if (hasObjectStoreCatalogOption || hasRestCatalogOption)
 	{
+		Oid			namespaceId = RangeVarGetAndCheckCreationNamespace(createStmt->base.relation, NoLock, NULL);
+
 		/*
 		 * Read-only external catalog tables are a special case of Iceberg
 		 * tables. They are recognized as Iceberg tables, but are not
@@ -684,8 +686,6 @@ ProcessCreateIcebergTableFromForeignTableStmt(ProcessUtilityParams * params)
 		 */
 		if (catalogNamespaceProvided == NULL && hasExternalCatalogReadOnlyOption)
 		{
-			Oid			namespaceId = RangeVarGetAndCheckCreationNamespace(createStmt->base.relation, NoLock, NULL);
-
 			catalogNamespace = get_namespace_name(namespaceId);
 
 			/* add catalog_namespace table options */
@@ -761,6 +761,30 @@ ProcessCreateIcebergTableFromForeignTableStmt(ProcessUtilityParams * params)
 								errmsg("writable %s catalog iceberg tables do not "
 									   "allow explicit catalog options", hasObjectStoreCatalogOption ? "object store" : "REST")));
 			}
+
+			if (hasRestCatalogOption)
+			{
+				/*
+				 * For writable rest catalog iceberg tables, we register the
+				 * namespace in the rest catalog. We do that early in the
+				 * command processing so that any errors in the registration
+				 * are caught before we create the actual table.
+				 *
+				 * Note that registering a namespace is not a transactional
+				 * operation from pg_lake's perspective. If the subsequent
+				 * table creation fails, the namespace registration will
+				 * remain. We accept that tradeoff for simplicity as
+				 * re-registering an existing namespace is a no-op. For a
+				 * writable rest catalog iceberg table, the namespace is
+				 * always the table's schema name. Similarly, the catalog name
+				 * is always the database name. We normally encode that in
+				 * GetRestCatalogName() etc., but here we need to do it early
+				 * before the table is created.
+				 */
+				RegisterNamespaceToRestCatalog(get_database_name(MyDatabaseId),
+											   get_namespace_name(namespaceId),
+											   hasExternalCatalogReadOnlyOption);
+			}
 		}
 		else if (createStmt->base.tableElts == NIL && hasExternalCatalogReadOnlyOption)
 		{

pg_lake_table/tests/pytests/test_polaris_catalog.py

@@ -62,7 +62,7 @@ def test_create_namespace(
     pg_conn.commit()
 
     run_command(
-        f"""SELECT lake_iceberg.register_namespace_to_rest_catalog('{server_params.PG_DATABASE}', '{namespace}')""",
+        f"""CREATE TABLE "{namespace}".tbl(a int) USING iceberg WITH (catalog='rest');""",
         pg_conn,
     )
     pg_conn.commit()
@@ -132,13 +132,13 @@ def test_create_namespace_in_tx(
     run_command(f'''CREATE SCHEMA "{namespace}_2"''', pg_conn)
 
     run_command(
-        f"""SELECT lake_iceberg.register_namespace_to_rest_catalog('{server_params.PG_DATABASE}', '{namespace}')""",
+        f"""CREATE TABLE "{namespace}".tbl(a int) USING iceberg WITH (catalog='rest');""",
         pg_conn,
     )
     pg_conn.commit()
 
     run_command(
-        f"""SELECT lake_iceberg.register_namespace_to_rest_catalog('{server_params.PG_DATABASE}', '{namespace}_2')""",
+        f"""CREATE TABLE "{namespace}_2".tbl(a int) USING iceberg WITH (catalog='rest');""",
         pg_conn,
     )
     pg_conn.commit()
@@ -176,7 +176,7 @@ def test_create_namespace_rollback(
     run_command(f'''CREATE SCHEMA "{namespace}"''', pg_conn)
 
     run_command(
-        f"""SELECT lake_iceberg.register_namespace_to_rest_catalog('{server_params.PG_DATABASE}', '{namespace}')""",
+        f"""CREATE TABLE "{namespace}".tbl(a int) USING iceberg WITH (catalog='rest');""",
         pg_conn,
     )