[MisterRaindrop]

#58

---

Change logs

pxf_fdw can now scan a foreign table with several gang workers per segment on Cloudberry. Parallel scan is off by default and is enabled with the enable_parallel option on the foreign server or the foreign table; a value on the table overrides the server.

How it works:

• The planner adds a partial ForeignPath when the table has enable_parallel, the session has enable_parallel = on and max_parallel_workers_per_gather > 0, and the Postgres planner is used. The plan is a Parallel Foreign Scan under a gang-parallel Gather Motion.
• At execution, worker w on segment s reports virtual segment s + w * N out of N * W (N segments, W workers) in X-GP-SEGMENT-ID / X-GP-SEGMENT-COUNT. The server's existing round-robin fragment split then gives each worker a disjoint part of what the segment would read. No PXF server change is needed.
• Workers are numbered with ParallelWorkerNumberOfSlice / TotalParallelWorkerNumberOfSlice, which are already on Cloudberry main.

No parallel path is offered for:

• tables with a reject_limit, because the limit is counted per process;
• partitions, UNION ALL members and subqueries, which may run under a Parallel Append that runs each child in only some of the workers.

Rows and cost of the parallel path are derived from the non-parallel estimate divided by the number of workers. The non-parallel plans of 16 queries are unchanged with both the Postgres planner and ORCA.

Tests (features/parallel, FDW mode): result correctness, single fragment, rescan, Parallel Append, option precedence, reject limit, and EXPLAIN. The option is documented in fdw/PXF_FDW.md.

Known limitations:

• The number of workers is max_parallel_workers_per_gather, because PXF does not know the data size at plan time.
• Cloudberry does not plan a parallel partial HashAggregate over randomly distributed input, so GROUP BY and count(DISTINCT) queries use the non-parallel scan.
• Each worker opens its own request to PXF, so W workers put W times the concurrent load on the PXF server.

Contributor's checklist

Here are some reminders before you submit your pull request:

• Make sure that your Pull Request has a clear title and commit message. You can take the Git commit template as a reference.
• Learn the code contribution and doc contribution guides for better collaboration.
• Make sure that CICD workflow is successful.
• List your communications in the GitHub Issues or Discussions (if has or needed).
• Feel free to ask for the cloudberry committers or other people to help review and approve.

🤖 Generated with Claude Code

[MisterRaindrop]

@ostinru
Thanks for the review. The current code is not yet ready for the review stage.

My approach to kernel parallel processing is still too simplistic. Maybe I will change or refactor later.

[MisterRaindrop]

All deployments are local.
Sizes: 100MB, 1GB, 10GB
Workers: 4
Format：csv

Size	Rows	Workers	COUNT seq (ms)	COUNT par (ms)	Speedup	SUM seq (ms)	SUM par (ms)	Speedup
100MB x 20file	487,700	4	282	311	0.91x	290	188	1.54x
1GB x 20file	4,994,140	4	2352	1514	1.55x	2448	1314	1.86x
10GB x 20file	49,941,480	4	21524	11589	1.86x	21954	11547	1.90x

When exploring parallelization, the good news is that parallelization does indeed improve efficiency. For small data volumes, the improvement is not obvious and may even be less efficient than non-parallel processing. Only when the data volume is large does it show a noticeable improvement.

However, the current improvement still falls short of the expected level. Theoretically, the speedup factor should be almost equal to the number of workers. The reason it hasn’t reached the expected level may be due to bottlenecks in I/O or CPU. Further exploration will be conducted in the future.

[my-ship-it]

The plan seems not correct?

[MisterRaindrop]

You're right, it wasn't. That expected output was written by hand: a PostgreSQL Gather with Workers Planned under a one-segment Gather Motion. Cloudberry plans a parallel pxf_fdw scan as gang-parallel workers under the Gather Motion, so there is no Gather node and no Workers Planned, and CI runs on three segments.

3981ccb replaced it with the plan Cloudberry actually produces:

 Finalize Aggregate
   ->  Gather Motion 12:1  (slice1; segments: 12)
         ->  Partial Aggregate
               ->  Parallel Foreign Scan on pxf_parallel_enabled

It uses EXPLAIN (COSTS OFF) and normalizes the Gather Motion size, so the test doesn't depend on the number of segments. ParallelScanTest.testParallelExplain passes in the features_fdw jobs of the latest CI run. Sorry for the slow reply.

