From 4c65cd7f16424e137e2e11324fd59d05473dcab5 Mon Sep 17 00:00:00 2001 From: ChronicallyJD Date: Sun, 9 Aug 2026 07:55:47 -0600 Subject: [PATCH] bench: a bulk Citus arm, so the bulk row compares like with like (#445) The ClickBench section compared pgcolumnar's 16-worker parallel_copy against Citus's SINGLE-connection COPY, because the harness had no bulk Citus arm. That is our best path against their non-best one, on the exact question #445 was opened about. Citus columnar does accept concurrent writers into one table. Measured on the bench, 4M rows, own cluster, three runs: citus, serial COPY 3.06 3.07 3.06 citus, 8 concurrent COPY 0.509 0.518 0.513 scales 5.9x pgcolumnar, serial COPY 4.80 4.80 4.79 pgcolumnar, parallel_copy 8 1.137 1.138 1.135 scales 4.2x All four arms loaded 4,000,000 rows. So the premise that serial COPY is the only path Citus has does not hold, and the published ratio compares unlike things. ## The arm N concurrent connections, each fed a byte range from pgcolumnar.file_split_offsets -- OUR splitter, the same newline-aligned boundaries parallel_copy gives its own loaders -- through COPY ... FROM PROGRAM. So the two bulk arms differ in the engine and not in how the file was divided, and nothing copies a 15 GB file N times. It carries the same row assertion as the pgcolumnar bulk arm, for the same reason: an arm that errored leaves an empty table and returns fast, which reads as a win. That is #465's failure with a different cause. ## The preflight that would have caught my own mistake max_worker_processes is now preflighted beside max_prepared_transactions. parallel_copy registers one background worker per loader plus a coordinator, and the logical replication launcher already holds a slot, so an N-worker arm needs N + 2. The stock default is 8, so an 8-worker arm fails at "could not register pgcolumnar parallel_copy loader 7 of 8" and leaves an EMPTY table. N + 2 is measured, not reasoned. Sweeping the setting against three worker counts, the smallest value that loaded every row was: workers 2 -> 4 workers 4 -> 6 workers 8 -> 10 and one below each failed on the LAST loader with the table left empty. I hit exactly this while measuring the Citus arms, and only caught it because the row count was asserted; the arm returned quickly with nothing in it. ## Tests bench_guards.sh covers the new guard the way it covers max_prepared_transactions, including the two thresholds either side of the measured minimum, a serial arm needing none, and non-numeric input being refused rather than compared. 17 checks to 28. The arm's mechanism was verified end to end against real citus_columnar on a 200,000-row fixture before shipping: 5 offsets for 4 workers, every row present, sum(a) exact, zero duplicate keys, and the per-worker COPY counts summing to the file. Cardinality alone would not have caught a boundary that loses one row and gains another. The ClickBench numbers themselves are unchanged here. #526 now says the bulk comparison is unmeasured, which is what the evidence supports; the number comes from a run of this arm. Refs #445, #465, #526. --- bench/cb_guards.sh | 48 ++++++++++++++++++++++++++ bench/run_clickbench.sh | 76 ++++++++++++++++++++++++++++++++++++++++- test/bench_guards.sh | 44 ++++++++++++++++++++++++ 3 files changed, 167 insertions(+), 1 deletion(-) diff --git a/bench/cb_guards.sh b/bench/cb_guards.sh index 5c718a00..de829de1 100644 --- a/bench/cb_guards.sh +++ b/bench/cb_guards.sh @@ -41,6 +41,54 @@ cb_prepared_xacts_message() { "Set max_prepared_transactions = ${workers} (or more) and restart the postmaster; it cannot be changed in a session." } +# cb_worker_slots_needed +# +# An N-worker parallel_copy needs N + 2 worker slots: one background worker per +# loader, one coordinator, and one already held by the logical replication +# launcher. A serial arm registers nothing and needs none. +# +# N + 2 is measured rather than reasoned. Sweeping max_worker_processes against +# three worker counts, the smallest value that loaded every row was 4 for 2 +# workers, 6 for 4, and 10 for 8. One below each failed on the LAST loader and +# left the table empty. +cb_worker_slots_needed() { + local workers="${1:-}" + case "$workers" in '' | *[!0-9]*) printf '0\n'; return 1 ;; esac + if [ "$workers" -eq 0 ]; then printf '0\n'; else printf '%s\n' "$((workers + 2))"; fi +} + +# cb_worker_slots_ok +# +# The same shape as cb_prepared_xacts_ok and for the same reason: raising +# max_worker_processes costs a postmaster restart, so it must be asked before any +# arm is loaded rather than discovered in a load log. +# +# The stock default is 8. An 8-worker arm therefore fails on the stock setting, +# at "could not register pgcolumnar parallel_copy loader 7 of 8", and leaves an +# EMPTY table -- which returns fast and reads as excellent scaling. That is the +# #465 failure with a different cause. +cb_worker_slots_ok() { + local current="${1:-}" workers="${2:-}" need + case "$current" in '' | *[!0-9]*) return 1 ;; esac + case "$workers" in '' | *[!0-9]*) return 1 ;; esac + need="$(cb_worker_slots_needed "$workers")" || return 1 + [ "$current" -ge "$need" ] +} + +# cb_worker_slots_message +# +# Names the value to set, not merely the worker count: N + 2 is not a number the +# operator can be expected to derive from a per-loader error message. +cb_worker_slots_message() { + local current="${1:-}" workers="${2:-}" need + need="$(cb_worker_slots_needed "$workers")" + printf '%s\n' \ + "max_worker_processes is ${current:-unset}, and a ${workers}-worker parallel arm needs ${need}." \ + "parallel_copy registers one worker per loader plus a coordinator, and the logical replication launcher holds one slot." \ + "Set max_worker_processes = ${need} (or more) and restart the postmaster; it cannot be changed in a session." \ + "Below that the arm fails on its last loader and leaves an empty table, which reads as a very fast load." +} + # cb_rows_ok # # A load that lost rows is a failure and not a fast result. An errored parallel diff --git a/bench/run_clickbench.sh b/bench/run_clickbench.sh index ce1fba43..f195da36 100755 --- a/bench/run_clickbench.sh +++ b/bench/run_clickbench.sh @@ -423,7 +423,17 @@ if [ "$CB_PCOPY_WORKERS" -gt 0 ]; then cb_prepared_xacts_message "$CB_PREPARED" "$CB_PCOPY_WORKERS" >&2 die "the parallel arm cannot run; set PGC_CB_PCOPY_WORKERS=0 to skip it" fi - note " parallel arm: $CB_PCOPY_WORKERS workers, max_prepared_transactions=$CB_PREPARED" + # The other setting that costs a restart, and the one that actually bit: + # parallel_copy registers a background worker per loader plus a coordinator, + # and the logical replication launcher already holds a slot, so an N-worker + # arm needs N + 2. The stock default is 8, so an 8-worker arm fails on it at + # "could not register ... loader 7 of 8" and leaves an EMPTY table. + CB_WSLOTS=$($PSQL -At -c "SHOW max_worker_processes") + if ! cb_worker_slots_ok "$CB_WSLOTS" "$CB_PCOPY_WORKERS"; then + cb_worker_slots_message "$CB_WSLOTS" "$CB_PCOPY_WORKERS" >&2 + die "the parallel arm cannot run; set PGC_CB_PCOPY_WORKERS=0 to skip it" + fi + note " parallel arm: $CB_PCOPY_WORKERS workers, max_prepared_transactions=$CB_PREPARED, max_worker_processes=$CB_WSLOTS" fi # --------------------------------------------------------------------------- @@ -571,6 +581,63 @@ if [ "$CB_PCOPY_WORKERS" -gt 0 ] && [ -n "${ROWS[hits_col]:-}" ]; then fi fi +# ---- the Citus bulk arm, so the bulk row compares like with like ----------- +# +# Citus columnar accepts concurrent writers into one table. Measured on this box: +# 8 concurrent COPY loaded 4M rows 5.9x faster than one, with every row present. +# So publishing our 16-worker parallel_copy against their single COPY compares +# our best path with their non-best one, which is not a comparison worth making +# and is the specific claim #445 was opened about. +# +# The split is OURS -- pgcolumnar.file_split_offsets, the same newline-aligned +# byte boundaries parallel_copy gives its own loaders -- so the two bulk arms +# differ in the engine and not in how the file was divided. FROM PROGRAM feeds +# each connection its range without copying a 15 GB file N times. +if [ "$CB_PCOPY_WORKERS" -gt 0 ] && [ -n "${ROWS[hits_citus]:-}" ]; then + cz_tbl=hits_citus_pcopy + $PSQL -c "DROP TABLE IF EXISTS $cz_tbl;" >/dev/null 2>&1 + ddl_for "$cz_tbl" "USING columnar" | $PSQL -v ON_ERROR_STOP=1 >/dev/null 2>"$CB_DATA/ddl.czpcopy.err" + if [ -s "$CB_DATA/ddl.czpcopy.err" ]; then + head -5 "$CB_DATA/ddl.czpcopy.err"; die "citus bulk arm DDL failed" + fi + read -r -a cz_off <<<"$($PSQL -At -c \ + "SELECT array_to_string(pgcolumnar.file_split_offsets('$TSV', $CB_PCOPY_WORKERS), ' ')")" + if [ "${#cz_off[@]}" -ne "$((CB_PCOPY_WORKERS + 1))" ]; then + die "file_split_offsets returned ${#cz_off[@]} offsets for $CB_PCOPY_WORKERS workers" + fi + t0=$(date +%s.%N) + cz_rc=0 + for ((i = 0; i < CB_PCOPY_WORKERS; i++)); do + start=${cz_off[$i]} + len=$(( ${cz_off[$((i + 1))]} - start )) + [ "$len" -gt 0 ] || continue + # tail -c is 1-based from the start of the file. + $PSQL -v ON_ERROR_STOP=1 -c \ + "\\copy $cz_tbl FROM PROGRAM 'tail -c +$((start + 1)) $TSV | head -c $len'" \ + >> "$CB_DATA/load.czpcopy.log" 2>&1 || cz_rc=1 & + done + wait + [ "$cz_rc" = 0 ] || { tail -10 "$CB_DATA/load.czpcopy.log"; die "the citus bulk arm failed; it is not reported as a fast load"; } + $PSQL -c "VACUUM ANALYZE $cz_tbl;" >/dev/null 2>&1 + t1=$(date +%s.%N) + LOAD_S[citus_pcopy]=$(awk -v a="$t0" -v b="$t1" 'BEGIN { printf "%.1f", b - a }') + ROWS[citus_pcopy]=$($PSQL -At -c "SELECT count(*) FROM $cz_tbl") + SIZE_B[citus_pcopy]=$($PSQL -At -c "SELECT pg_total_relation_size('$cz_tbl')") + note " citus_pcopy (${CB_PCOPY_WORKERS}w): ${LOAD_S[citus_pcopy]}s, ${ROWS[citus_pcopy]} rows, ${SIZE_B[citus_pcopy]} bytes" + + # Same assertion as the pgcolumnar bulk arm, for the same reason: an arm that + # errored leaves an empty table and returns fast, which reads as a win. + if cb_rows_ok "${ROWS[citus_pcopy]}" "$TSV_ROWS"; then + printf 'ok premise: %s\n' "the citus bulk arm loaded every row of the file" + else + printf 'FAIL premise: %s (got [%s] want [%s])\n' \ + "the citus bulk arm loaded every row of the file" \ + "${ROWS[citus_pcopy]}" "$TSV_ROWS" >&2 + fail=1 + fi +fi + + # Every arm must hold the same rows as the file. A load that silently dropped # rows makes every query below faster and wrong. for arm in "${ARMS[@]}"; do @@ -765,6 +832,13 @@ done if [ -n "${LOAD_S[columnar_pcopy]:-}" ]; then printf '%-16s %12s %16s %10s\n' "columnar_pcopy" \ "${LOAD_S[columnar_pcopy]}" "${SIZE_B[columnar_pcopy]}" "-" +fi +# Printed beside columnar_pcopy, because the bulk row is only worth reading as a +# pair. A bulk number for one engine against a serial number for the other is the +# comparison this arm exists to stop. +if [ -n "${LOAD_S[citus_pcopy]:-}" ]; then + printf '%-16s %12s %16s %10s\n' "citus_pcopy" \ + "${LOAD_S[citus_pcopy]}" "${SIZE_B[citus_pcopy]}" "-" printf ' parallel_copy with %s workers, beside the serial number above; both are the point\n' \ "$CB_PCOPY_WORKERS" fi diff --git a/test/bench_guards.sh b/test/bench_guards.sh index 8357a5a9..efd591fa 100755 --- a/test/bench_guards.sh +++ b/test/bench_guards.sh @@ -69,6 +69,50 @@ check "one short is refused, which is the off-by-one that would run 7 workers" \ check "a serial arm needs no prepared transactions" \ "$(cb_prepared_xacts_ok 0 0 && echo ok || echo refused)" "ok" +# ---- max_worker_processes, the other setting that costs a restart ----------- +# +# parallel_copy registers one background worker per loader, plus a coordinator, +# and the logical replication launcher already holds one slot. So an N-worker arm +# needs N + 2, not N. The stock default is 8, which is why an 8-worker arm fails +# at "could not register pgcolumnar parallel_copy loader 7 of 8" and leaves an +# EMPTY table -- a fast, wrong, publishable-looking result. +# +# N + 2 is measured, not reasoned. Sweeping max_worker_processes against three +# worker counts on the bench, the smallest value that loaded every row was: +# +# workers 2 -> 4 workers 4 -> 6 workers 8 -> 10 +# +# and one below each failed on the LAST loader with the table left empty. +check "the stock 8 is refused for an 8-worker arm, which is the case that bit us" \ + "$(cb_worker_slots_ok 8 8 && echo ok || echo refused)" "refused" +check "N + 1 is still refused: the coordinator needs a slot too" \ + "$(cb_worker_slots_ok 9 8 && echo ok || echo refused)" "refused" +check "N + 2 is accepted, the measured minimum" \ + "$(cb_worker_slots_ok 10 8 && echo ok || echo refused)" "ok" +check "more than enough is accepted" \ + "$(cb_worker_slots_ok 32 8 && echo ok || echo refused)" "ok" +check "the rule holds at another worker count (4 needs 6)" \ + "$(cb_worker_slots_ok 6 4 && echo ok || echo refused)" "ok" +check "and one below it does not" \ + "$(cb_worker_slots_ok 5 4 && echo ok || echo refused)" "refused" + +# A serial arm registers no workers at all. +check "a serial arm needs no worker slots" \ + "$(cb_worker_slots_ok 0 0 && echo ok || echo refused)" "ok" + +# Non-numeric input must be refused rather than compared. A psql that failed +# yields an empty string, and "" -ge "" is not a comparison anyone wants. +check "a missing current value is refused, not compared" \ + "$(cb_worker_slots_ok "" 8 && echo ok || echo refused)" "refused" + +wmsg="$(cb_worker_slots_message 8 8)" +check "the worker-slot message names the setting" \ + "$([ "$(grep -c 'max_worker_processes' <<<"$wmsg")" -ge 1 ] && echo yes || echo no)" "yes" +check "and the value it must reach, not merely the worker count" \ + "$([ "$(grep -c '10' <<<"$wmsg")" -ge 1 ] && echo yes || echo no)" "yes" +check "and says it needs a restart" \ + "$([ "$(grep -ci 'restart' <<<"$wmsg")" -ge 1 ] && echo yes || echo no)" "yes" + # The message is the deliverable here: the operator has to know WHAT to set and # that it costs a restart. A bare "failed" sends them to the load log, which # reports a per-worker error and not the cause.