PXF fdw supports parallel data loading.

[MisterRaindrop]

Parallel support for FDW is still an early-stage commit. It requires modifications to the Cloudberry kernel to support parallelism, as well as refactoring the FDW code in PXF.

Currently, I have a demo that can run through simple cases.

However, I have a question:

since we now rely on the stable branch of Cloudberry, can I still submit code to the stable branch REL_2_STABLE?

[MisterRaindrop]

How are the versions of PXF and the cloudberry kernel planned?

[ostinru]

How are the versions of PXF and the cloudberry kernel planned?

For Greenplum+PXF these component versions they were loosely connected. Probably, only major versions should match (e.g. Greenplum 6.x and PXF 6.x). In production we have clusters with latests Greenplum 6 and PXF 6.2.3 (there were a bug fixed in lates 6.10.x).

So, I think we can try to keep single version of PXF for all Cloudberry versions (unless we need changes in Cloudberry core).

[my-ship-it]

Parallel support for FDW is still an early-stage commit. It requires modifications to the Cloudberry kernel to support parallelism, as well as refactoring the FDW code in PXF.

Currently, I have a demo that can run through simple cases.

However, I have a question:

since we now rely on the stable branch of Cloudberry, can I still submit code to the stable branch REL_2_STABLE?

I think there is no problem. Generally, plugins are based on the stable branch of the kernel, and the stable branch ensures the compatibility of ABI interfaces.

[MisterRaindrop]

Parallel support for FDW is still an early-stage commit. It requires modifications to the Cloudberry kernel to support parallelism, as well as refactoring the FDW code in PXF.

Currently, I have a demo that can run through simple cases.

However, I have a question:

since we now rely on the stable branch of Cloudberry, can I still submit code to the stable branch REL_2_STABLE?

I think there is no problem. Generally, plugins are based on the stable branch of the kernel, and the stable branch ensures the compatibility of ABI interfaces.

The issue is that parallel testing requires reliance on the modified Cloudberry kernel code I worked on, otherwise, effective testing cannot be conducted.

[ostinru]

Parallel support for FDW is still an early-stage commit.

Which part of pipeline Cloudberry => pxf_fdw => pxf => S3/HDFS are you going to query concurrently? Do you want to send more than one HTTP requests per one postgreSQL process?

[MisterRaindrop]

Parallel support for FDW is still an early-stage commit.

Which part of pipeline Cloudberry => pxf_fdw => pxf => S3/HDFS are you going to query concurrently? Do you want to send more than one HTTP requests per one postgreSQL process?

The parallel capability of FDW in Cloudberry (PostgreSQL parallelization) is utilized. When parallelism is enabled, multiple processes are launched to perform queries in parallel. This involves kernel support for parallelism in Cloudberry, modifications to the FDW and server in Cloudberry-PXF, and I have already successfully run simple test cases.

[MisterRaindrop]

Cloudberry has N physical segments (e.g., 3). Each segment opens one PXF connection, so only 3 processes read data in parallel. For large
datasets, this is a bottleneck.

Solution: Virtual Segment IDs

Multiply parallelism by having multiple workers per segment, each pretending to be a different segment from PXF's perspective.

Formula:
virtual_seg_id = physical_seg_id + worker_index * physical_seg_count
virtual_seg_count = physical_seg_count * num_workers

Example: 3 segments × 4 workers = 12 virtual segments
Segment 0: workers get virtual IDs 0, 3, 6, 9
Segment 1: workers get virtual IDs 1, 4, 7, 10
Segment 2: workers get virtual IDs 2, 5, 8, 11

PXF's existing round-robin (fragment % seg_count == seg_id) distributes fragments across 12 virtual segments. No PXF server changes needed.

How It Works

Planning (pxfGetForeignPaths):

• Adds a non-parallel path (fallback) and a parallel partial path via add_partial_path()
• Only when enable_parallel = true and max_parallel_workers_per_gather > 0
• PostgreSQL planner generates a Gather node → Cloudberry executor launches gang workers

Execution (pxfIterateForeignScan):

• On first call, checks TotalParallelWorkerNumberOfSlice > 0 (Cloudberry kernel variable)
• If true → InitGangParallelState() computes virtual segment IDs
• InitCopyStateVirtual() → PxfBridgeImportStartVirtual() builds standard HTTP headers, then overrides two headers:
  churl_headers_override(headers, "X-GP-SEGMENT-ID", virtual_seg_id);
  churl_headers_override(headers, "X-GP-SEGMENT-COUNT", virtual_seg_count);
• If not parallel → standard InitCopyState() → PxfBridgeImportStart() with physical segment IDs

Cleanup (pxfEndForeignScan):

• Unified path: churl_handle → churl_headers → cstate → uri → pxfsstate

Example

SET optimizer = off; // off orca
SET enable_parallel = true;
SET max_parallel_workers_per_gather = 4; // set parallel
CREATE FOREIGN TABLE my_table (id int, val text)
    SERVER pxf_server
    OPTIONS (resource 'my_hdfs_path', profile 'hdfs:text',
             enable_parallel 'true');

We need to set enable_parallel to true so that the pxf table can be parallel.

[MisterRaindrop]

@my-ship-it hi max, I have implemented parallelism. In testing, the submission has effectively improved efficiency by 1 to 2 times. Could you review it to see if there are any issues?