[MisterRaindrop]

A correction to my reply above: testParallelExplain passing in CI didn't show anything. atmsort.pl sends statements that look like EXPLAIN ... SELECT to explain.pl, which isn't installed in the CI image, so the actual and expected output were both empty. 8757e6e returns the plan from a function as ordinary query rows, so it is really compared now. The expected plan is the same as before. I checked that a wrong expected plan fails, with and without explain.pl.

[ostinru]

The current code is not yet ready for the review stage.

@MisterRaindrop , is it ready for review?

[MisterRaindrop]

The current code is not yet ready for the review stage.

@MisterRaindrop , is it ready for review?

it is ready for review

[tuhaihe]

Review by ocr delegate model (backed by Claude Opus 5.5), FYI

---

review-comment.md
test-report.md

[MisterRaindrop]

Thanks @tuhaihe, the review was useful and the repros made each item easy to check. Addressed in 8757e6e, f4a487c, ce0afaf and d722f96:

1. Reject limit per worker. No parallel path is offered for a table with a reject_limit. New test features/parallel/reject_limit: 12 files with one bad row each; limit 3 still fails with parallel enabled, limit 20 succeeds, uses a plain scan and returns the 1200 good rows. The old build fails this test.

2. Costing and row estimate. The 30x gap comes from adjust_costs_for_mpp_foreign_scan() in allpaths.c, which multiplies the rows of a strewn ForeignPath by numsegments * 10, but only for the pathlist, not the partial pathlist. The partial path now starts from the same estimate and divides it by the workers (get_parallel_divisor() is the worker count on Cloudberry, as there is no leader). Both paths charge cpu_tuple_cost per estimated row, and the parallel path divides it by the workers.

What this changes for GROUP BY: the parallel path is no longer chosen. Cloudberry does not plan a parallel partial HashAggregate over randomly distributed input, even with the parallel costs set to 0 (a randomly distributed heap table behaves the same way). So the plan is now the non-parallel Streaming Partial HashAggregate: no slowdown on CSV, but also no speedup on Parquet. COUNT/SUM, filters and joins still use the parallel scan. I did not size the worker count or skip small tables, as PXF does not know the data size at plan time; the docs say a small table can be slower.

3. Rescan test. I couldn't put a partial path on a rescanned side either, so I simplified the rescan to close the import and let the next iterate start a new one, and said in a comment that a partial path is never rescanned.

4. Duplicated import setup. InitCopyStateVirtual and PxfBridgeImportStartVirtual are gone. PxfBridgeImportStart overrides the segment headers when gang_parallel is set.

5. PxfCopyFromErrorCallback. Restored to main, together with pxfEndForeignScan and the iterate loop.

Nits. worker_index, the bare scope and utils/guc.h are removed. The cost change is in the commit message: 32 plans (16 queries, both optimizers) are the same as with 2.2.0. enable_parallel is documented in fdw/PXF_FDW.md. I left 77ff52c and 0bbe260 as they are to avoid a force-push during the review; they can be reworded or squashed before merge.

One more fix: the EXPLAIN tests compared nothing. atmsort.pl hands any statement matching explain ... select to explain.pl, which the CI image doesn't have, so both sides were empty. The plan is now returned by a function as query rows, and a wrong expected plan fails the test.

Happy to have the benchmark re-run on d722f96.

[tuhaihe]

Thanks for the update! Below are a re-review and a cluster benchmark of the new head (7f708c1), both prepared with Claude Code.

The results look good: COUNT/SUM still get ~1.8x (CSV) / ~1.7x (Parquet) at 4 workers, the reject_limit issue is fixed, and CSV GROUP BY is no longer slower than main. The one trade-off worth noting is that Parquet GROUP BY no longer uses a parallel scan (1.70x → ~1.0x); details and a few minor suggestions are in the review below.

Thanks for the great work!

Re-review (review-comment-round2.md)

Re-review of the new head (7f708c1)

Thanks for the quick turnaround. I re-reviewed fa855b3..7f708c1 and re-ran the benchmark from the last round on the same EC2 setup, against both main and the previous head. Details are in the attached test report.

