WIP, drop table

Author: sfc-gh-okalaci

pg_lake_iceberg/include/pg_lake/rest_catalog/rest_catalog.h

@@ -45,7 +45,8 @@ typedef enum RestCatalogOperationType
 	REST_CATALOG_ADD_SNAPSHOT = 1,
 	REST_CATALOG_ADD_SCHEMA = 2,
 	REST_CATALOG_ADD_PARTITION = 3,
-	REST_CATALOG_REMOVE_SNAPSHOT = 4
+	REST_CATALOG_REMOVE_SNAPSHOT = 4,
+	REST_CATALOG_DROP_TABLE = 5
 } RestCatalogOperationType;
 
 
@@ -74,6 +75,7 @@ extern PGDLLEXPORT char *GetMetadataLocationFromRestCatalog(const char *restCata
 extern PGDLLEXPORT char *GetMetadataLocationForRestCatalogForIcebergTable(Oid relationId);
 extern PGDLLEXPORT void ReportHTTPError(HttpResult httpResult, int level);
 extern PGDLLEXPORT List *PostHeadersWithAuth(void);
+extern PGDLLEXPORT List *DeleteHeadersWithAuth(void);
 extern PGDLLEXPORT RestCatalogRequest * GetAddSnapshotCatalogRequest(IcebergSnapshot * newSnapshot, Oid relationId);
 extern PGDLLEXPORT RestCatalogRequest * GetAddSchemaCatalogRequest(Oid relationId, DataFileSchema * dataFileSchema);
 extern PGDLLEXPORT RestCatalogRequest * GetAddPartitionCatalogRequest(Oid relationId, List *partitionSpec);

pg_lake_iceberg/src/rest_catalog/rest_catalog.c

@@ -440,6 +440,19 @@ PostHeadersWithAuth(void)
 					  pstrdup("Content-Type: application/json"));
 }
 
+
+
+/*
+* Creates the headers for a DELETE request with authentication.
+*/
+List *
+DeleteHeadersWithAuth(void)
+{
+	return list_make1(psprintf("Authorization: Bearer %s", RestCatalogFetchAccessToken()));
+}
+
+
+
 /*
 * Creates the headers for a GET request with authentication.
 */

pg_lake_table/src/transaction/track_iceberg_metadata_changes.c

@@ -53,7 +53,9 @@ typedef struct RestCatalogRequestPerTable
 	char	   *catalogNamespace;
 	char	   *catalogTableName;
 
-	List	   *createTableRequests;
+	RestCatalogRequest *createTableRequest;
+	RestCatalogRequest *dropTableRequest;
+
 	List	   *tableModifyRequests;
 }			RestCatalogRequestPerTable;
 
@@ -205,13 +207,17 @@ PostAllRestCatalogRequests(void)
 
 	while ((requestPerTable = hash_seq_search(&status)) != NULL)
 	{
-		ListCell   *requestCell = NULL;
+		RestCatalogRequest *createTableRequest = requestPerTable->createTableRequest;
+		RestCatalogRequest *dropTableRequest = requestPerTable->dropTableRequest;
 
-		foreach(requestCell, requestPerTable->createTableRequests)
+		if (createTableRequest != NULL || dropTableRequest != NULL)
 		{
-			RestCatalogRequest *request = (RestCatalogRequest *) lfirst(requestCell);
-
-			Assert(request->operationType == REST_CATALOG_CREATE_TABLE);
+			/*
+			 * We can only have one create or one drop table request per
+			 * table. If a table is created and dropped in the same
+			 * transaction, we skip both requests.
+			 */
+			Assert(createTableRequest == NULL || dropTableRequest == NULL);
 
 			const char *url =
 				psprintf(REST_CATALOG_TABLE,
@@ -220,16 +226,37 @@ PostAllRestCatalogRequests(void)
 						 URLEncodePath(requestPerTable->catalogNamespace),
 						 URLEncodePath(requestPerTable->catalogTableName));
 
-			HttpResult	httpResult = HttpPost(url, request->body, PostHeadersWithAuth());
+			if (createTableRequest != NULL)
+			{
+				HttpResult	httpResult = HttpPost(url, createTableRequest->body, PostHeadersWithAuth());
+
+				if (httpResult.status != 200)
+				{
+					ReportHTTPError(httpResult, WARNING);
 
-			if (httpResult.status != 200)
+					/*
+					 * Ouch, something failed. Should we stop sending the
+					 * requests?
+					 */
+				}
+			}
+			else if (dropTableRequest != NULL)
 			{
-				ReportHTTPError(httpResult, WARNING);
+				HttpResult	httpResult = HttpDelete(url, DeleteHeadersWithAuth());
+
+				if (httpResult.status != 204)
+				{
+					ReportHTTPError(httpResult, WARNING);
 
-				/*
-				 * Ouch, something failed. Should we stop sending the
-				 * requests?
-				 */
+					/*
+					 * Ouch, something failed. Should we stop sending the
+					 * requests?
+					 */
+				}
+			}
+			else
+			{
+				pg_unreachable();
 			}
 		}
 	}
@@ -481,7 +508,11 @@ RecordRestCatalogRequestInTx(Oid relationId, RestCatalogOperationType operationT
 	if (operationType == REST_CATALOG_CREATE_TABLE)
 	{
 		request->body = pstrdup(body);
-		requestPerTable->createTableRequests = list_make1(request);
+		requestPerTable->createTableRequest = request;
+	}
+	else if (operationType == REST_CATALOG_DROP_TABLE)
+	{
+		requestPerTable->dropTableRequest = request;
 	}
 	else if (operationType == REST_CATALOG_ADD_SNAPSHOT ||
 			 operationType == REST_CATALOG_ADD_SCHEMA ||
@@ -536,7 +567,16 @@ ApplyTrackedIcebergMetadataChanges(void)
 
 		/* relation is dropped */
 		if (!RelationExistsInTheIcebergCatalog(relationId))
+		{
+			/*
+			 * if created and dropped in the same tx, treat as no-op and skip
+			 * all
+			 */
+			if (!opTracker->relationCreated)
+				RecordRestCatalogRequestInTx(relationId, REST_CATALOG_DROP_TABLE, NULL);
+
 			continue;
+		}
 
 		List	   *allTransforms = AllPartitionTransformList(relationId);