bench: a bulk Citus arm, so the bulk row compares like with like (#445)

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 <workers>
+#
+# 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 <current> <workers>
+#
+# 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 <current> <workers>
+#
+# 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 <got> <want>
 #
 # A load that lost rows is a failure and not a fast result. An errored parallel

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

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.