Summary: every item from the last review is addressed, and I found nothing blocking. COUNT/SUM speedups are unchanged, the reject-limit bug is fixed, and CSV GROUP BY is no longer slower than main. The one real trade-off is that Parquet GROUP BY loses its speedup (1.70x → 0.98x at 4 workers). I'd like the maintainers to agree to that explicitly before merge.

Last round's items

Item	Status
reject_limit enforced per worker	Fixed. No partial path when reject_limit is set. Checked at p4/p8: the rows limit and percent limit both error like main, and LOG ERRORS logs exactly 48 rows.
Partial-path row estimate too low (CSV GROUP BY slower)	Fixed. I checked adjust_costs_for_mpp_foreign_scan() in the kernel and agree with your analysis. CSV GROUP BY at p2 went from 0.92x to 0.98x (now the same plan as main).
Small tables always go parallel	Unchanged, documented. A 100-row table costs about +3 ms with parallel on. Fine given the new docs section.
Duplicated *Virtual functions	Fixed. The header override in PxfBridgeImportStart is much cleaner.
Rescan	Fixed. The four rescan queries match non-parallel results.
Error callback rewrite	Fixed. Restored to main.
Nits / docs	Fixed.

Good catch on the EXPLAIN tests comparing nothing; the plpgsql wrapper is a nice fix.

Remaining points

1. Parquet GROUP BY regression vs the previous head (needs a decision). With the corrected estimate, the planner no longer chooses a parallel scan under any GROUP BY. That is the right result for CSV, where the old plan spilled to a Sort, but on Parquet the old plan was 1.70x faster (39.7 s → 23.4 s on 10 GB). I think shipping the conservative version is reasonable, since COUNT/SUM/filters still get about 1.7–1.8x. Could you add one line to the Parallel Scan docs saying aggregates with GROUP BY usually won't use a parallel scan? A follow-up issue to revisit GROUP BY costing would also help.
2. EndForeignScan now calls churl_cleanup for every scan, not only parallel ones. I measured this with interleaved base/new runs. LIMIT 10 is about +2.5 ms (59 → 61.5 ms). LIMIT 100000, full scans and cancels show no change. The PXF "Remote connection closed by the client" WARN count is identical. No action needed; I'd mention it in the commit message since it changes behaviour on the non-parallel path.
3. SegmentRows() mirrors a kernel heuristic. If the kernel ever applies the numsegments * 10 factor to partial paths too, the rows would be scaled twice. The comment covers it; maybe also point to adjust_costs_for_mpp_foreign_scan by name so it's easy to find.
4. Nit: the expected output for parallel/reject_limit still contains a CI machine path (/pr61ci/parallel_bad_rows) in a CONTEXT line. Matchsubs hides it, but regenerating it would be cleaner.
5. Commit history: please squash or reword 77ff52c and 0bbe260 before merge, as you suggested.

From my side this is ready to merge once point 1 is agreed and points 4–5 are cleaned up.

Benchmark report (test-report-round2.md)

PR #61 benchmark, round 2: main vs previous head vs new head

Setup

• Machine: EC2 m6i.4xlarge (16 vCPU Xeon 8375C, 61 GiB), Rocky 9, 300 GB gp3. Same box type as round 1.
• Cloudberry: apache/cloudberry cd260a80 (same as round 1), release build (-O3, no cassert). Demo cluster with 3 primaries and no mirrors.
• PXF: server from main (40961b7), -Xmx4g, local file server. Only pxf_fdw.so was swapped between builds, each followed by gpstop -ar:
  • base: 40961b7
  • old: fa855b3
  • new: 7f708c1
• Method: optimizer=off, 1 warm-up and 5 timed runs, median reported. pN = enable_parallel=on, max_parallel_workers_per_gather=N.
• Order: base → old → new → base again. The final base rerun was within 0.7% (CSV) and 1.3% (Parquet) of the first one.
• Correctness: query results were identical across every build and setting (md5 of the result rows).

Datasets

• csvX_N: CSV data, total size X, split into N files.
• pq10g_24: 10 GB Parquet in 24 files.

The queries are count(*), sum(...), and GROUP BY grp with count and sum.

Speedup vs main at 4 workers, previous head vs new head

