rebase on top 70

Author: sfc-gh-okalaci

pg_lake_iceberg/include/pg_lake/iceberg/catalog.h

@@ -55,6 +55,7 @@ extern PGDLLEXPORT void DeleteInternalIcebergCatalogTable(Oid relationId);
 
 extern PGDLLEXPORT List *GetAllInternalIcebergRelationIds(void);
 extern PGDLLEXPORT char *GetIcebergMetadataLocation(Oid relationId, bool forUpdate);
+extern PGDLLEXPORT char *GetIcebergCatalogMetadataLocation(Oid relationId, bool forUpdate);
 extern PGDLLEXPORT char *GetIcebergCatalogPreviousMetadataLocation(Oid relationId, bool forUpdate);
 extern PGDLLEXPORT void UpdateExternalCatalogMetadataLocation(char *catalogName, char *schemaName, char *tableName, const char *metadataLocation,
 															  const char *previousMetadataLocation);

pg_lake_iceberg/src/iceberg/catalog.c

@@ -33,7 +33,6 @@
 
 char	   *IcebergDefaultLocationPrefix = NULL;
 
-static char *GetIcebergCatalogMetadataLocation(Oid relationId, bool forUpdate);
 static char *GetIcebergExternalMetadataLocation(Oid relationId);
 static char *GetIcebergCatalogMetadataLocationInternal(Oid relationId, bool isPrevMetadata, bool forUpdate);
 static char *GetIcebergCatalogColumnInternal(Oid relationId, char *columnName, bool forUpdate, bool errorIfNotFound);
@@ -360,7 +359,37 @@ GetIcebergMetadataLocation(Oid relationId, bool forUpdate)
 
 	if (IsInternalIcebergTable(relationId))
 	{
-		return GetIcebergCatalogMetadataLocation(relationId, forUpdate);
+		IcebergCatalogType catalogType = GetIcebergCatalogType(relationId);
+
+		/*
+		 * We always get the metadata location from the iceberg catalog table
+		 * for internal iceberg tables.
+		 *
+		 * Writable rest catalog iceberg tables are a bit different. They are
+		 * internal iceberg tables, all their metadata is stored in our
+		 * catalogs, such as data files, column stats, etc. However, their
+		 * metadata location is stored in the rest catalog itself. So, we need
+		 * to fetch the metadata location from the rest catalog for those
+		 * tables, see below.
+		 *
+		 * But we still need to acquire locks on the iceberg catalog table row
+		 * for internal iceberg tables to prevent concurrent updates. So, we
+		 * call GetIcebergCatalogMetadataLocation().
+		 */
+		char	   *metadataLocation = GetIcebergCatalogMetadataLocation(relationId, forUpdate);
+
+		if (catalogType == REST_CATALOG_READ_WRITE)
+		{
+			/*
+			 * our internal catalog never stores metadata location for
+			 * writable rest catalog tables
+			 */
+			Assert(metadataLocation == NULL);
+
+			metadataLocation = GetIcebergExternalMetadataLocation(relationId);
+		}
+
+		return metadataLocation;
 	}
 	else
 	{
@@ -381,7 +410,7 @@ GetIcebergMetadataLocation(Oid relationId, bool forUpdate)
 * If the metadata row for the table is going to be updated, the caller should
 * pass forUpdate as true.
 */
-static char *
+char *
 GetIcebergCatalogMetadataLocation(Oid relationId, bool forUpdate)
 {
 	Assert(IsInternalIcebergTable(relationId));
@@ -402,7 +431,7 @@ GetIcebergExternalMetadataLocation(Oid relationId)
 
 	char	   *currentMetadataPath = NULL;
 
-	if (icebergCatalogType == REST_CATALOG_READ_ONLY)
+	if (icebergCatalogType == REST_CATALOG_READ_ONLY || icebergCatalogType == REST_CATALOG_READ_WRITE)
 	{
 		currentMetadataPath = GetMetadataLocationForRestCatalogForIcebergTable(relationId);
 	}

pg_lake_iceberg/src/iceberg/metadata_operations.c

@@ -171,12 +171,16 @@ ApplyIcebergMetadataChanges(Oid relationId, List *metadataOperations, List *allT
 	 * Although writable rest catalog iceberg tables have their metadata
 	 * location stored in the rest catalog itself, we still need to read the
 	 * pg_lake metadata as forUpdate=true acquires necessary locks to prevent
-	 * concurrent updates.
+	 * concurrent updates. To achieve this, we use
+	 * GetIcebergCatalogMetadataLocation function as that's the common
+	 * practice in the code.
 	 */
-	char	   *metadataPath = GetIcebergMetadataLocation(relationId, forUpdate);
+	char	   *metadataPath = GetIcebergCatalogMetadataLocation(relationId, forUpdate);
+
 	bool		createNewTable = HasCreateTableOperation(metadataOperations);
-	
+
 	IcebergTableMetadata *metadata = NULL;
+
 	if (createNewTable)
 	{
 		metadata = GenerateInitialIcebergTableMetadata(relationId);
@@ -203,6 +207,7 @@ ApplyIcebergMetadataChanges(Oid relationId, List *metadataOperations, List *allT
 	}
 	else
 	{
+		metadataPath = GetIcebergMetadataLocation(relationId, false);
 		metadata = ReadIcebergTableMetadata(metadataPath);
 
 		prevLastUpdatedMs = metadata->last_updated_ms;

pg_lake_table/src/ddl/drop_table.c

@@ -463,7 +463,7 @@ MarkAllReferencedFilesForDeletion(Oid relationId)
 	TimestampTz orphanedAt = GetCurrentTransactionStartTimestamp();
 	char	   *metadataLocation = GetIcebergMetadataLocation(relationId, true);
 	MemoryContext savedContext = CurrentMemoryContext;
-	
+
 	List	   *allFiles = NIL;
 	volatile bool success = true;
 

pg_lake_table/src/transaction/track_iceberg_metadata_changes.c

@@ -101,12 +101,15 @@ TrackIcebergMetadataChangesInTx(Oid relationId, List *metadataOperationTypes)
 		return;
 
 	/*
-	 * we might also defer acquiring locks to precommit hook but let's keep
-	 * them here to prevent any subtle bug
+	 * We might also defer acquiring locks to precommit hook but let's keep
+	 * them here to prevent any subtle bug. We call
+	 * GetIcebergCatalogMetadataLocation() to acquire the necessary locks, not
+	 * for the actual metadata location as our serialization of iceberg
+	 * metadata changes relies on those locks.
 	 */
 	bool		forUpdate = true;
 
-	GetIcebergMetadataLocation(relationId, forUpdate);
+	GetIcebergCatalogMetadataLocation(relationId, forUpdate);
 
 	ListCell   *operationCell = NULL;