fix: share table-AM parallel scan groups across workers (#1068, rebased + ledger)

CHANGELOG.md

@@ -615,87 +615,18 @@ 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.
-- An index fetch pinned once per projected column, while a sequential scan
-  already coalesced adjacent chunk ranges into one read.
-
-  `pgcolumnar_fetch_row` issued two `PgColumnarReadLogicalData` calls per
-  column (validity bitmap, then the value stream). The scan path
-  (`pgcolumnar_native_read_projected`) sorts those ranges and merges the ones
-  that touch. Adjacent columns in a row group are laid out back to back, so a
-  wide btree fetch of a small group pinned the same pages once per column.
-
-  Measured on PostgreSQL 18 with `EXPLAIN (ANALYZE, BUFFERS)` executor pins
-  (planning excluded): 16 int columns, one row via the index, 64 pins for one
-  column and 94 for sixteen -- exactly two extra pins per extra column. After
-  the fetch path coalesces the same way the scan does, both counts are 61.
-  New twins `native_fetch_coalesce` and `test_native_fetch_coalesce.py`.
-
-  THE VALIDITY COPY IS BOUNDED BY THE CHUNK BEFORE IT RUNS. The coalesced path
-  copies `validityBytes` out of a span buffer that is only guaranteed to hold
-  `page_length` bytes for the chunk being served, and the test reconciling the
-  two ran three lines AFTER the copy. A chunk whose catalog `page_length` was
-  smaller than its validity bitmap therefore read past the allocation.
-
-  Reproduced against a build with `-fsanitize=address`, by poisoning
-  `pgcolumnar.column_chunk.page_length` on the last chunk by `page_offset` and
-  issuing a plain index-scan `SELECT`:
-
-      AddressSanitizer: heap-buffer-overflow
-      READ of size 625, 0 bytes after a 2640-byte region
-        pgcolumnar_fetch_coalesce_read   (the memcpy)
-        pgcolumnar_fetch_row
-        printtup
-
-  The backend died and the cluster entered recovery. Main cannot have this
-  shape: its non-coalesced fill reads straight from storage into an
-  exactly-sized destination, so no in-memory extent exists to exceed. The span
-  buffer and the copy out of it are both new here.
-
-  Hoisting the `page_length >= validityBytes` test above the copy closes it. An
-  inconsistent chunk is left for the non-coalesced path, which refuses it.
-
-  The regression arm is an ORDERING pin, not a behavioural one, and that is
-  deliberate: reading ~117 bytes past a palloc'd span reads adjacent heap and
-  returns quietly without a sanitizer, so a behavioural arm would report PASS
-  on the broken code. Both harnesses assert the order, each reading the source
-  its own way -- awk over line numbers in the shell suite, a regex over
-  character offsets in the pytest twin. Proved by MOVING the guard below the
-  copy rather than deleting it, which leaves both statements present and
-  reddens only the ordering arm.
-
-  AND A CHUNK THE CHECKED DECODE PATH WOULD REFUSE IS LEFT FOR IT, so the refusal
-  keeps its SQLSTATE. The range-building loop now defers any chunk whose
-  page_length is under the validity bitmap or whose value stream would not fit a
-  uint32.
-
-  Without that, this change SHADOWS #1063's typed refusal. `pgcolumnar_fetch_row`
-  calls the coalescing helper before the per-column loop reaches
-  `pgcolumnar_chunk_value_bytes`, and the helper builds its ranges straight from
-  `page_length`, so a poisoned length spans ~4GB and palloc raises first.
-  Measured on the two composed:
-
-      without the defer   native_chunk_length_bound   5 passed + 1 failed
-                          ERROR: invalid memory alloc request size 4294971754
-      with the defer      native_chunk_length_bound   6 passed + 0 failed  (XX001)
-      with the defer      native_fetch_coalesce       7 passed + 0 failed
-
-  The last line matters: the wide-fetch pin still passes, so deferring the
-  inconsistent chunk is not disabling coalescing to make a test green.
-
-  Reported by @jdatcmd, who composed the two branches rather than reading them.
-
-  THE ORDERING ARM IS ANCHORED ON THE CONTAINMENT TEST, because the function now
-  holds two guards with the same text -- the deferral above and the bound on the
-  copy. An unanchored search finds the first, which is in the wrong loop, and the
-  arm would then pass with the bound deleted. The containment test belongs only
-  to the distribution loop. Proved by deleting ONLY that guard and leaving the
-  deferral: both arms redden.
-
-  The two source patterns use bracket expressions rather than backslash-escaped
-  parens. `awk -v` processes escapes in the value and `\(` is undefined, so mawk
-  keeps the backslash and matches while gawk strips it -- silently not matching
-  for one pattern, and exiting fatally on `Unmatched (` for the other. CI runners
-  carry gawk. Verified identical under both.
+- A table-AM parallel scan was a single claimer.
+
+  `pgcolumnar_read_start` treated `phs_nallocated` as a first-wins flag: the
+  first participant loaded every row group and the others marked themselves
+  exhausted. Workers launched, then sat idle while one backend (usually the
+  leader) read the table. The custom-scan path already claims distinct groups
+  from a shared counter; the AM path now uses `phs_nallocated` the same way,
+  as a group index, not a mutex.
+
+  Measured with the custom scan off, two workers, and leader participation
+  off: both workers produced rows (19000 and 31000 of 50000). Restoring
+  first-wins returns one worker to 0.
 
 - `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,
@@ -2420,6 +2351,7 @@ true until the next version shipped.
 
 ### Fixed
 
+
 - The standing parity arm graded a hand-written list, and nothing enforced it
   (#432, #1046).
 

src/columnar_reader.c

@@ -986,9 +986,12 @@ PgColumnarRuntimeGroupsRemoved(PgColumnarReadState *readState)
 
 /*
  * pgcolumnar_read_start
- *		Lazily load the stripe list on the first fetch. For a parallel scan a
- *		single worker claims the whole scan and the others see it exhausted,
- *		which is a correct (if not parallel-accelerated) behaviour.
+ *		Lazily load the stripe list on the first fetch. Every parallel
+ *		participant loads the group list; work is claimed per group in
+ *		pgcolumnar_next_group_index from phs_nallocated, the same way the
+ *		custom scan claims from its DSM counter. The old first-wins use of
+ *		that counter left one backend (usually the leader) to read every
+ *		group and the launched workers idle.
  */
 static void
 pgcolumnar_read_start(PgColumnarReadState *readState)
@@ -998,16 +1001,6 @@ pgcolumnar_read_start(PgColumnarReadState *readState)
 
 	readState->started = true;
 
-	if (readState->parallelScan != NULL)
-	{
-		ParallelBlockTableScanDesc bpscan =
-			(ParallelBlockTableScanDesc) readState->parallelScan;
-		uint64		claim = pg_atomic_fetch_add_u64(&bpscan->phs_nallocated, 1);
-
-		if (claim != 0)
-			readState->exhausted = true;
-	}
-
 	if (!readState->exhausted)
 	{
 		/*
@@ -3192,20 +3185,28 @@ PgColumnarReadFoldColumn(PgColumnarReadState *readState, int attidx,
  *		The next native row group to scan, or -1 when none remain. The native
  *		counterpart of pgcolumnar_next_stripe_index: a parallel custom scan claims
  *		it from the shared atomic so each worker reads distinct row groups (gap
- *		23, D6e); a serial scan walks rowGroupIndex.
+ *		23, D6e); a table-AM parallel scan claims from phs_nallocated the same
+ *		way; a serial scan walks rowGroupIndex.
  */
 static int64
 pgcolumnar_next_group_index(PgColumnarReadState *readState)
 {
 	int			ngroups = list_length(readState->rowGroupList);
-	uint32		gi;
+	uint64		gi;
 
 	if (readState->parallelCounter != NULL)
 		gi = pg_atomic_fetch_add_u32(readState->parallelCounter, 1);
+	else if (readState->parallelScan != NULL)
+	{
+		ParallelBlockTableScanDesc bpscan =
+			(ParallelBlockTableScanDesc) readState->parallelScan;
+
+		gi = pg_atomic_fetch_add_u64(&bpscan->phs_nallocated, 1);
+	}
 	else
-		gi = (uint32) readState->rowGroupIndex++;
+		gi = (uint64) readState->rowGroupIndex++;
 
-	return (gi < (uint32) ngroups) ? (int64) gi : -1;
+	return (gi < (uint64) ngroups) ? (int64) gi : -1;
 }
 
 void

src/columnar_tableam.c

@@ -749,7 +749,7 @@ pgcolumnar_scan_getnextslot(TableScanDesc sscan, ScanDirection direction,
 }
 
 /* -------------------------------------------------------------------------
- * parallel scan: single-worker claim (see pgcolumnar_reader.c)
+ * parallel scan: shared group claim via phs_nallocated (see pgcolumnar_reader.c)
  * ------------------------------------------------------------------------- */
 
 static Size
@@ -1983,8 +1983,8 @@ pgcolumnar_index_build_range_scan(Relation table_rel, Relation index_rel,
 	/*
 	 * Obtain the reader. A parallel index build passes the TableScanDesc it
 	 * opened with table_beginscan_parallel; that scan already holds a reader
-	 * bound to the shared parallel scan, whose single-participant claim (see
-	 * pgcolumnar_read_start) makes exactly one participant read the whole table.
+	 * bound to the shared parallel scan, whose per-group claim (see
+	 * pgcolumnar_next_group_index) hands each participant distinct row groups.
 	 * We must read through that reader, not a private one: a private full-table
 	 * reader in every participant would index every row once per participant,
 	 * producing duplicate (key, TID) entries. When no scan is supplied (a serial

test/check_ledger.tsv

@@ -1241,3 +1241,17 @@ native_join_vector_agg	native_join_vector_agg	unique join fold answer equals GUC
 native_join_vector_agg	native_join_vector_agg	unique join fold answer equals heap	15;16;17;18;19	never	-
 native_join_vector_agg	native_join_vector_agg	unique join uses core Agg when GUC off	15;16;17;18;19	never	-
 native_join_vector_agg	native_join_vector_agg	unique join uses vectorized agg when GUC on	15;16;17;18;19	2026-09-12	drop JOINREL fold
+parallel_am_scan	parallel_am_scan	a parallel index build indexes every row of the table	15;16;17;18;19	never	-
+parallel_am_scan	parallel_am_scan	a parallel table-AM scan returns the same row count as serial	15;16;17;18;19	never	-
+parallel_am_scan	parallel_am_scan	premise: ANALYZE printed a rows= line per launched worker	15;16;17;18;19	never	-
+parallel_am_scan	parallel_am_scan	premise: EXPLAIN ANALYZE launched two workers	15;16;17;18;19	never	-
+parallel_am_scan	parallel_am_scan	premise: both sides of the comparison returned a value	15;16;17;18;19	never	-
+parallel_am_scan	parallel_am_scan	premise: the comparison reads the table through the index	15;16;17;18;19	never	-
+parallel_am_scan	parallel_am_scan	premise: the index build requested parallel workers	15;16;17;18;19	never	-
+parallel_am_scan	parallel_am_scan	premise: the parallel plan has Gather	15;16;17;18;19	never	-
+parallel_am_scan	parallel_am_scan	premise: the parallel plan is still a Seq Scan, not a custom scan	15;16;17;18;19	never	-
+parallel_am_scan	parallel_am_scan	premise: the parallel plan uses two workers	15;16;17;18;19	never	-
+parallel_am_scan	parallel_am_scan	premise: the serial plan is not a columnar custom scan	15;16;17;18;19	never	-
+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	-

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 1235
+checks_never_observed_red 1249