Dataset	count (old → new)	sum (old → new)	GROUP BY (old → new)
csv100m_24	1.33 → 1.33	1.34 → 1.35	1.03 → 1.00
csv1g_24	1.71 → 1.68	1.56 → 1.60	1.00 → 0.96
csv10g_24	1.81 → 1.83	1.85 → 1.85	1.01 → 0.98
csv10g_3	1.81 → 1.82	1.84 → 1.86	1.01 → 0.98
csv10g_12	1.81 → 1.83	1.83 → 1.86	1.01 → 0.98
csv10g_48	1.81 → 1.81	1.82 → 1.85	1.00 → 0.98
pq10g_24	1.69 → 1.71	1.67 → 1.67	1.70 → 0.98

At 2 and 8 workers:

Query	p2 (old → new)	p8 (old → new)
csv10g_24 count	1.56 → 1.62	1.79 → 1.79
csv10g_24 GROUP BY	0.92 → 0.98	1.07 → 0.98
pq10g_24 GROUP BY	1.48 → 0.99	1.79 → 0.99

Absolute times, median ms (new head)

Dataset	Query	base	new, no parallel	new p2	new p4	new p8	old p4
csv10g_24	count	17,950	18,935	11,091	9,805	10,012	9,905
csv10g_24	sum	19,013	19,365	11,930	10,260	10,369	10,270
csv10g_24	groupby	20,965	21,475	21,288	21,291	21,308	20,692
pq10g_24	count	39,946	40,470	28,194	23,408	23,226	23,586
pq10g_24	sum	39,918	40,209	29,560	23,911	23,656	23,910
pq10g_24	groupby	39,725	40,211	39,931	40,401	39,935	23,407
csv1g_24	count	2,057	2,006	1,978	1,223	1,093	1,206
csv100m_24	count	249	265	218	187	189	188

The new head looks about 3–5% slower than base without parallel on CSV, but that did not reproduce. An interleaved A/B run (base, new, base, new) on count(*) over csv10g_24 gave 17,973 / 17,935 ms on base and 18,011 / 17,917 ms on new. The gap in the table comes from when the runs happened, not from the PR.

GROUP BY plans (pq10g_24, p4)

• base: Gather 3:1 → Sort → Finalize HashAgg → Redistribute 3:3 → Streaming Partial HashAgg → Foreign Scan
• old head: Gather 12:1 → GroupAggregate → Sort → Redistribute 12:12 → Parallel Foreign Scan (rows=250)
• new head: same as base (Foreign Scan cost 50000..50300). GROUP BY no longer uses a parallel scan.
• new head, COUNT/SUM: Finalize Agg → Gather 12:1 → Partial Agg → Parallel Foreign Scan (rows=7500)

CPU (average %, out of 1600)

Query	base	new p4	old p4
csv10g_24 sum	java 111 / postgres 287	java 382 / postgres 986	java 397 / postgres 991
pq10g_24 count	java 298 / postgres 146	java 923 / postgres 460	java 917 / postgres 446

The CPU profile is the same as with the previous head: CSV is bound on the postgres side, Parquet on the Java side. 8 workers is no faster than 4.

Functional checks

Check	Result
R1 reject_limit	Fixed. At p4/p8 the plan is a plain Foreign Scan. A rows limit of 10 and a percent limit of 2% both error like main (the previous head returned 23,952 and 2,328 rows instead). With LOG ERRORS and limit 100: 23,952 rows and 48 errors logged in every configuration.
R2 small tables	Still parallel, as documented. 100-row table: 25.0 ms parallel vs 21.8 ms not. 1 MB table: 29.8 vs 27.3 ms.
R3 non-parallel plans	No shape change. With optimizer=off, each PXF Foreign Scan's cost goes up by 300; nothing else changes. With ORCA, plans are byte-identical to main.
R4 end-of-scan cleanup	LIMIT 10: about +2.5 ms (59.3/59.0 → 61.4/61.9 ms). LIMIT 100000 and full scans: no change. Same PXF WARN count (126 "Remote connection closed by the client") on both builds. Cancels with statement_timeout (non-parallel, p4 CSV, p8 Parquet) were clean: no leftover backends, and the next query worked. No PANIC or FATAL.
R5 rescan	All four rescan queries match the non-parallel results.
R6 UNION ALL / partitions / hash join / skew	Results match non-parallel. Partitioned tables run their children as non-parallel scans under Parallel Append.

[tuhaihe]

+1 for LICENSE. Thanks!