feat: add Cloudberry gang-parallel support to PXF FDW

- Introduced virtual segment ID handling for parallel execution in Cloudberry.
- Added PxfBridgeImportStartVirtual function to manage imports with virtual segment IDs.
- Updated PxfFdwScanState structure to include fields for gang-parallel execution.
- Enhanced foreign scan functions to support gang-parallel mode, ensuring unique fragment distribution among workers.
- Implemented initialization and cleanup routines for gang-parallel state management.

Author: MisterRaindrop

fdw/pxf_bridge.c

@@ -23,6 +23,8 @@
 
 #include "cdb/cdbtm.h"
 #include "cdb/cdbvars.h"
+#include "miscadmin.h"
+#include "storage/lock.h"
 #include "utils/builtins.h"
 
 /* helper function declarations */
@@ -392,3 +394,66 @@ PxfBridgeImportStartFragment(PxfFdwScanState *pxfsstate, int fragmentIndex)
 	/* Update current fragment tracking */
 	pxfsstate->current_fragment = fragmentIndex;
 }
+
+/*
+ * ============================================================================
+ * Cloudberry Gang-Parallel Support (Virtual Segment ID)
+ *
+ * In Cloudberry/Greenplum, parallel execution uses "gang expansion" where
+ * multiple processes share the same physical segment ID. PostgreSQL's DSM
+ * callbacks (InitializeDSMForeignScan, InitializeWorkerForeignScan) are
+ * NOT invoked in this model.
+ *
+ * Instead of fragment-by-fragment coordination, we use "virtual segment IDs":
+ * each gang worker sends a unique virtual segment ID to PXF, so PXF's
+ * existing round-robin fragment distribution splits the data among workers
+ * automatically — no PXF server changes needed.
+ *
+ * Example: 3 physical segments × 4 workers = 12 virtual segments.
+ * Worker i on physical segment S sends virtual_seg_id = S + i * seg_count,
+ * with virtual_seg_count = seg_count * workers.
+ * ============================================================================
+ */
+
+/*
+ * PxfBridgeImportStartVirtual
+ *		Start import with virtual segment ID for Cloudberry gang-parallel mode.
+ *
+ * Same as PxfBridgeImportStart, but after building the standard HTTP headers,
+ * overrides X-GP-SEGMENT-ID and X-GP-SEGMENT-COUNT with the virtual values.
+ * This makes PXF's round-robin assign a unique subset of fragments to each
+ * gang worker, eliminating data duplication.
+ */
+void
+PxfBridgeImportStartVirtual(PxfFdwScanState *pxfsstate,
+							int virtualSegId, int virtualSegCount)
+{
+	char		seg_id_str[16];
+	char		seg_count_str[16];
+
+	pxfsstate->churl_headers = churl_headers_init();
+
+	BuildUriForRead(pxfsstate);
+	BuildHttpHeaders(pxfsstate->churl_headers,
+					 pxfsstate->options,
+					 pxfsstate->relation,
+					 pxfsstate->filter_str,
+					 pxfsstate->retrieved_attrs,
+					 pxfsstate->projectionInfo);
+
+	/* Override physical segment ID/count with virtual values */
+	pg_ltoa(virtualSegId, seg_id_str);
+	pg_ltoa(virtualSegCount, seg_count_str);
+	churl_headers_override(pxfsstate->churl_headers, "X-GP-SEGMENT-ID", seg_id_str);
+	churl_headers_override(pxfsstate->churl_headers, "X-GP-SEGMENT-COUNT", seg_count_str);
+
+	elog(DEBUG3, "pxf_fdw: PxfBridgeImportStartVirtual physical_seg=%d "
+		 "virtual_seg_id=%d virtual_seg_count=%d",
+		 PXF_SEGMENT_ID, virtualSegId, virtualSegCount);
+
+	pxfsstate->churl_handle = churl_init_download(pxfsstate->uri.data,
+												   pxfsstate->churl_headers);
+
+	/* read some bytes to make sure the connection is established */
+	churl_read_check_connectivity(pxfsstate->churl_handle);
+}

fdw/pxf_bridge.h

@@ -75,12 +75,25 @@ typedef struct PxfFdwScanState
 	CopyFromState	cstate;
 	ProjectionInfo *projectionInfo;
 
-	/* Parallel execution state */
-	bool		is_parallel;		/* true if running in parallel mode */
+	/* Parallel execution state (PG DSM-based) */
+	bool		is_parallel;		/* true if running in DSM parallel mode */
 	PxfParallelScanState *pstate;	/* pointer to shared state in DSM */
 	PxfFragmentData *fragments;		/* array of fragment metadata */
 	int			num_fragments;		/* total number of fragments */
 	int			current_fragment;	/* current fragment being processed */
+
+	/*
+	 * Cloudberry gang-parallel state (virtual segment ID based).
+	 * Used when Cloudberry's gang expansion creates multiple processes per
+	 * segment but PG's DSM callbacks are not invoked. Each gang worker gets
+	 * a unique virtual segment ID so PXF's round-robin distributes fragments
+	 * evenly without data duplication.
+	 */
+	bool		gang_parallel;		/* true when using virtual segment IDs */
+	bool		plan_parallel_aware;	/* plan node's parallel_aware flag */
+	int			worker_index;		/* this worker's index within the segment gang */
+	int			virtual_seg_id;		/* virtual segment ID sent to PXF */
+	int			virtual_seg_count;	/* virtual segment count sent to PXF */
 } PxfFdwScanState;
 
 /*
@@ -123,4 +136,10 @@ int			PxfBridgeGetNextFragment(PxfParallelScanState *pstate);
 /* Start import for a specific fragment in parallel mode */
 void		PxfBridgeImportStartFragment(PxfFdwScanState *pxfsstate, int fragmentIndex);
 
+/* Cloudberry gang-parallel support (virtual segment ID based) */
+
+/* Start import with virtual segment ID for gang-parallel mode */
+void		PxfBridgeImportStartVirtual(PxfFdwScanState *pxfsstate,
+										int virtualSegId, int virtualSegCount);
+
 #endif							/* _PXFBRIDGE_H */