test: a statement does not see its own writes after a mid-statement flush (#1341)

test/check_ledger.tsv

@@ -2718,6 +2718,15 @@ settings_reference	settings_reference	each documented default is the server's	15
 settings_reference	settings_reference	every documented setting exists on the server	15;16;17;18;19	2026-10-15	the compression_level row renamed compression_levl in docs/configuration.md
 settings_reference	settings_reference	every setting the server defines is documented, the developer hooks aside	15;16;17;18;19	2026-10-15	the backticks around every setting name removed from docs/configuration.md;the compression_level row deleted from docs/configuration.md;the compression_level row renamed compression_levl in docs/configuration.md
 settings_reference	settings_reference	the reference documents at least thirty settings, so the comparisons below read something	15;16;17;18;19	2026-10-15	the backticks around every setting name removed from docs/configuration.md
+statement_self_visibility	statement_self_visibility	INSERT INTO t SELECT FROM t runs to completion	15;16;17;18;19	2026-10-15	a serial scan re-reads its row-group list when it reaches the end, so it picks up groups its own statement flushed (src/columnar_reader.c pgcolumnar_next_group_index)
+statement_self_visibility	statement_self_visibility	UPDATE of every row runs to completion	15;16;17;18;19	2026-10-15	a serial scan re-reads its row-group list when it reaches the end, so it picks up groups its own statement flushed (src/columnar_reader.c pgcolumnar_next_group_index)
+statement_self_visibility	statement_self_visibility	an INSERT in a WITH clause reading its own table returns the rows it inserted, as heap does	15;16;17;18;19	2026-10-15	a serial scan re-reads its row-group list when it reaches the end, so it picks up groups its own statement flushed (src/columnar_reader.c pgcolumnar_next_group_index)
+statement_self_visibility	statement_self_visibility	and each statement saw the earlier ones but not itself, as heap does	15;16;17;18;19	2026-10-15	a serial scan re-reads its row-group list when it reaches the end, so it picks up groups its own statement flushed (src/columnar_reader.c pgcolumnar_next_group_index)
+statement_self_visibility	statement_self_visibility	and it changed each row exactly once, as heap does	15;16;17;18;19	2026-10-15	a serial scan re-reads its row-group list when it reaches the end, so it picks up groups its own statement flushed (src/columnar_reader.c pgcolumnar_next_group_index)
+statement_self_visibility	statement_self_visibility	and it read only the rows that were there when it started, as heap does	15;16;17;18;19	2026-10-15	a serial scan re-reads its row-group list when it reaches the end, so it picks up groups its own statement flushed (src/columnar_reader.c pgcolumnar_next_group_index)
+statement_self_visibility	statement_self_visibility	and the table holds them once, as heap does	15;16;17;18;19	2026-10-15	a serial scan re-reads its row-group list when it reaches the end, so it picks up groups its own statement flushed (src/columnar_reader.c pgcolumnar_next_group_index)
+statement_self_visibility	statement_self_visibility	in one transaction, an INSERT, an UPDATE of every row and an INSERT SELECT run to completion	15;16;17;18;19	2026-10-15	a serial scan re-reads its row-group list when it reaches the end, so it picks up groups its own statement flushed (src/columnar_reader.c pgcolumnar_next_group_index)
+statement_self_visibility	statement_self_visibility	the INSERT flushed row groups while its scan was still reading	15;16;17;18;19	2026-10-15	a serial scan re-reads its row-group list when it reaches the end, so it picks up groups its own statement flushed (src/columnar_reader.c pgcolumnar_next_group_index)
 storage_row_repair	storage_row_repair	REFRESH MATERIALIZED VIEW repairs the view's row	15;16;17;18;19	2026-10-12	PgColumnarRetargetStorageRelation returns at once, so no repair repoints a storage row
 storage_row_repair	storage_row_repair	a same-type ALTER COLUMN ... USING repairs the table's row	15;16;17;18;19	2026-10-12	PgColumnarRetargetStorageRelation returns at once, so no repair repoints a storage row
 storage_row_repair	storage_row_repair	after both repairs the notes' query finds nothing	15;16;17;18;19	2026-10-12	PgColumnarRetargetStorageRelation returns at once, so no repair repoints a storage row;the release notes' query is edited from NOT EXISTS to EXISTS

test/check_ledger_budget.txt

@@ -2400,4 +2400,18 @@ suites_not_covered 231
 #     never         2282     = 2282
 #     dated          725 + 1 =  726
 #     2282 + 726 == 3008
+#
+# #1341, a statement does not see its own writes after a mid-statement flush
+# (jdatcmd): 2282 -> 2282. statement_self_visibility enters with all 9 checks from
+# five-major runs (9/9 on each of 15-19). Every shape starts from fresh tables, so
+# each was observed alone (OffgridwithJD, review of #1416), and all 9 are dated on
+# PG17 by one mutant: a serial scan that re-reads its row-group list when it
+# reaches the end. Each statement then fed on the groups it had flushed until its
+# statement_timeout, or, in the CTE, until id overflowed, and each comparison with
+# heap differed.
+#
+#     ledger rows   3008 + 9 = 3017
+#     never         2282 + 0 = 2282
+#     dated          726 + 9 = 735
+#     2282 + 735 == 3017
 checks_never_observed_red 2282

test/pytest/TESTS.md

@@ -164,6 +164,7 @@ behaviour, the source of that number is named.
 - [116. test_settings_reference.py: the settings reference matches the server](#116-test_settings_referencepy-the-settings-reference-matches-the-server)
 - [117. test_large_values.py: megabyte-scale values are stored exactly](#117-test_large_valuespy-megabyte-scale-values-are-stored-exactly)
 - [118. test_recovery_minimal_wal.py: tables that skip WAL survive a crash](#118-test_recovery_minimal_walpy-tables-that-skip-wal-survive-a-crash)
+- [119. test_statement_self_visibility.py: a statement does not see its own writes](#119-test_statement_self_visibilitypy-a-statement-does-not-see-its-own-writes)
 
 ## 1. How to read a test in here
 
@@ -7007,3 +7008,13 @@ The pytest twin of `test/recovery_minimal_wal.sh` (#1341), every check by the sh
 | test | what it holds |
 | --- | --- |
 | `test_tables_skipping_wal_survive_a_crash` | the premise, and create-and-load and truncate-and-reload at both thresholds, each across a crash |
+
+## 119. test_statement_self_visibility.py: a statement does not see its own writes
+
+The pytest twin of `test/statement_self_visibility.sh` (#1341), every check by the shell's name. A scan reads its row-group list once, when it starts, so the groups its own statement flushes later are not in it. With row groups of 1000 rows, `INSERT INTO t SELECT FROM t`, an `UPDATE` of every row, both in one transaction after an earlier write, and an `INSERT` in a `WITH` clause each run on a columnar table and on heap, and heap is the answer. An arm checks that the `INSERT` really flushed row groups while its scan was reading. Every statement runs under a `statement_timeout`, because a scan that fed on its own output would never end. Both tables run every statement, so a columnar failure cannot make the comparison vacuous.
+
+### Every test
+
+| test | what it holds |
+| --- | --- |
+| `test_a_statement_does_not_see_its_own_writes` | the four shapes run to completion and answer as heap does, and the mid-statement flush |

test/pytest/expected_tests.txt

@@ -687,4 +687,6 @@ guard_tests 435
 # over the cluster files on this tree.
 # cluster_tests 576 -> 577 for #1341's test_row_tid_refuses_a_row_number_below_one in
 # test_rollup.py, by collection over the cluster files on this tree.
-cluster_tests 577
+# cluster_tests 577 -> 578 for #1341's test_statement_self_visibility.py, one test, by collection
+# over the cluster files on this tree.
+cluster_tests 578

test/pytest/test_compare_to_bash.py

@@ -145,6 +145,7 @@
             "scan_decode_cost",
             "settings_reference",
             "sorted_pathkeys",
+            "statement_self_visibility",
             "stats_privilege",
             "storage_row_repair",
             "temporal",

test/pytest/test_statement_self_visibility.py

@@ -0,0 +1,106 @@
+"""A statement does not see the rows it writes, even after it has flushed some of
+them into row groups part-way through (#1341). The pytest twin of
+`test/statement_self_visibility.sh`, every check by the shell's name.
+
+A scan reads its row-group list once, when it starts, so groups its own statement
+flushes later are not in it. A reader that picked them up would make
+INSERT INTO t SELECT FROM t feed on its own output, and UPDATE t reach a row twice.
+
+Each shape runs on a columnar table and on heap, and heap is the answer. The
+columnar table writes row groups of 1000 rows, so every statement crosses several
+flushes. Every statement runs under a statement_timeout, because a scan that fed on
+its own output would otherwise never end. Both tables run every statement, so a
+columnar failure cannot leave heap untouched and the comparison vacuous.
+"""
+import psycopg
+
+AMS = ("pgcolumnar", "heap")
+
+
+def _session(cur):
+    cur.execute("SET pgcolumnar.stripe_row_limit = 1000")
+    cur.execute("SET statement_timeout = '60s'")
+
+
+def _run(conn, sql):
+    """Run `sql` (with %t for the table) on both tables -> 'ok' or the errors."""
+    errs = []
+    for am in AMS:
+        try:
+            with conn.cursor() as cur:
+                _session(cur)
+                cur.execute(sql.replace("%t", f"sv_{am}"))
+        except psycopg.Error as exc:
+            errs.append(f"{am}: {str(exc).splitlines()[0]}")
+            conn.execute("ROLLBACK")  # a failed BEGIN ... COMMIT leaves the session in it
+    return " ".join(errs) or "ok"
+
+
+def _answer(conn, sql, am):
+    """The row the query returns, or its error's first line if it failed."""
+    try:
+        with conn.cursor() as cur:
+            _session(cur)
+            cur.execute(sql.replace("%t", f"sv_{am}"))
+            return cur.fetchone()
+    except psycopg.Error as exc:
+        return f"ERROR: {str(exc).splitlines()[0]}"
+
+
+def _same(conn, sql):
+    c, h = (_answer(conn, sql, am) for am in AMS)
+    return "same" if c == h else f"columnar={c} heap={h}"
+
+
+def _groups(conn):
+    with conn.cursor() as cur:
+        cur.execute("SELECT count(*) FROM pgcolumnar.row_group "
+                    "WHERE storage_id = pgcolumnar.get_storage_id('sv_pgcolumnar')")
+        return cur.fetchone()[0]
+
+
+def _fresh(conn):
+    """Both tables made again, holding rows 1..5000. Every shape starts here, so a
+    shape that goes wrong cannot fail the ones after it."""
+    with conn.cursor() as cur:
+        _session(cur)
+        for am in AMS:
+            cur.execute(f"DROP TABLE IF EXISTS sv_{am}")
+            cur.execute(f"CREATE TABLE sv_{am} (id int, v int) USING {am}")
+            cur.execute(f"INSERT INTO sv_{am} SELECT g, 0 FROM generate_series(1, 5000) g")
+
+
+def test_a_statement_does_not_see_its_own_writes(pgc_conn, expect):
+    _fresh(pgc_conn)
+    before = _groups(pgc_conn)
+
+    expect.text(_run(pgc_conn, "INSERT INTO %t SELECT id + 5000, v FROM %t"), "ok",
+                "INSERT INTO t SELECT FROM t runs to completion")
+    after = _groups(pgc_conn)
+    expect.text("yes" if after - before >= 2 else f"no ({before} groups before, {after} after)", "yes",
+                "the INSERT flushed row groups while its scan was still reading")
+    expect.text(_same(pgc_conn, "SELECT count(*), count(DISTINCT id), max(id) FROM %t"), "same",
+                "and it read only the rows that were there when it started, as heap does")
+
+    _fresh(pgc_conn)
+    expect.text(_run(pgc_conn, "UPDATE %t SET v = v + 1"), "ok",
+                "UPDATE of every row runs to completion")
+    expect.text(_same(pgc_conn, "SELECT count(*), min(v), max(v) FROM %t"), "same",
+                "and it changed each row exactly once, as heap does")
+
+    _fresh(pgc_conn)
+    expect.text(_run(pgc_conn, "BEGIN; INSERT INTO %t SELECT g, 0 FROM generate_series(20001, 23000) g; "
+                               "UPDATE %t SET v = v + 1; INSERT INTO %t SELECT id + 100000, v FROM %t; COMMIT"),
+                "ok",
+                "in one transaction, an INSERT, an UPDATE of every row and an INSERT SELECT run to completion")
+    expect.text(_same(pgc_conn, "SELECT count(*), count(DISTINCT id), min(v), max(v), sum(v) FROM %t"), "same",
+                "and each statement saw the earlier ones but not itself, as heap does")
+
+    # No WHERE on the scan: every row the INSERT writes would pass it, so a scan
+    # that fed on its own output would meet those rows.
+    _fresh(pgc_conn)
+    expect.text(_same(pgc_conn, "WITH w AS (INSERT INTO %t SELECT id + 1000000, v FROM %t "
+                                "RETURNING 1) SELECT count(*) FROM w"), "same",
+                "an INSERT in a WITH clause reading its own table returns the rows it inserted, as heap does")
+    expect.text(_same(pgc_conn, "SELECT count(*) FROM %t"), "same",
+                "and the table holds them once, as heap does")

test/run_all_versions.sh

@@ -302,6 +302,7 @@ SUITES=(
 	sorted_mark_rename
 	sorted_pathkeys
 	sorted_projection
+	statement_self_visibility
 	stats_privilege
 	storage_row_repair
 	tablesample

test/statement_self_visibility.sh

@@ -0,0 +1,90 @@
+#!/usr/bin/env bash
+#
+# A statement does not see the rows it writes, even after it has flushed some of
+# them into row groups part-way through (#1341). A scan reads its row-group list
+# once, when it starts; groups the same statement flushes later are not in it. A
+# reader that picked them up would make INSERT INTO t SELECT FROM t feed on its own
+# output and UPDATE t reach a row twice -- the Halloween problem.
+#
+# Each shape runs on a columnar table and on heap, and heap is the answer. The
+# columnar table writes row groups of 1000 rows, so every statement below crosses
+# several flushes. Every statement runs under a statement_timeout: a scan that fed
+# on its own output would otherwise never end.
+#
+# Usage:  test/statement_self_visibility.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}"
+
+SET="SET pgcolumnar.stripe_row_limit = 1000; SET statement_timeout = '60s';"
+groups() {	# groups TABLE -> how many row groups the table has
+	q "SELECT count(*) FROM pgcolumnar.row_group WHERE storage_id = pgcolumnar.get_storage_id('$1')"
+}
+run() {	# run SQL: on both tables, with %t standing for the table; -> ok or the errors
+	# Both tables, always: stopping at a columnar error would leave heap unchanged
+	# too, and the "as heap does" arms after it would compare two untouched tables.
+	local am out t errs=""
+	for am in pgcolumnar heap; do
+		t="sv_$am"
+		out="$(psql_run "$SET ${1//%t/$t}" 2>&1)" || errs="$errs$am: $(grep -m1 ERROR <<<"$out") "
+	done
+	echo "${errs:-ok}"
+}
+answer() {	# answer SQL -> the last line psql prints, or its ERROR line if the query failed
+	local out
+	out="$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" \
+		-v ON_ERROR_STOP=1 -qAt -c "$1" 2>&1)" || { grep -m1 ERROR <<<"$out"; return; }
+	tail -1 <<<"$out"
+}
+same() {	# same SQL -> "same" when both tables answer alike, else both answers
+	local c h
+	c="$(answer "$SET ${1//%t/sv_pgcolumnar}")"; h="$(answer "$SET ${1//%t/sv_heap}")"
+	[ "$c" = "$h" ] && echo same || echo "columnar=$c heap=$h"
+}
+fresh() {	# both tables made again, holding rows 1..5000. Every shape starts here,
+	# so a shape that goes wrong cannot fail the ones after it (OffgridwithJD,
+	# review of #1416).
+	local am
+	for am in pgcolumnar heap; do
+		psql_run "$SET DROP TABLE IF EXISTS sv_$am; CREATE TABLE sv_$am (id int, v int) USING $am;
+		          INSERT INTO sv_$am SELECT g, 0 FROM generate_series(1, 5000) g;"
+	done
+}
+
+fresh
+before="$(groups sv_pgcolumnar)"
+
+# ---- INSERT INTO t SELECT FROM t ------------------------------------------------
+check "INSERT INTO t SELECT FROM t runs to completion" "$(run "INSERT INTO %t SELECT id + 5000, v FROM %t;")" "ok"
+after="$(groups sv_pgcolumnar)"
+check "the INSERT flushed row groups while its scan was still reading" \
+	"$([ "$((after - before))" -ge 2 ] && echo yes || echo "no ($before groups before, $after after)")" "yes"
+check "and it read only the rows that were there when it started, as heap does" \
+	"$(same "SELECT count(*) || '|' || count(DISTINCT id) || '|' || max(id) FROM %t")" "same"
+
+# ---- UPDATE of every row --------------------------------------------------------
+fresh
+check "UPDATE of every row runs to completion" "$(run "UPDATE %t SET v = v + 1;")" "ok"
+check "and it changed each row exactly once, as heap does" \
+	"$(same "SELECT count(*) || '|' || min(v) || '|' || max(v) FROM %t")" "same"
+
+# ---- both again, in one transaction after an earlier write ----------------------
+fresh
+check "in one transaction, an INSERT, an UPDATE of every row and an INSERT SELECT run to completion" \
+	"$(run "BEGIN; INSERT INTO %t SELECT g, 0 FROM generate_series(20001, 23000) g;
+	        UPDATE %t SET v = v + 1; INSERT INTO %t SELECT id + 100000, v FROM %t; COMMIT;")" "ok"
+check "and each statement saw the earlier ones but not itself, as heap does" \
+	"$(same "SELECT count(*) || '|' || count(DISTINCT id) || '|' || min(v) || '|' || max(v) || '|' || sum(v) FROM %t")" "same"
+
+# ---- a data-modifying CTE --------------------------------------------------------
+# No WHERE on the scan: every row the INSERT writes would pass it, so a scan that fed
+# on its own output would meet those rows (OffgridwithJD, review of #1416).
+fresh
+check "an INSERT in a WITH clause reading its own table returns the rows it inserted, as heap does" \
+	"$(same "WITH w AS (INSERT INTO %t SELECT id + 1000000, v FROM %t RETURNING 1)
+	         SELECT count(*) FROM w")" "same"
+check "and the table holds them once, as heap does" "$(same "SELECT count(*) FROM %t")" "same"
+
+pgc_summary