Skeleton for parallel commit

Author: sfc-gh-mslot

pg_lake_engine/include/pg_lake/util/s3_writer_utils.h

@@ -20,3 +20,4 @@
 extern PGDLLEXPORT void CopyLocalFileToS3(char *localFilePath, char *s3Uri);
 extern PGDLLEXPORT void CopyLocalFileToS3WithCleanupOnAbort(char *localFilePath, char *s3Uri);
 extern PGDLLEXPORT void CopyLocalManifestFileToS3WithCleanupOnAbort(char *localFilePath, char *s3Uri);
+extern PGDLLEXPORT void FinishAllUploads(void);

pg_lake_engine/src/utils/s3_writer_utils.c

@@ -26,7 +26,22 @@
 #include "pg_lake/util/rel_utils.h"
 #include "utils/builtins.h"
 
+
+/*
+ * ScheduledUpload represents an upload to be performed by FinishAllUpload.
+ */
+typedef struct ScheduledUpload
+{
+	char *localFile;
+	char *remoteUrl;
+}	ScheduledUpload;
+
 static char *CopyLocalFileToS3Command(char *localFileUri, char *s3Uri);
+static void ScheduleFileUpload(char *localFile, char *remoteUrl);
+static void ResetPendingUploads(void);
+
+static MemoryContext UploadContext = NULL;
+static List *PendingUploads = NIL;
 
 
 /*
@@ -39,7 +54,7 @@ CopyLocalFileToS3WithCleanupOnAbort(char *localFilePath, char *s3Uri)
 {
 	InsertInProgressFileRecord(s3Uri);
 
-	CopyLocalFileToS3(localFilePath, s3Uri);
+	ScheduleFileUpload(localFilePath, s3Uri);
 }
 
 /*
@@ -55,7 +70,7 @@ CopyLocalManifestFileToS3WithCleanupOnAbort(char *localFilePath, char *s3Uri)
 
 	InsertInProgressFileRecordExtended(s3Uri, isPrefix, deferDeletion);
 
-	CopyLocalFileToS3(localFilePath, s3Uri);
+	ScheduleFileUpload(localFilePath, s3Uri);
 }
 
 /*
@@ -85,3 +100,59 @@ CopyLocalFileToS3Command(char *localFileUri, char *s3Uri)
 
 	return command.data;
 }
+
+
+/*
+ * ScheduleFileUpload schedules a file for being uploaded into object storage
+ * when FinishAllUploads is called.
+ */
+static void
+ScheduleFileUpload(char *localFile, char *remoteUrl)
+{
+	if (UploadContext == NULL)
+	{
+		UploadContext = AllocSetContextCreate(TopTransactionContext,
+											  "Upload scheduler",
+											  ALLOCSET_DEFAULT_SIZES);
+
+		/* reset PendingUploads on abort */
+		MemoryContextCallback *cb = MemoryContextAllocZero(UploadContext,
+														   sizeof(MemoryContextCallback));
+
+		cb->func = (MemoryContextCallbackFunction) ResetPendingUploads;
+		cb->arg = NULL;
+		MemoryContextRegisterResetCallback(UploadContext, cb);
+	}
+
+	MemoryContext oldContext = MemoryContextSwitchTo(UploadContext);
+
+	ScheduledUpload *upload = palloc0(sizeof(ScheduledUpload));
+
+	upload->localFile = pstrdup(localFile);
+	upload->remoteUrl = pstrdup(remoteUrl);
+
+	PendingUploads = lappend(PendingUploads, upload);
+
+	MemoryContextSwitchTo(oldContext);
+}
+
+
+/*
+ * ResetPendingUploads prevents PendingUploads containing dangling pointers
+ * when the memory context is reset.
+ */
+static void
+ResetPendingUploads(void)
+{
+	PendingUploads = NIL;
+}
+
+
+/*
+ * FinishAllUpload completes all of the pending uploads in parallel.
+ */
+void
+FinishAllUploads(void)
+{
+	
+}

pg_lake_table/src/transaction/track_iceberg_metadata_changes.c

@@ -37,6 +37,7 @@
 #include "pg_lake/transaction/transaction_hooks.h"
 #include "pg_lake/util/injection_points.h"
 #include "pg_lake/json/json_utils.h"
+#include "pg_lake/util/s3_writer_utils.h"
 #include "pg_lake/util/url_encode.h"
 
 #define ONE_MB (1 * 1024 * 1024)
@@ -772,6 +773,9 @@ ApplyTrackedIcebergMetadataChanges(void)
 		}
 	}
 
+	/* now write all the metadata files to object storage in parallel */
+	FinishAllUploads();
+
 	INJECTION_POINT_COMPAT("after-apply-iceberg-changes");
 
 	ExternalHeavyAssertsOnIcebergMetadataChange();