diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f5400c8..1adc3187 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1367,6 +1367,25 @@ Measured by the identity of the postmaster each No bash suite changes, so no ledger row moves and the census does not. `cluster_tests` 427 -> 430, re-derived by collection on the reseated tree. +- A covering projection scan could not run in parallel. + + `PgColumnarSetRelPathlist` offered the covering projection as a serial + CustomPath (`parallel_aware = false`, `parallel_safe = false`) and the + parallel base scan as a partial path with no projection name. Those cannot + both be true of one plan: either Gather wins and the projection is dropped, + or the serial projection wins and the workers are dropped. Measured on a + 32,000-row scrambled table under parallel settings: the covering query + planned as a serial `Columnar Projection` (`projection-only`), while the + same query with the projection-scan GUC off planned Gather over a parallel + base scan. + + The executor already partitions whatever storage `BeginCustomScan` opened + (the DSM stripe counter is attached to `readState`). A partial covering + path now carries the projection name, divides CPU the same way the parallel + base path does, and keeps I/O undivided. After the change the same fixture + plans `Gather` plus `Columnar Projection: byik` and still returns each + covering row once. I/O is still the base relation's pages; pricing from the + projection's own storage pages is a separate defect. - Three suites ported to pytest, and the queue re-derived (#432). @@ -1538,6 +1557,25 @@ Measured by the identity of the postmaster each surface grew with it. No check names change, so no ledger row moves and the census stays at 1391. +- A covering projection scan could not run in parallel. + + `PgColumnarSetRelPathlist` offered the covering projection as a serial + CustomPath (`parallel_aware = false`, `parallel_safe = false`) and the + parallel base scan as a partial path with no projection name. Those cannot + both be true of one plan: either Gather wins and the projection is dropped, + or the serial projection wins and the workers are dropped. Measured on a + 32,000-row scrambled table under parallel settings: the covering query + planned as a serial `Columnar Projection` (`projection-only`), while the + same query with the projection-scan GUC off planned Gather over a parallel + base scan. + + The executor already partitions whatever storage `BeginCustomScan` opened + (the DSM stripe counter is attached to `readState`). A partial covering + path now carries the projection name, divides CPU the same way the parallel + base path does, and keeps I/O undivided. After the change the same fixture + plans `Gather` plus `Columnar Projection: byik` and still returns each + covering row once. I/O is still the base relation's pages; pricing from the + projection's own storage pages is a separate defect. - The union-merge page did not say why rebasing works where merging does not (#1116 follow-up). diff --git a/src/columnar_customscan.c b/src/columnar_customscan.c index 8954cb86..1b5665ef 100644 --- a/src/columnar_customscan.c +++ b/src/columnar_customscan.c @@ -2753,6 +2753,9 @@ PgColumnarSetRelPathlist(PlannerInfo *root, RelOptInfo *rel, Index rti, CustomPath *cpath; Cost serialStartupCost; Cost serialTotalCost; + Cost projRun = 0; + double projScale = 1.0; + char *projName = NULL; Path *seqpath = NULL; List *keep = NIL; ListCell *lc; @@ -3026,17 +3029,16 @@ PgColumnarSetRelPathlist(PlannerInfo *root, RelOptInfo *rel, Index rti, */ { AttrNumber sortAttno = 0; - char *projName = pgcolumnar_choose_projection(root, rel, rte->relid, - &sortAttno); + + projName = pgcolumnar_choose_projection(root, rel, rte->relid, + &sortAttno); if (projName != NULL) { CustomPath *ppath = makeNode(CustomPath); Cost serialRun; - Cost projRun; double sel; double baseSurvival; - double scale; double groups; double floorFrac; int limit; @@ -3083,12 +3085,12 @@ PgColumnarSetRelPathlist(PlannerInfo *root, RelOptInfo *rel, Index rti, baseSurvival = pgcolumnar_zonemap_survival(rel, rte->relid); if (baseSurvival < 1e-9) baseSurvival = 1e-9; - scale = sel / baseSurvival; - if (scale > 1.0) - scale = 1.0; - if (scale < 0.0) - scale = 0.0; - projRun = serialRun * scale; + projScale = sel / baseSurvival; + if (projScale > 1.0) + projScale = 1.0; + if (projScale < 0.0) + projScale = 0.0; + projRun = serialRun * projScale; ppath->path.startup_cost = serialStartupCost; ppath->path.total_cost = serialStartupCost + projRun; ppath->path.pathkeys = NIL; @@ -3251,6 +3253,63 @@ PgColumnarSetRelPathlist(PlannerInfo *root, RelOptInfo *rel, Index rti, #endif ppath->methods = &pgcolumnar_path_methods; add_partial_path(rel, &ppath->path); + + /* + * A serial covering path cannot compete with that partial path: + * it is parallel_aware = false, so either Gather wins and the + * projection is dropped, or the serial projection wins and the + * workers are dropped. The executor already partitions whatever + * storage BeginCustomScan opened -- the DSM stripe counter is + * attached to readState, covering projection included -- so a + * covering scan can be parallel. Offer a partial path that + * carries the projection name. + * + * I/O is still the base relation's pages, scaled by the same + * factor as the serial covering path. Pricing from the + * projection's own storage pages is a separate defect. + */ + if (projName != NULL) + { + CustomPath *prpath = makeNode(CustomPath); + Cost ioRunProj; + Cost cpuRunProj; + + prpath->path.pathtype = T_CustomScan; + prpath->path.parent = rel; + prpath->path.pathtarget = rel->reltarget; + prpath->path.param_info = NULL; + prpath->path.parallel_aware = true; + prpath->path.parallel_safe = true; + prpath->path.parallel_workers = workers; + /* + * Clamp ioRunProj to projRun. With projRun = serialRun * + * projScale this is unreachable: ioRun was already clamped + * to serialRun one level up, and multiplying both sides by + * the same non-negative projScale preserves the order. It + * becomes live if projRun is ever computed independently + * (for example from the projection's own pages). When the + * clamp binds fully, cpuRunProj is zero and the partial + * covering path totals exactly like the serial covering + * path, so Gather loses. + */ + ioRunProj = ioRun * projScale; + if (ioRunProj > projRun) + ioRunProj = projRun; + cpuRunProj = projRun - ioRunProj; + prpath->path.rows = clamp_row_est(rel->rows / divisor); + prpath->path.startup_cost = serialStartupCost; + prpath->path.total_cost = serialStartupCost + + ioRunProj + cpuRunProj / divisor; + prpath->path.pathkeys = NIL; + prpath->flags = 0; + prpath->custom_paths = NIL; + prpath->custom_private = list_make1(makeString(projName)); +#if PG_VERSION_NUM >= 170000 + prpath->custom_restrictinfo = rel->baserestrictinfo; +#endif + prpath->methods = &pgcolumnar_path_methods; + add_partial_path(rel, &prpath->path); + } } } } diff --git a/test/check_ledger.tsv b/test/check_ledger.tsv index e718e104..3b645460 100644 --- a/test/check_ledger.tsv +++ b/test/check_ledger.tsv @@ -1442,6 +1442,15 @@ parallel_scan_cost parallel_scan_cost premise: the serial plan has no Gather 15; 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 - +projection_parallel projection_parallel a covering projection can be a parallel scan 15;16;17;18;19 2026-09-18 - +projection_parallel projection_parallel a parallel covering projection returns the covering rows once 15;16;17;18;19 never - +projection_parallel projection_parallel premise: ANALYZE printed a rows= line per launched worker 15;16;17;18;19 never - +projection_parallel projection_parallel premise: EXPLAIN ANALYZE launched two workers 15;16;17;18;19 never - +projection_parallel projection_parallel premise: a covering projection exists 15;16;17;18;19 never - +projection_parallel projection_parallel premise: a parallel base scan is available when the projection is off 15;16;17;18;19 never - +projection_parallel projection_parallel premise: a serial covering query uses the projection 15;16;17;18;19 never - +projection_parallel projection_parallel premise: the table holds every inserted row 15;16;17;18;19 never - +projection_parallel projection_parallel workers share the covering projection scan, it is not a single claimer 15;16;17;18;19 2026-09-18 first-wins-claimer projection_scan_cost projection_scan_cost a clause that mentions the sort key but cannot prune on it does not cheapen a covering projection 15;16;17;18;19 never - projection_scan_cost projection_scan_cost a non-sort-key restriction does not cheapen a covering projection 15;16;17;18;19 2026-09-18 - projection_scan_cost projection_scan_cost a tight covering projection is cheaper relative to the base than a loose one 15;16;17;18;19 2026-09-17 - diff --git a/test/check_ledger_budget.txt b/test/check_ledger_budget.txt index e81f0e96..844fbe07 100644 --- a/test/check_ledger_budget.txt +++ b/test/check_ledger_budget.txt @@ -206,4 +206,7 @@ suites_not_covered 249 # branch previously stated 1459 against a5c7d5d. Re-counted on THIS tree, # never by adding: # awk -F'\t' '$5=="never"' test/check_ledger.tsv | wc -l -checks_never_observed_red 1492 +# RE-DERIVED ON THE COMPOSED TREE after a rebase (#1127), by counting rather +# than by adding this branch's nine rows to either side: main moved while this +# sat, so neither the branch's previous value nor main's is the composed one. +checks_never_observed_red 1499 diff --git a/test/projection_parallel.sh b/test/projection_parallel.sh new file mode 100755 index 00000000..ee3ec6e8 --- /dev/null +++ b/test/projection_parallel.sh @@ -0,0 +1,178 @@ +#!/usr/bin/env bash +# +# pgColumnar: a covering projection scan must be able to run in parallel. +# +# PgColumnarSetRelPathlist offers a covering projection as a serial CustomPath +# (parallel_aware = false, parallel_safe = false) and a parallel base scan as a +# partial path with no projection name. Those cannot both be true of one plan: +# either Gather wins and the projection is dropped, or the serial projection +# wins and the workers are dropped. A covering query under parallel settings +# should be both. +# +# Gather in the plan is not enough. A partial path that no worker claims a +# stripe from still looks parallel, and the leader (or a single claimer) +# still returns the covering rows once. The load-bearing arms are EXPLAIN +# ANALYZE: two workers launched, and both produced rows. Same public seam +# parallel_am_scan already uses for the table-AM claimer. +# +# Independent of test/pytest/test_projection_parallel.py. Same public seam +# (EXPLAIN / EXPLAIN ANALYZE of a covering projection query, plus the +# query's count). Own table, own row count, own bounds. Neither file is +# read by the other. +# +# Usage: test/projection_parallel.sh [PG_CONFIG] +# Written fresh for pgColumnar. + +set -uo pipefail +. "$(dirname "${BASH_SOURCE[0]}")/lib.sh" +pgc_setup "${1:-/usr/lib/postgresql/18/bin/pg_config}" + +N=32000 +LO=40 +HI=8039 +WANT=$((HI - LO + 1)) +psql_run "CREATE TABLE cvppar (ik int, val int, blob text) USING pgcolumnar;" +# The covering projection is stored sorted on ik, so BETWEEN LO AND HI +# occupies consecutive groups. 181 matching rows at the 100-row floor +# is two groups; 2000 matching rows is twenty. Both geometries still +# let one worker finish the range on PG15 before the other claimed. +# 8000 matching rows (eighty groups) plus 12x md5 decode work is +# what kept both workers busy on every major this run measured. +psql_run "SELECT pgcolumnar.set_options('cvppar', stripe_row_limit => 1000, chunk_group_row_limit => 100);" +# Include blob so each claimed group has real decode work. A two-int +# projection finished so fast that one worker could claim every group +# before the other started; Gather and the count still passed. +psql_run "SELECT pgcolumnar.add_projection('cvppar', 'byik', ARRAY['ik','val','blob'], ARRAY['ik']);" +# Scrambled so the base layout cannot prune on ik; the covering projection is +# stored sorted on ik. +psql_run "INSERT INTO cvppar SELECT g, g % 17, repeat(md5(g::text), 12) FROM generate_series(1, $N) g ORDER BY md5(g::text);" +# Two workers, matching parallel_am_scan: both launched workers must produce +# rows. Four workers on this fixture can leave one idle, which would make +# "every launched worker produced rows" a test of scheduling rather than of +# the shared claim. +psql_run "ALTER TABLE cvppar SET (parallel_workers = 2);" +psql_run "ANALYZE cvppar;" + +PAR="SET parallel_setup_cost = 0; + SET parallel_tuple_cost = 0; + SET parallel_leader_participation = off; + SET min_parallel_table_scan_size = 0; + SET jit = off; + SET pgcolumnar.enable_ungrouped_vector_agg = off; + SET pgcolumnar.enable_group_vectorization = off;" + +Q="SELECT ik, val, blob FROM cvppar WHERE ik BETWEEN $LO AND $HI" + +explain_cov() { + # $1 = max_parallel_workers_per_gather + # $2 = on|off for pgcolumnar.enable_projection_scan + env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \ + -d "$PGC_DB" -Atq \ + -c "$PAR" \ + -c "SET max_parallel_workers_per_gather = $1;" \ + -c "SET pgcolumnar.enable_projection_scan = $2;" \ + -c "EXPLAIN (COSTS OFF) $Q;" \ + | grep -v '^SET$' +} + +analyze_cov() { + # $1 = max_parallel_workers_per_gather + # $2 = on|off + env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \ + -d "$PGC_DB" -Atq \ + -c "$PAR" \ + -c "SET max_parallel_workers_per_gather = $1;" \ + -c "SET pgcolumnar.enable_projection_scan = $2;" \ + -c "EXPLAIN (COSTS OFF, VERBOSE, ANALYZE, TIMING OFF, SUMMARY OFF) $Q;" \ + | grep -v '^SET$' +} + +count_cov() { + # $1 = max_parallel_workers_per_gather + # $2 = on|off + env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \ + -d "$PGC_DB" -Atq \ + -c "$PAR" \ + -c "SET max_parallel_workers_per_gather = $1;" \ + -c "SET pgcolumnar.enable_projection_scan = $2;" \ + -c "SELECT count(*) FROM cvppar WHERE ik BETWEEN $LO AND $HI;" \ + | grep -v '^SET$' +} + +shape() { + local plan="$1" + local g p + g=$(printf '%s\n' "$plan" | grep -c -i 'Gather' || true) + p=$(printf '%s\n' "$plan" | grep -c 'Columnar Projection: byik' || true) + if [ "$g" -ge 1 ] && [ "$p" -ge 1 ]; then + echo gather+projection + elif [ "$g" -ge 1 ]; then + echo gather-only + elif [ "$p" -ge 1 ]; then + echo projection-only + else + echo neither + fi +} + +# Per-worker actual rows from ANALYZE text. A worker that produced nothing +# still prints rows=0, so a missing line is not a zero -- it is no measurement. +worker_rows() { + echo "$1" | grep -oE 'Worker [0-9]+:.*rows=[0-9]+' \ + | grep -oE 'rows=[0-9]+' | grep -oE '[0-9]+' +} + +serial_plan="$(explain_cov 0 on)" +par_off_plan="$(explain_cov 2 off)" +par_on_plan="$(explain_cov 2 on)" +par_on_ana="$(analyze_cov 2 on)" +par_on_count="$(count_cov 2 on)" + +rows_list="$(worker_rows "$par_on_ana")" +n_lines="$(echo "$rows_list" | grep -c . || true)" +n_busy="$(echo "$rows_list" | awk '$1>0{n++} END{print n+0}')" + +echo "-- serial:" +printf '%s\n' "$serial_plan" +echo "-- parallel, projection off:" +printf '%s\n' "$par_off_plan" +echo "-- parallel, projection on:" +printf '%s\n' "$par_on_plan" +echo "-- parallel covering analyze:" +printf '%s\n' "$par_on_ana" +echo "-- worker rows: $(echo "$rows_list" | tr "\n" " ") busy=$n_busy lines=$n_lines" +echo "-- parallel covering count=$par_on_count want=$WANT" + +check "premise: the table holds every inserted row" \ + "$(q "SELECT count(*) FROM cvppar")" "$N" + +check "premise: a covering projection exists" \ + "$(q "SELECT count(*) FROM pgcolumnar.projection_declaration WHERE rel = 'cvppar'::regclass AND name = 'byik'")" "1" + +check "premise: a serial covering query uses the projection" \ + "$(shape "$serial_plan")" "projection-only" + +check "premise: a parallel base scan is available when the projection is off" \ + "$(shape "$par_off_plan")" "gather-only" + +# The defect: the covering projection path cannot be parallel, so the planner +# cannot keep both. got is gather-only or projection-only on the unfixed tree. +check "a covering projection can be a parallel scan" \ + "$(shape "$par_on_plan")" "gather+projection" + +check "a parallel covering projection returns the covering rows once" \ + "$par_on_count" "$WANT" + +check "premise: EXPLAIN ANALYZE launched two workers" \ + "$(echo "$par_on_ana" | grep -oE 'Workers Launched: [0-9]+' | head -1 | grep -oE '[0-9]+')" "2" + +check "premise: ANALYZE printed a rows= line per launched worker" \ + "$n_lines" "2" + +# THE DEFECT the plan-shape arms cannot see: Gather is present and the count +# is right when one backend claims every stripe. Sharing means both launched +# workers produced rows. +check "workers share the covering projection scan, it is not a single claimer" \ + "$n_busy" "2" + +pgc_summary diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md index 0f775466..9cbebd83 100644 --- a/test/pytest/TESTS.md +++ b/test/pytest/TESTS.md @@ -122,6 +122,7 @@ behaviour, the source of that number is named. - [74. test_base_scan_io.py: a base scan is not priced from sibling projection pages](#74-test_base_scan_iopy-a-base-scan-is-not-priced-from-sibling-projection-pages) - [75. test_range_pruning.py: a range prunes on overlap, containment, and under its own collation](#75-test_range_pruningpy-a-range-prunes-on-overlap-containment-and-under-its-own-collation) - [76. test_docs_upgrade_chain.py: the documented upgrade chain must be the one that ships](#76-test_docs_upgrade_chainpy-the-documented-upgrade-chain-must-be-the-one-that-ships) +- [77. test_projection_parallel.py: a covering projection can be a parallel scan](#77-test_projection_parallelpy-a-covering-projection-can-be-a-parallel-scan) ## 1. How to read a test in here @@ -5993,3 +5994,27 @@ pytest twin is refused before it can report a vacuous pass. The shell twin is six arms in `test/docs_style.sh`. It folds the file with `tr` and cuts sentences with `sed`; this half splits on a lookbehind and collects with `re`. The two halves share no code. + +## 77. test_projection_parallel.py: a covering projection can be a parallel scan + +The covering-projection path was a serial CustomPath (`parallel_aware = false`, +`parallel_safe = false`) while the parallel base scan was a partial path with +no projection name. Those cannot both be true of one plan: either Gather wins +and the projection is dropped, or the serial projection wins and the workers +are dropped. + +The executor already partitions whatever storage `BeginCustomScan` opened (the +DSM stripe counter is attached to `readState`), so a covering scan can be +parallel. A partial covering-projection path is now offered. + +This file asserts the PLANNER shape, EXPLAIN ANALYZE worker rows, and the +query's count. Gather in the plan is not enough: a single claimer still +returns the covering rows once. Public seam: `EXPLAIN` / `EXPLAIN +(ANALYZE, VERBOSE)` of a covering projection query, plus `count(*)`. The +shell twin uses `cvppar` / `byik` / 32000 rows / `ik BETWEEN 40 AND 8039`; +this file uses `pcvgath` / `onskey` / 50000 rows / `skey BETWEEN 200 AND +12299`. Assertion names match. + +| test | what it asserts | +| --- | --- | +| `test_projection_parallel` | the table and covering projection exist; a serial covering query uses the projection; a parallel base scan is available when the projection is off; a covering projection can be a parallel scan; a parallel covering projection returns the covering rows once; EXPLAIN ANALYZE launched two workers and printed a rows= line for each; both launched workers produced rows | diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt index fbbc411c..5fa2e5da 100644 --- a/test/pytest/expected_tests.txt +++ b/test/pytest/expected_tests.txt @@ -484,4 +484,9 @@ guard_tests 403 # #1189). Re-derived by collection on this tree, never by adding one: # `464 tests collected`. `guard_tests` was re-derived in the same run and # did NOT move: 393. -cluster_tests 467 +# 467 -> 468: one test in test_projection_parallel.py (#1127). RE-DERIVED ON THE +# COMPOSED TREE after a rebase, not carried across it: this branch read 464 +# against a main that said 463, and #1196 and #1202 moved main to 467 in +# between. Neither side's number is the composed one and their difference is +# not the delta. +cluster_tests 468 diff --git a/test/pytest/test_compare_to_bash.py b/test/pytest/test_compare_to_bash.py index 890dd1ed..b3131fb9 100644 --- a/test/pytest/test_compare_to_bash.py +++ b/test/pytest/test_compare_to_bash.py @@ -113,6 +113,7 @@ "parallel_am_scan", "parallel_scan_cost", "projection_drop_column", + "projection_parallel", "projection_privilege", "projection_scan_cost", "projection_update", diff --git a/test/pytest/test_projection_parallel.py b/test/pytest/test_projection_parallel.py new file mode 100644 index 00000000..5e806274 --- /dev/null +++ b/test/pytest/test_projection_parallel.py @@ -0,0 +1,208 @@ +"""A covering projection scan must be able to run in parallel. + +PgColumnarSetRelPathlist offers a covering projection as a serial CustomPath +(parallel_aware = false, parallel_safe = false) and a parallel base scan as a +partial path with no projection name. Those cannot both be true of one plan: +either Gather wins and the projection is dropped, or the serial projection +wins and the workers are dropped. + +Gather in the plan is not enough. A partial path that no worker claims a +stripe from still looks parallel, and a single claimer still returns the +covering rows once. The load-bearing arms are EXPLAIN ANALYZE: two workers +launched, and both produced rows. Same public seam parallel_am_scan already +uses for the table-AM claimer. + +This file asserts the PLANNER shape, ANALYZE worker rows, and the query's +count. Independent of test/projection_parallel.sh: same public seam, own +fixture, own observations. Assertion names match the shell suite so the +two can be compared by name, not by importing each other. +""" + + +def _nodes(plan): + stack = [plan[0]["Plan"]] + while stack: + node = stack.pop(0) + yield node + stack.extend(node.get("Plans") or ()) + + +def _custom_scan(plan): + for node in _nodes(plan): + if node.get("Node Type") == "Custom Scan": + return node + return None + + +def _gather(plan): + for node in _nodes(plan): + if node.get("Node Type") == "Gather": + return node + return None + + +def _shape(plan): + has_g = _gather(plan) is not None + node = _custom_scan(plan) + has_p = bool(node and node.get("Columnar Projection")) + if has_g and has_p: + return "gather+projection" + if has_g: + return "gather-only" + if has_p: + return "projection-only" + return "neither" + + +def _worker_rows(plan): + rows = [] + scan = _custom_scan(plan) + if scan is None: + return rows + for worker in scan.get("Workers") or (): + if "Actual Rows" in worker: + rows.append(worker["Actual Rows"]) + return rows + + +def _apply_parallel(cur): + cur.execute("SET parallel_setup_cost = 0") + cur.execute("SET parallel_tuple_cost = 0") + cur.execute("SET parallel_leader_participation = off") + cur.execute("SET min_parallel_table_scan_size = 0") + cur.execute("SET jit = off") + cur.execute("SET pgcolumnar.enable_ungrouped_vector_agg = off") + cur.execute("SET pgcolumnar.enable_group_vectorization = off") + + +def _plan(conn, sql, workers, projection_scan, analyze=False): + opts = ( + "ANALYZE, VERBOSE, TIMING OFF, SUMMARY OFF, " + if analyze + else "" + ) + with conn.cursor() as cur: + _apply_parallel(cur) + cur.execute(f"SET max_parallel_workers_per_gather = {workers}") + cur.execute( + "SET pgcolumnar.enable_projection_scan = " + + ("on" if projection_scan else "off") + ) + cur.execute(f"EXPLAIN ({opts}FORMAT JSON, COSTS OFF) " + sql) + return cur.fetchone()[0] + + +def _count(conn, sql, workers, projection_scan): + with conn.cursor() as cur: + _apply_parallel(cur) + cur.execute(f"SET max_parallel_workers_per_gather = {workers}") + cur.execute( + "SET pgcolumnar.enable_projection_scan = " + + ("on" if projection_scan else "off") + ) + cur.execute(sql) + return cur.fetchone()[0] + + +def test_projection_parallel(pgc_conn, expect): + n = 50000 + lo = 200 + hi = 12299 + want = hi - lo + 1 + with pgc_conn.cursor() as cur: + cur.execute( + "CREATE TABLE pcvgath (skey int, payload int, filler text) " + "USING pgcolumnar" + ) + # The covering projection is stored sorted on skey, so the + # BETWEEN range occupies consecutive groups. 400 matching rows + # at the 100-row floor is four groups. 12000 matching rows is + # 120 groups, plus 8x md5 decode work -- different geometry + # from the shell twin, same reason: a short range finishes + # in one worker before the other claims. + cur.execute( + "SELECT pgcolumnar.set_options('pcvgath', stripe_row_limit => 2000, " + "chunk_group_row_limit => 100)" + ) + # Include filler so each claimed group has real decode work. + # A two-int projection can finish before the second worker claims. + cur.execute( + "SELECT pgcolumnar.add_projection('pcvgath', 'onskey', " + "ARRAY['skey','payload','filler'], ARRAY['skey'])" + ) + # Scrambled insert, different N / stripe / hash salt from the shell twin. + cur.execute( + f"INSERT INTO pcvgath SELECT g, g % 23, repeat(md5((g + 9)::text), 8) " + f"FROM generate_series(1, {n}) g ORDER BY md5((g + 9)::text)" + ) + # Two workers, matching the shell twin's claimer arms and + # parallel_am_scan. Three workers on this fixture can leave one idle. + cur.execute("ALTER TABLE pcvgath SET (parallel_workers = 2)") + cur.execute("ANALYZE pcvgath") + cur.execute("SELECT count(*) FROM pcvgath") + expect.num(cur.fetchone()[0], n, "premise: the table holds every inserted row") + cur.execute( + "SELECT count(*) FROM pgcolumnar.projection_declaration " + "WHERE rel = 'pcvgath'::regclass AND name = 'onskey'" + ) + expect.num(cur.fetchone()[0], 1, "premise: a covering projection exists") + + sql = ( + f"SELECT skey, payload, filler FROM pcvgath " + f"WHERE skey BETWEEN {lo} AND {hi}" + ) + count_sql = ( + f"SELECT count(*) FROM pcvgath WHERE skey BETWEEN {lo} AND {hi}" + ) + + serial = _plan(pgc_conn, sql, 0, True) + par_off = _plan(pgc_conn, sql, 2, False) + par_on = _plan(pgc_conn, sql, 2, True) + par_on_ana = _plan(pgc_conn, sql, 2, True, analyze=True) + par_on_count = _count(pgc_conn, count_sql, 2, True) + + worker_rows = _worker_rows(par_on_ana) + n_busy = sum(1 for r in worker_rows if r and r > 0) + launched = (_gather(par_on_ana) or {}).get("Workers Launched") + + print("-- serial:", _shape(serial)) + print("-- parallel, projection off:", _shape(par_off)) + print("-- parallel, projection on:", _shape(par_on)) + print(f"-- worker rows {worker_rows} busy={n_busy} launched={launched}") + print(f"-- parallel covering count={par_on_count} want={want}") + + expect.text( + _shape(serial), + "projection-only", + "premise: a serial covering query uses the projection", + ) + expect.text( + _shape(par_off), + "gather-only", + "premise: a parallel base scan is available when the projection is off", + ) + expect.text( + _shape(par_on), + "gather+projection", + "a covering projection can be a parallel scan", + ) + expect.num( + par_on_count, + want, + "a parallel covering projection returns the covering rows once", + ) + expect.num( + launched, + 2, + "premise: EXPLAIN ANALYZE launched two workers", + ) + expect.num( + len(worker_rows), + 2, + "premise: ANALYZE printed a rows= line per launched worker", + ) + expect.num( + n_busy, + 2, + "workers share the covering projection scan, it is not a single claimer", + ) diff --git a/test/run_all_versions.sh b/test/run_all_versions.sh index bbd8742e..00363568 100755 --- a/test/run_all_versions.sh +++ b/test/run_all_versions.sh @@ -254,6 +254,7 @@ SUITES=( planner_choice_quality preimage_rewrite projection_drop_column + projection_parallel projection_privilege projection_rename_restore projection_rewrite