Change the object store table/catalog layout (#97)

This change does 2 things:

- When a table is created with `catalog=obeject_store`, the catalog
file used to be in `defaultPrefix/_catalog/frompg/pg_db.json` after this
commit: `defaultPrefix/frompg/catalog/pg_db.json`

- When an `catalog=object_store` iceberg table is created,
it used to be located in `defaultPrefix/dbname/schema/table_name/oid`.
After this commit: `defaultPrefix/frompg/tables/dbname/schema/table_name/oid`

The goal is to be able to differentiate the tables written by
Postgres, and other systems. The `frompg` allows this. Also,
we can control the access of `/catalog` folder such that only
priviledged users can access that.

Author: sfc-gh-okalaci

pg_lake_iceberg/include/pg_lake/iceberg/catalog.h

@@ -60,7 +60,7 @@ extern PGDLLEXPORT void UpdateExternalCatalogMetadataLocation(char *catalogName,
 															  const char *previousMetadataLocation);
 extern PGDLLEXPORT void UpdateInternalCatalogMetadataLocation(Oid relationId, const char *metadataLocation, const char *previousMetadataLocation);
 extern PGDLLEXPORT void UpdateAllInternalIcebergTablesToReadOnly(void);
-extern PGDLLEXPORT const char *GetIcebergDefaultLocationPrefix(void);
+extern PGDLLEXPORT char *GetIcebergDefaultLocationPrefix(void);
 extern PGDLLEXPORT bool IcebergTablesCatalogExists(void);
 extern PGDLLEXPORT void ErrorIfReadOnlyIcebergTable(Oid relationId);
 extern PGDLLEXPORT bool WarnIfReadOnlyIcebergTable(Oid relationId);

pg_lake_iceberg/src/iceberg/catalog.c

@@ -740,7 +740,7 @@ UpdateAllInternalIcebergTablesToReadOnly(void)
  * GetIcebergDefaultLocationPrefix returns the default location prefix
  * for iceberg tables. Trailing slash is removed, if present.
  */
-const char *
+char *
 GetIcebergDefaultLocationPrefix(void)
 {
 	if (IcebergDefaultLocationPrefix == NULL)

pg_lake_iceberg/src/object_store_catalog/object_store_catalog.c

@@ -504,7 +504,7 @@ GetExternalObjectStoreCatalogFilePath(const char *catalogName)
 				 errdetail("Set the GUC to use catalog=object_store.")));
 	}
 
-	return psprintf("%s/_catalog/%s/%s.json", defaultPrefix,
+	return psprintf("%s/%s/catalog/%s.json", defaultPrefix,
 					ExternalObjectStorePrefix, URLEncodePath(catalogName));
 }
 
@@ -529,7 +529,7 @@ GetInternalObjectStoreCatalogFilePath(const char *catalogName)
 				 errdetail("Set the GUC to use catalog=object_store.")));
 	}
 
-	return psprintf("%s/_catalog/%s/%s.json", defaultPrefix,
+	return psprintf("%s/%s/catalog/%s.json", defaultPrefix,
 					InternalObjectStorePrefix, URLEncodePath(catalogName));
 }
 

pg_lake_table/src/ddl/create_table.c

@@ -822,7 +822,40 @@ ProcessCreateIcebergTableFromForeignTableStmt(ProcessUtilityParams * params)
 	}
 
 	DefElem    *locationOption = GetOption(createStmt->options, "location");
