fix: do not divide parallel custom-scan I/O by worker count, with the leader arm covered (#1065, rebased + coverage)

CHANGELOG.md

@@ -615,26 +615,16 @@ true until the next version shipped.
   This is the third arm in this file to be repaired for counting a string across a
   whole file. The `deltuples` comment 15 lines above records the first, fixed by
   scoping; these two were left as whole-file counts and did the same thing again.
-- A fetching index scan on a correlated key stayed cheaper than the custom scan
-  through ~50,000 rows, while doing about 27x the work (#913).
-
-  The index-fetch penalty prices distinct row-group decodes. On a clustered key
-  that count is ceil(rows / stripe), so it does not grow through the first group.
-  Core's heap-fetch cost grows with rows; the extra columnar work of reconstructing
-  each fetched row after the group is cached did not. The 50,000-row range then
-  stayed on the index (cost 2243 against the custom scan's 2504) while the point
-  lookup was already correctly on it.
-
-  The per-row term is `cpu_tuple_cost * rows * decodeUnits`, the same units #503
-  uses for a projection. It is not a conversion from heap instructions-per-cost:
-  #766 showed that conversion predicts the wrong winner. Uncapped it grows with
-  the whole table and costs a clustered ORDER BY off its index (#355). Cap it at
-  half a group: that is enough to move the 50,000-row range and small enough to
-  leave the ordered scan on the index.
-
-  Plan choice is the property, not a cost number: at 50,000 rows the planner now
-  picks the custom scan; a point lookup still uses the index; a clustered ORDER BY
-  stays on the index; both paths return the same aggregate.
+- A parallel custom scan divided its whole run cost by the worker count.
+
+  Core seqscan divides CPU across workers and leaves disk I/O whole. The
+  partial columnar path divided `(total - startup)` by `workers`, so an
+  I/O-dominated scan was quoted at half its serial cost with two workers.
+  Measured: serial run 10825, parallel Custom Scan 5412.5 (ratio 2.000).
+  Leaving I/O undivided, the same fixture is 10112.5 (ratio 1.070).
+
+  `get_parallel_divisor` is static in core; the leader-participation heuristic
+  is reproduced so CPU uses the same divisor a parallel seqscan does.
 
 - `compare_to_bash.py`'s corpus arm called a WRAPPED name fabricated. A name too long
   for one line is written as adjacent literals, and Python joins them at parse time,

src/columnar_customscan.c

@@ -2546,6 +2546,47 @@ pgcolumnar_refined_scan_cost(RelOptInfo *rel, Oid relid, Path *seqpath,
 	*out_total = startup + run * survival;
 }
 
+/*
+ * pgcolumnar_parallel_divisor
+ *		Core's get_parallel_divisor (costsize.c) is static, so the
+ *		leader-participation heuristic is reproduced here. CPU is divided
+ *		the same way a parallel seqscan is; I/O is not.
+ */
+static double
+pgcolumnar_parallel_divisor(Path *path)
+{
+	double		parallel_divisor = path->parallel_workers;
+
+	if (parallel_leader_participation)
+	{
+		double		leader_contribution;
+
+		leader_contribution = 1.0 - (0.3 * path->parallel_workers);
+		if (leader_contribution > 0)
+			parallel_divisor += leader_contribution;
+	}
+
+	return parallel_divisor;
+}
+
+/*
+ * pgcolumnar_scan_io_run_cost
+ *		The page-read portion of a columnar scan, after projected-width
+ *		scaling (#171) and zone-map survival (#434). This is the term core
+ *		leaves undivided on a parallel seqscan.
+ */
+static Cost
+pgcolumnar_scan_io_run_cost(RelOptInfo *rel, Oid relid)
+{
+	double		survival = pgcolumnar_zonemap_survival(rel, relid);
+	double		widthFrac = pgcolumnar_projected_width_fraction(rel, relid);
+	Cost		pageCost = seq_page_cost * (double) rel->pages;
+
+	if (widthFrac < 1.0)
+		pageCost *= widthFrac;
+	return pageCost * survival;
+}
+
 /*
  * PgColumnarSetRelPathlist
  *		set_rel_pathlist_hook: for a columnar base relation, replace the
@@ -2874,7 +2915,7 @@ PgColumnarSetRelPathlist(PlannerInfo *root, RelOptInfo *rel, Index rti,
 	 * Add a parallel-aware partial path (gap 23) so the planner can put a Gather
 	 * over a parallel columnar scan. Workers each claim distinct stripes from a
 	 * shared counter set up by the DSM callbacks. The cost model mirrors a
-	 * parallel seqscan: the per-tuple work is divided among the workers.
+	 * parallel seqscan: CPU is divided among the workers, disk I/O is not.
 	 *
 	 * Costed from the serial columnar path rather than from the seqscan, and no
 	 * longer conditional on a seqscan surviving (#362). add_path frees the
@@ -2979,7 +3020,10 @@ PgColumnarSetRelPathlist(PlannerInfo *root, RelOptInfo *rel, Index rti,
 		if (workers > 0)
 		{
 			CustomPath *ppath = makeNode(CustomPath);
-			double		divisor = (double) workers;
+			double		divisor;
+			Cost		serialRun;
+			Cost		ioRun;
+			Cost		cpuRun;
 
 			ppath->path.pathtype = T_CustomScan;
 			ppath->path.parent = rel;
@@ -2988,10 +3032,23 @@ PgColumnarSetRelPathlist(PlannerInfo *root, RelOptInfo *rel, Index rti,
 			ppath->path.parallel_aware = true;
 			ppath->path.parallel_safe = true;
 			ppath->path.parallel_workers = workers;
-			ppath->path.rows = rel->rows / divisor;
+			/*
+			 * Core seqscan divides CPU by get_parallel_divisor and leaves disk
+			 * I/O whole (costsize.c: the disk run cost cannot be amortized).
+			 * This path used to divide the entire (total - startup) by the
+			 * worker count, so an I/O-dominated scan was quoted at 1/N of its
+			 * serial cost.
+			 */
+			divisor = pgcolumnar_parallel_divisor(&ppath->path);
+			serialRun = serialTotalCost - serialStartupCost;
+			ioRun = pgcolumnar_scan_io_run_cost(rel, rte->relid);
+			if (ioRun > serialRun)
+				ioRun = serialRun;
+			cpuRun = serialRun - ioRun;
+			ppath->path.rows = clamp_row_est(rel->rows / divisor);
 			ppath->path.startup_cost = serialStartupCost;
 			ppath->path.total_cost = serialStartupCost +
-				(serialTotalCost - serialStartupCost) / divisor;
+				ioRun + cpuRun / divisor;
 			ppath->path.pathkeys = NIL;
 			ppath->flags = 0;
 			ppath->custom_paths = NIL;

test/check_ledger.tsv

@@ -1261,3 +1261,16 @@ parallel_am_scan	parallel_am_scan	premise: the serial plan is not a columnar cus
 parallel_am_scan	parallel_am_scan	premise: the table holds every inserted row	15;16;17;18;19	never	-
 parallel_am_scan	parallel_am_scan	premise: with the custom scan off the serial plan is a Seq Scan	15;16;17;18;19	never	-
 parallel_am_scan	parallel_am_scan	workers share the table-AM scan, it is not a single claimer	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	an I/O-dominated parallel scan is not priced at serial/workers	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	an I/O-dominated parallel scan is not priced at serial/workers with the leader participating	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	premise: both leader settings produced a row estimate to compare	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	premise: both scans have a positive run cost	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	premise: the leader-on parallel scan has a positive run cost	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	premise: the leader-on plan is still a parallel columnar scan	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	premise: the parallel plan has Gather	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	premise: the parallel plan is a columnar scan	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	premise: the parallel plan uses two workers	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	premise: the serial plan has no Gather	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	premise: the serial plan is a columnar scan	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	premise: the table holds every inserted row	15;16;17;18;19	never	-
+parallel_scan_cost	parallel_scan_cost	the leader-participation branch changes the divisor	15;16;17;18;19	never	-

test/check_ledger_budget.txt

@@ -58,4 +58,4 @@ suites_not_covered 249
 # that is not this one. Neither survives. Re-derived by COUNTING on the merged tree,
 # which is the only resolution this number has:
 #     awk -F'\t' '$5=="never"' test/check_ledger.tsv | wc -l
-checks_never_observed_red 1255
+checks_never_observed_red 1268

test/parallel_scan_cost.sh

@@ -0,0 +1,165 @@
+#!/usr/bin/env bash
+#
+# pgColumnar: a parallel custom scan must not divide I/O by the worker count.
+#
+# The partial path prices itself as serial_startup + (serial_run / workers).
+# Core seqscan divides CPU only and leaves disk I/O whole (costsize.c: "the disk
+# run cost cannot be amortized at all"). Dividing the whole run quotes an
+# I/O-dominated scan at 1/N of its serial cost, so Gather wins against honestly
+# costed alternatives.
+#
+# This suite pins the PLANNER number, not a runtime. Costs move with the
+# constants; the property is that an I/O-only run is not halved when two
+# workers are granted. Independent of test/pytest/test_parallel_scan_cost.py:
+# same public seam (EXPLAIN of a columnar scan), own fixture, own observations.
+#
+# Usage:  test/parallel_scan_cost.sh [PG_CONFIG]
+# Written fresh for pgColumnar.
+
+set -uo pipefail
+. "$(dirname "${BASH_SOURCE[0]}")/lib.sh"
+pgc_setup "${1:-/usr/local/pg17/bin/pg_config}"
+
+N=40000
+psql_run "CREATE TABLE psc (id int, k int, payload text) USING pgcolumnar;"
+psql_run "INSERT INTO psc SELECT g, g%17, md5(g::text) FROM generate_series(1,$N) g;"
+psql_run "ALTER TABLE psc SET (parallel_workers = 2);"
+psql_run "ANALYZE psc;"
+
+# Session GUCs that isolate the I/O term: seq_page_cost is raised so the
+# serial run is mostly pages. Dividing the whole run by 2 then halves I/O;
+# leaving I/O whole leaves the run almost unchanged. CPU terms stay at their
+# defaults so the parallel path is still a little cheaper than serial and
+# Gather still wins -- zeroing CPU made the two paths equal and the planner
+# declined Gather, which hid the number this suite exists to read.
+# parallel_leader_participation is off so the divisor is the worker count.
+setg() { q "ALTER DATABASE $PGC_DB SET $1 = $2;" >/dev/null; }
+setg seq_page_cost 100
+setg parallel_setup_cost 0
+setg parallel_tuple_cost 0
+setg parallel_leader_participation off
+setg min_parallel_table_scan_size 0
+setg jit off
+
+explain_scan() {
+	# $1 = max_parallel_workers_per_gather
+	env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \
+		-d "$PGC_DB" -Atq \
+		-c "SET max_parallel_workers_per_gather = $1;" \
+		-c "EXPLAIN (COSTS ON) SELECT id, k, payload FROM psc;"
+}
+
+serial_plan="$(explain_scan 0)"
+par_plan="$(explain_scan 2)"
+
+scan_cost_pair() {
+	# startup and total of the first Custom Scan (PgColumnarScan) line.
+	echo "$1" | grep -F "Custom Scan (PgColumnarScan)" | head -1 \
+		| grep -oE "cost=[0-9.]+\\.\\.[0-9.]+" | head -1 \
+		| sed -E "s/cost=([0-9.]+)\\.\\.([0-9.]+)/\\1 \\2/"
+}
+
+s_pair="$(scan_cost_pair "$serial_plan")"
+p_pair="$(scan_cost_pair "$par_plan")"
+s_start="${s_pair%% *}"
+s_total="${s_pair##* }"
+p_start="${p_pair%% *}"
+p_total="${p_pair##* }"
+
+s_run="$(awk -v t="$s_total" -v s="$s_start" "BEGIN{ print t-s }")"
+p_run="$(awk -v t="$p_total" -v s="$p_start" "BEGIN{ print t-s }")"
+ratio="$(awk -v s="$s_run" -v p="$p_run" "BEGIN{ if (p<=0) print 0; else printf \"%.3f\", s/p }")"
+
+echo "-- serial Custom Scan cost=$s_start..$s_total run=$s_run"
+echo "-- parallel Custom Scan cost=$p_start..$p_total run=$p_run ratio=$ratio"
+
+check "premise: the table holds every inserted row" \
+	"$(q "SELECT count(*) FROM psc")" "$N"
+
+check "premise: the serial plan is a columnar scan" \
+	"$(echo "$serial_plan" | grep -c "Custom Scan (PgColumnarScan)")" "1"
+
+check "premise: the serial plan has no Gather" \
+	"$(echo "$serial_plan" | grep -c "Gather")" "0"
+
+check "premise: the parallel plan has Gather" \
+	"$(echo "$par_plan" | grep -c "Gather")" "1"
+
+check "premise: the parallel plan uses two workers" \
+	"$(echo "$par_plan" | grep -oE "Workers Planned: [0-9]+" | head -1 | grep -oE "[0-9]+")" "2"
+
+check "premise: the parallel plan is a columnar scan" \
+	"$(echo "$par_plan" | grep -c "Custom Scan (PgColumnarScan)")" "1"
+
+check "premise: both scans have a positive run cost" \
+	"$(awk -v s="$s_run" -v p="$p_run" "BEGIN{ print (s>0 && p>0) ? \"yes\" : \"no\" }")" "yes"
+
+# On the unfixed path the ratio is 2.000 (whole run / 2 workers). Core leaves
+# I/O whole, so with CPU terms zeroed the ratio stays near 1. 1.35 is
+# unreachable by dividing the whole run, and reachable only if I/O remains.
+check "an I/O-dominated parallel scan is not priced at serial/workers" \
+	"$(awk -v r="$ratio" "BEGIN{ print (r < 1.35) ? \"io-kept\" : \"halved\" }")" "io-kept"
+
+# ---- the same property with the leader participating, which is the default ----
+#
+# EVERYTHING ABOVE RUNS WITH parallel_leader_participation OFF. That makes the
+# divisor exactly the worker count and leaves the
+# `if (parallel_leader_participation)` arm of pgcolumnar_parallel_divisor
+# unexecuted -- so the copied heuristic was covered only in the configuration
+# nobody runs. The GUC defaults ON.
+#
+# Reported in review of #1065.
+setg parallel_leader_participation on
+par_plan_on="$(explain_scan 2)"
+
+# The row estimate is the observable that proves the branch ran: the partial path
+# divides rel->rows by the SAME divisor, so leader-on and leader-off cannot agree.
+# With two workers the divisor is 2 against 2 + (1 - 0.3*2) = 2.4, which is a
+# visible difference in the plan rather than an inference about the code.
+scan_rows() {
+	awk '/Custom Scan \(PgColumnarScan\)/ {
+		if (match($0, /rows=[0-9]+/)) {
+			print substr($0, RSTART + 5, RLENGTH - 5)
+			exit
+		}
+	}' <<-EOF
+	$1
+	EOF
+}
+
+rows_off="$(scan_rows "$par_plan")"
+rows_on="$(scan_rows "$par_plan_on")"
+
+check "premise: the leader-on plan is still a parallel columnar scan" \
+	"$(printf '%s' "$par_plan_on" | grep -c 'Custom Scan (PgColumnarScan)')" "1"
+
+check "premise: both leader settings produced a row estimate to compare" \
+	"$([ -n "$rows_off" ] && [ -n "$rows_on" ] && echo yes || echo no)" "yes"
+
+# If this ever reports "same", the leader-participation branch did not run and
+# every assertion below it is about the wrong divisor.
+check "the leader-participation branch changes the divisor" \
+	"$([ "$rows_off" != "$rows_on" ] && echo differs || echo same)" "differs"
+
+p_pair_on="$(scan_cost_pair "$par_plan_on")"
+p_start_on="${p_pair_on%% *}"
+p_total_on="${p_pair_on##* }"
+p_run_on="$(awk -v t="$p_total_on" -v s="$p_start_on" "BEGIN{ print t-s }")"
+ratio_on="$(awk -v s="$s_run" -v p="$p_run_on" \
+	"BEGIN{ if (p<=0) print 0; else printf \"%.3f\", s/p }")"
+
+echo "-- leader on: parallel Custom Scan cost=$p_start_on..$p_total_on run=$p_run_on ratio=$ratio_on"
+echo "-- rows: leader off=$rows_off leader on=$rows_on"
+
+check "premise: the leader-on parallel scan has a positive run cost" \
+	"$(awk -v p="$p_run_on" "BEGIN{ print (p>0) ? \"yes\" : \"no\" }")" "yes"
+
+# The bound is the same as the leader-off arm: dividing the WHOLE run would put
+# the ratio at the divisor, and the divisor is larger here, so a regression is if
+# anything easier to see in this configuration.
+check "an I/O-dominated parallel scan is not priced at serial/workers with the leader participating" \
+	"$(awk -v r="$ratio_on" "BEGIN{ print (r < 1.35) ? \"io-kept\" : \"halved\" }")" "io-kept"
+
+setg parallel_leader_participation off
+
+pgc_summary

test/pytest/TESTS.md

@@ -93,6 +93,7 @@ behaviour, the source of that number is named.
 - [45. test_native_fetch_coalesce.py: index fetch I/O is not per-column](#45-test_native_fetch_coalescepy-index-fetch-io-is-not-per-column)
 - [46. test_parallel_am_scan.py: a table-AM parallel scan must share work](#46-test_parallel_am_scanpy-a-table-am-parallel-scan-must-share-work)
 - [47. test_index_fetch_penalty_crossover.py: the correlated range must not fetch](#47-test_index_fetch_penalty_crossoverpy-the-correlated-range-must-not-fetch)
+- [48. test_parallel_scan_cost.py: a parallel custom scan must not divide I/O](#48-test_parallel_scan_costpy-a-parallel-custom-scan-must-not-divide-io)
 
 ## 1. How to read a test in here
 
@@ -4385,18 +4386,6 @@ Public seam: `EXPLAIN ANALYZE` worker rows on a Parallel Seq Scan. Leader
 participation is off so the two launched workers are the claimers under
 test. The shell twin uses its own table (`pam`, 50000 rows, groups of 100);
 this file uses `ampar`, 80000 rows, groups of 200. Assertion names match.
-
-### Every arm
-
-| test | what it holds |
-| --- | --- |
-| `test_parallel_am_scan` | the serial plan is a Seq Scan, not a custom scan; the parallel plan is a Seq Scan under Gather with two workers launched; a parallel AM scan returns the same count as serial; both launched workers produced rows |
-| `test_a_parallel_index_build_covers_the_whole_table` | a parallel index build requests workers and indexes every row -- compared as count and SUM through the index against a sequential scan, because a group read twice cancelling a group skipped leaves the count right |
-
-The load-bearing assertion is `workers share the table-AM scan, it is not a
-single claimer`. It is unreachable while `phs_nallocated` is first-wins, and
-reachable only when each worker claims its own row groups.
-| `test_index_fetch_penalty_crossover` | a 50,000-row correlated range uses the custom scan; a point lookup still uses the index; both paths agree on the aggregate; a clustered ORDER BY stays on the index |
 ## 47. test_index_fetch_penalty_crossover.py: the correlated range must not fetch
 
 #913. A fetching index scan on a correlated key is priced below the custom scan
@@ -4413,4 +4402,34 @@ fixture, own observations. Assertion names match the shell suite.
 | test | what it asserts |
 | --- | --- |
 | `test_native_chunk_length_bound` | a point lookup uses the index and returns the row; after `page_length` grows by 2^32, both the fetch and a sequential scan raise XX001 and the backend survives each |
+| `test_parallel_scan_cost` | the serial plan is a columnar scan with no Gather; the parallel plan is a columnar scan under Gather with two workers; both have a positive run cost; an I/O-dominated parallel scan is not priced at serial/workers |
+
+The load-bearing assertion classifies the ratio `serial_run / parallel_run` as
+`io-kept` (below 1.35) rather than `halved` (2.000 on the unfixed path). It is
+unreachable by dividing the whole run, and reachable only if I/O remains.
+## 48. test_parallel_scan_cost.py: a parallel custom scan must not divide I/O
+
+The port of `test/parallel_scan_cost.sh`. The partial path priced itself as
+`serial_startup + (serial_run / workers)`. Core seqscan divides CPU only and
+leaves disk I/O whole. Dividing the whole run quotes an I/O-dominated scan at
+half its serial cost with two workers, so Gather beat honestly costed
+alternatives.
+
+Public seam: `EXPLAIN` of a columnar scan. This file raises `seq_page_cost` so
+I/O dominates the serial run; the shell twin does the same with a different
+page-cost and its own table. CPU terms stay at their defaults so the parallel
+path is still a little cheaper than serial and Gather still appears -- the
+number this suite exists to read. Assertion names match the shell suite.
+
+### Every arm
+
+| test | what it holds |
+| --- | --- |
+| `test_parallel_am_scan` | the serial plan is a Seq Scan, not a custom scan; the parallel plan is a Seq Scan under Gather with two workers launched; a parallel AM scan returns the same count as serial; both launched workers produced rows |
+| `test_a_parallel_index_build_covers_the_whole_table` | a parallel index build requests workers and indexes every row -- compared as count and SUM through the index against a sequential scan, because a group read twice cancelling a group skipped leaves the count right |
+
+The load-bearing assertion is `workers share the table-AM scan, it is not a
+single claimer`. It is unreachable while `phs_nallocated` is first-wins, and
+reachable only when each worker claims its own row groups.
+| `test_index_fetch_penalty_crossover` | a 50,000-row correlated range uses the custom scan; a point lookup still uses the index; both paths agree on the aggregate; a clustered ORDER BY stays on the index |
 

test/pytest/expected_tests.txt

@@ -238,4 +238,4 @@ guard_tests 346
 # the expected answer for a file that needs a cluster, and checking it was the point
 # rather than assuming it.
 # cluster_tests re-derived by collection on the rebased tree, never by adding a delta measured on another tree.
-cluster_tests 416
+cluster_tests 417