-	const char *defaultLocationPrefix = GetIcebergDefaultLocationPrefix();
+	char	   *defaultLocationPrefix = GetIcebergDefaultLocationPrefix();
+
+	if (hasObjectStoreCatalogOption)
+	{
+		const char *objectStoreCatalogLocationPrefix = GetObjectStoreDefaultLocationPrefix();
+
+		if (objectStoreCatalogLocationPrefix == NULL)
+		{
+			ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+							errmsg("object store catalog iceberg tables require "
+								   "pg_lake_iceberg.object_store_catalog_location_prefix "
+								   "to be set")));
+		}
+
+		if (InternalObjectStorePrefix == NULL)
+		{
+			ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+							errmsg("object store catalog iceberg tables require "
+								   "pg_lake_iceberg.internal_iceberg_storage_prefix "
+								   "to be set")));
+		}
+
+		/* here we only deal with writable tables */
+		Assert(!HasReadOnlyOption(createStmt->options));
+
+		/*
+		 * For hasObjectStoreCatalogOption, we also append
+		 * InternalObjectStorePrefix/tables to the location
+		 */
+		defaultLocationPrefix = psprintf("%s/%s/%s",
+										 defaultLocationPrefix,
+										 InternalObjectStorePrefix,
+										 "tables");
+	}
 
 	/*
 	 * We will set the location by using the default location prefix when user
@@ -841,19 +874,6 @@ ProcessCreateIcebergTableFromForeignTableStmt(ProcessUtilityParams * params)
 	}
 
 
-	if (hasObjectStoreCatalogOption)
-	{
-		const char *objectStoreCatalogLocationPrefix = GetObjectStoreDefaultLocationPrefix();
-
-		if (objectStoreCatalogLocationPrefix == NULL)
-		{
-			ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-							errmsg("object store catalog iceberg tables require "
-								   "pg_lake_iceberg.object_store_catalog_location_prefix "
-								   "to be set")));
-		}
-	}
-
 	/*
 	 * Our CREATE FOREIGN TABLE statement is fully ready for execution, so we
 	 * go to the parent ProcessUtility.

pg_lake_table/tests/pytests/test_object_store_catalog.py

@@ -722,6 +722,62 @@ def test_iceberg_multiple_dbs(
     superuser_conn.autocommit = False
 
 
+def test_create_table_with_default_location_object_store(
+    pg_conn,
+    superuser_conn,
+    s3,
+    extension,
+    with_default_location,
+    adjust_object_store_settings,
+):
+    dbname = run_query("SELECT current_database()", pg_conn)[0][0]
+    run_command(
+        "CREATE SCHEMA test_create_table_with_default_location_object_store", pg_conn
+    )
+    run_command(
+        f"""CREATE TABLE test_create_table_with_default_location_object_store.tbl (a int, b int)
+                    USING iceberg WITH (catalog='object_store')""",
+        pg_conn,
+    )
+    pg_conn.commit()
+    wait_until_object_store_writable_table_pushed(
+        pg_conn, "test_create_table_with_default_location_object_store", "tbl"
+    )
+
+    # assert metadata location
+    result = run_query(
+        """SELECT metadata_location FROM iceberg_tables
+                           WHERE table_namespace = 'test_create_table_with_default_location_object_store' and table_name = 'tbl'
+                       """,
+        pg_conn,
+    )
+    first_table_metadata_location = result[0][0]
+
+    table_oid = run_query(
+        """SELECT oid FROM pg_class
+                            WHERE oid = 'test_create_table_with_default_location_object_store.tbl'::regclass
+                          """,
+        pg_conn,
+    )[0][0]
+
+    prefix = run_query(
+        "SHOW pg_lake_iceberg.internal_object_store_catalog_prefix", superuser_conn
+    )[0][0]
+    superuser_conn.commit()
+
+    assert (
+        f"s3://{TEST_BUCKET}/{prefix}/tables/{dbname}/test_create_table_with_default_location_object_store/tbl/{table_oid}"
+        in first_table_metadata_location
+    )
+
+    # drop the table and create it again
+    run_command(
+        "DROP SCHEMA test_create_table_with_default_location_object_store CASCADE",
+        pg_conn,
+    )
+    pg_conn.commit()
+
+
 def test_complex_types_object_store(
     pg_conn, s3, extension, with_default_location, adjust_object_store_settings
 ):