test: rewrite_group_scan gains its pytest twin, and a rewrite that does nothing no longer passes it (#1341)

test/check_ledger.tsv

@@ -4372,6 +4372,13 @@ replication	replication	vacuum rewrote r into a new relfilenode on the primary	1
 replication	replication	with replay paused the standby does NOT see the new rows	15;16;17;18;19	never	-
 replication	replication	with the module back, the replayed table reads correctly	15;16;17;18;19	2026-10-16	columnar_storage.c: PgColumnarWriteLogicalData logs no data or gap page (if (false) for both RelationNeedsWAL)
 replication	replication	with the same content the primary reads	15;16;17;18;19	never	-
+rewrite_group_scan	rewrite_group_scan	a group under the cap still rewrites correctly	15;16;17;18;19	2026-10-20	compact_rewrite does nothing and reports 0;the rewrite drops the first row of every group it moves
+rewrite_group_scan	rewrite_group_scan	a rewrite of a group over the cache cap finishes	15;16;17;18;19	2026-10-20	compact_rewrite does nothing and reports 0
+rewrite_group_scan	rewrite_group_scan	an index lookup finds a rewritten row	15;16;17;18;19	2026-10-21	columnar_vacuum.c: rewrite_one_group skips PgColumnarIndexInsertRow, so the moved rows get no index entries
+rewrite_group_scan	rewrite_group_scan	and it agrees with a heap mirror	15;16;17;18;19	2026-10-21	columnar_vacuum.c: rewrite_one_group skips PgColumnarIndexInsertRow, so the moved rows get no index entries;the rewrite drops the first row of every group it moves
+rewrite_group_scan	rewrite_group_scan	premise: the fixture holds the rows that survive the delete	15;16;17;18;19	never	-
+rewrite_group_scan	rewrite_group_scan	the index lookup's plan reads rgs_id	15;16;17;18;19	never	-
+rewrite_group_scan	rewrite_group_scan	the rewrite preserved every surviving row	15;16;17;18;19	2026-10-20	the rewrite drops the first row of every group it moves
 rewrite_storage_oid	rewrite_storage_oid	a matview created WITH DATA leaves its storage row pointing at itself	15;16;17;18;19	2026-10-12	the repair gate skips every relation, as if no inheritor were locked by the statement;the rewrite repair gate no longer covers CreateTableAsStmt, so a matview created WITH DATA keeps the transient's OID
 rewrite_storage_oid	rewrite_storage_oid	a table that was not rewritten still points at itself	15;16;17;18;19	2026-10-16	the storage repoint after a rewrite updated the first storage row, not the rewritten table's (mutation, #1339)
 rewrite_storage_oid	rewrite_storage_oid	a type rewrite leaves the storage row pointing at the live table	15;16;17;18;19	2026-10-12	the repair gate skips every relation, as if no inheritor were locked by the statement;the rewrite hook does not retarget storage.relation_oid

test/check_ledger_budget.txt

@@ -15,7 +15,7 @@
 #
 #   Adding a check to a suite that is already covered does not move it, which is
 #   what makes it safe to bound.
-suites_not_covered 126
+suites_not_covered 125
 #
 # checks_never_observed_red -- A CENSUS. NOT a ceiling, and it must not become
 #   one.
@@ -4943,4 +4943,24 @@ suites_not_covered 126
 #     never         3641 + 7 = 3648
 #     dated         1923 + 2 = 1925
 #     3648 + 1925 == 5573
-checks_never_observed_red 3648
+#
+# #1341, rewrite_group_scan seeded, with its twin (jdatcmd): 3505 -> 3507, and
+# suites_not_covered 149 -> 148. The suite had no rows. It enters with 7 checks from
+# five-major runs (7/7 on each of 15-19, shell and twin alike), after three fixes,
+# one added premise and one added check. Nothing asserted that the rewrite did any
+# work: the "finishes" check counted only a timeout message, so with compact_rewrite
+# returning 0 untouched every check passed. Both rewrites now read "finished" only for
+# a positive count. And the index was built after the rewrite, so the lookup read a
+# fresh index and a rewrite that inserted no index entries passed every check
+# (OffgridwithJD's review); it is now built before, and a new check holds the lookup to
+# a plan that reads it. 5 are dated by three mutations, each reddening the same names
+# in both twins: the rewrite doing nothing, dropping one row per group, and inserting
+# no index entries. 2 arrive never: the survivors premise and the plan check. The
+# #196 cliff itself (a per-row re-decode past the cache cap) is not among them: it is
+# a timing property, which the statement_timeout still bounds.
+#
+#     ledger rows   5573 + 7 = 5580
+#     never         3648 + 2 = 3650
+#     dated         1925 + 5 = 1930
+#     3650 + 1930 == 5580
+checks_never_observed_red 3650

test/pytest/TESTS.md

@@ -287,6 +287,7 @@ behaviour, the source of that number is named.
 - [239. test_groupagg_table_sizing.py: the vectorized grouped aggregate sizes its table from the estimate, bounded](#239-test_groupagg_table_sizingpy-the-vectorized-grouped-aggregate-sizes-its-table-from-the-estimate-bounded)
 - [240. test_fsst_margin.py: the FSST keep/drop margin decides, and never changes content](#240-test_fsst_marginpy-the-fsst-keepdrop-margin-decides-and-never-changes-content)
 - [241. test_index_delete_liveness.py: nbtree's deletion passes never destroy an entry for a live row](#241-test_index_delete_livenesspy-nbtrees-deletion-passes-never-destroy-an-entry-for-a-live-row)
+- [242. test_rewrite_group_scan.py: a rewrite reads each row group once, not once per row](#242-test_rewrite_group_scanpy-a-rewrite-reads-each-row-group-once-not-once-per-row)
 
 ## 1. How to read a test in here
 
@@ -9656,3 +9657,21 @@ Every test reads one value from a module fixture that builds a private database
 | `test_at_shipped_defaults_count_star_is_right` | at shipped defaults count(*) is right |
 | `test_at_shipped_defaults_count_star_and_count_b_agree` | at shipped defaults the table answers the same to count(*) and count(b) |
 | `test_oracle_the_heap_answers_count_star_correctly` | oracle: the heap answers count(*) correctly after the same churn |
+
+## 242. test_rewrite_group_scan.py: a rewrite reads each row group once, not once per row
+
+The pytest twin of `test/rewrite_group_scan.sh` (ported for #1341), every check by the shell's name. A group whose decoded form exceeds the fetch cache cap used to be re-decoded once per row during a rewrite (#196). The fixture crosses the cap, and the rewrite must finish within a generous statement_timeout, keep every surviving row, agree with a heap mirror, and keep the rows reachable through an index built before it. On a private database.
+
+### Every test
+
+A rewrite reads `finished` only when compact_rewrite returned a positive count. The fixture is held to its survivors first. The index `rgs_id` is built before the rewrite, so the rewrite has to insert the moved rows' index entries itself.
+
+| test | what it holds |
+| --- | --- |
+| `test_the_fixture_holds_its_survivors` | premise: the fixture holds the rows that survive the delete |
+| `test_a_rewrite_over_the_cache_cap_finishes` | a rewrite of a group over the cache cap finishes |
+| `test_the_rewrite_preserved_every_surviving_row` | the rewrite preserved every surviving row |
+| `test_the_rewrite_agrees_with_a_heap_mirror` | and it agrees with a heap mirror |
+| `test_a_group_under_the_cap_still_rewrites_correctly` | a group under the cap still rewrites correctly |
+| `test_the_index_lookup_reads_the_index` | the index lookup's plan reads rgs_id |
+| `test_an_index_lookup_finds_a_rewritten_row` | an index lookup finds a rewritten row |

test/pytest/expected_tests.txt

@@ -981,4 +981,6 @@ guard_tests 467
 # 17 tests, one per check, by collection over the cluster files.
 # cluster_tests 2153 -> 2162 for #1341's test_index_delete_liveness.py, the twin of index_delete_liveness.sh:
 # 9 tests, one per check, by collection over the cluster files.
-cluster_tests 2162
+# cluster_tests 2162 -> 2169 for #1341's test_rewrite_group_scan.py, the twin of rewrite_group_scan.sh:
+# 7 tests, one per check, by collection over the cluster files.
+cluster_tests 2169

test/pytest/test_compare_to_bash.py

@@ -91,6 +91,7 @@
             "smoke",
             "sort_status",
             "sort_status_privilege",
+            "rewrite_group_scan",
             "index_delete_liveness",
             "fsst_margin",
             "groupagg_table_sizing",

test/pytest/test_rewrite_group_scan.py

@@ -0,0 +1,151 @@
+"""A rewrite reads each row group once, not once per row (#196). The pytest twin of
+`test/rewrite_group_scan.sh`, every check by the shell's name (#1341).
+
+rewrite_one_group used to fetch every row by number, and each fetch decoded the whole
+group whenever the group's decoded form exceeded the fetch cache cap: a group 3.5% over
+the cap took longer than two minutes to rewrite. The fixture here crosses the cap (three
+columns, one varlena, 200,000 rows with a third deleted). Asserted: compact_rewrite
+rewrites it within a generous statement_timeout, keeps every surviving row exactly,
+agrees with a heap mirror, still rewrites a group under the cap correctly, and leaves
+the rows reachable through an index built before it, through a plan that reads that
+index.
+
+WHAT A REWRITE READING IS: "finished" only when compact_rewrite returned a positive
+count of rewritten groups, "rewrote nothing", or the error it raised. A timeout was the
+one failure counted before, so a rewrite that did nothing at all "finished", and every
+check passed with compact_rewrite returning 0 untouched. The fixture is held to its
+survivors, since every comparison agrees over an empty table.
+
+ON A PRIVATE DATABASE, built in the shell's order by a module fixture.
+"""
+import psycopg
+import pytest
+
+from pgc_vacuity import query_error
+
+DB = "pgc_rewrite_group_scan"
+ROWS = 200000
+LIMIT = "60s"
+DIGEST = "SELECT count(*) || '/' || md5(string_agg(id || ':' || k || ':' || v, ',' ORDER BY id)) FROM rgs"
+SMALL = "SELECT count(*) || '/' || coalesce(sum(id), 0) FROM rgs_s"
+PROBE = ROWS // 2 + 1
+
+
+class _Db:
+    def __init__(self, dsn):
+        self.dsn = dsn
+
+    def run(self, *sqls):
+        try:
+            with psycopg.connect(self.dsn, autocommit=True) as conn:
+                for sql in sqls:
+                    conn.execute(sql)
+        except psycopg.Error as exc:
+            print(f"-- step failed: {str(exc).splitlines()[0]}")
+
+    def q(self, sql, settings=()):
+        try:
+            with psycopg.connect(self.dsn, autocommit=True) as conn:
+                for s in settings:
+                    conn.execute(s)
+                row = conn.execute(sql).fetchone()
+        except psycopg.Error as exc:
+            return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__)
+        return "" if row is None or row[0] is None else str(row[0])
+
+    def plan(self, sql, settings=()):
+        """-> the whole EXPLAIN (COSTS OFF) text of sql, or its error as a value."""
+        try:
+            with psycopg.connect(self.dsn, autocommit=True) as conn:
+                for s in settings:
+                    conn.execute(s)
+                rows = conn.execute(f"EXPLAIN (COSTS OFF) {sql}").fetchall()
+        except psycopg.Error as exc:
+            return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__)
+        return "\n".join(r[0] for r in rows)
+
+    def rewrite(self, table):
+        """-> "finished", "rewrote nothing", or "no reading: <error>"."""
+        try:
+            with psycopg.connect(self.dsn, autocommit=True) as conn:
+                conn.execute(f"SET statement_timeout = '{LIMIT}'")
+                n = conn.execute(f"SELECT pgcolumnar.compact_rewrite('{table}', 0.2)").fetchone()[0]
+        except psycopg.Error as exc:
+            return f"no reading: {str(exc).splitlines()[0] if str(exc) else type(exc).__name__}"
+        return "finished" if n and n > 0 else "rewrote nothing"
+
+
+@pytest.fixture(scope="module")
+def rg(pgc_cluster):
+    with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a:
+        a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)")
+        a.execute(f"CREATE DATABASE {DB}")
+    try:
+        d = _Db(pgc_cluster.dsn(dbname=DB))
+        d.run("CREATE EXTENSION IF NOT EXISTS pgcolumnar")
+        r = {}
+        d.run("CREATE TABLE rgs (id int, k int, v text) USING pgcolumnar",
+              f"INSERT INTO rgs SELECT g, g % 100, repeat('m', 80) || g FROM generate_series(1, {ROWS}) g",
+              "DELETE FROM rgs WHERE id % 3 = 0",
+              "CREATE TABLE rgs_h (id int, k int, v text)",
+              f"INSERT INTO rgs_h SELECT g, g % 100, repeat('m', 80) || g FROM generate_series(1, {ROWS}) g "
+              "WHERE g % 3 <> 0")
+        r["survivors"] = d.q("SELECT count(*) FROM rgs")
+        # Built before the rewrite, so the rewrite must maintain it: built afterwards,
+        # it was fresh over the rewritten table and a rewrite inserting no index
+        # entries passed every check.
+        d.run("CREATE INDEX rgs_id ON rgs (id)")
+        r["before"] = d.q(DIGEST)
+        r["big"] = d.rewrite("rgs")
+        r["after"] = d.q(DIGEST)
+        r["mirror"] = d.q("SELECT count(*) FROM rgs r FULL JOIN rgs_h h USING (id) "
+                          "WHERE r.v IS DISTINCT FROM h.v OR r.k IS DISTINCT FROM h.k")
+        d.run("CREATE TABLE rgs_s (id int, k int, v text) USING pgcolumnar",
+              "INSERT INTO rgs_s SELECT g, g % 100, repeat('m', 80) || g FROM generate_series(1, 20000) g",
+              "DELETE FROM rgs_s WHERE id % 3 = 0")
+        small_before = d.q(SMALL)
+        small = d.rewrite("rgs_s")
+        # Pairs, not one string: a failed read stays a failed-query value the comparison
+        # refuses, instead of text inside a string that two failures could make agree.
+        r["small"], r["small want"] = (small, d.q(SMALL)), ("finished", small_before)
+        lookup = ["SET enable_seqscan = off", "SET pgcolumnar.enable_custom_scan = off"]
+        r["index plan"] = d.plan(f"SELECT v FROM rgs WHERE id = {PROBE}", lookup)
+        r["index"] = d.q(f"SELECT v FROM rgs WHERE id = {PROBE}", lookup)
+        r["index want"] = d.q(f"SELECT v FROM rgs_h WHERE id = {PROBE}")
+        yield r
+    finally:
+        with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a:
+            a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)")
+
+
+def _num(v):
+    return int(v) if isinstance(v, str) and v.isdigit() else v
+
+
+def test_the_fixture_holds_its_survivors(rg, expect):
+    expect.num(_num(rg["survivors"]), ROWS - ROWS // 3,
+               "premise: the fixture holds the rows that survive the delete")
+
+
+def test_a_rewrite_over_the_cache_cap_finishes(rg, expect):
+    expect.text(rg["big"], "finished", "a rewrite of a group over the cache cap finishes")
+
+
+def test_the_rewrite_preserved_every_surviving_row(rg, expect):
+    expect.text(rg["after"], rg["before"], "the rewrite preserved every surviving row")
+
+
+def test_the_rewrite_agrees_with_a_heap_mirror(rg, expect):
+    expect.text(rg["mirror"], "0", "and it agrees with a heap mirror")
+
+
+def test_a_group_under_the_cap_still_rewrites_correctly(rg, expect):
+    expect.text(rg["small"], rg["small want"], "a group under the cap still rewrites correctly")
+
+
+def test_the_index_lookup_reads_the_index(rg, expect):
+    expect.contains(rg["index plan"], "rgs_id", "the index lookup's plan reads rgs_id")
+
+
+def test_an_index_lookup_finds_a_rewritten_row(rg, expect):
+    expect.text(rg["index"], rg["index want"], "an index lookup finds a rewritten row")

test/pytest/unported_suites.txt

@@ -104,6 +104,5 @@ read_stream
 reader_buffer_reuse
 recluster_extent
 recluster_gate
-rewrite_group_scan
 row_triggers
 scan_direction

test/rewrite_group_scan.sh

@@ -49,23 +49,44 @@ psql_run "CREATE TABLE rgs_h (id int, k int, v text);" >/dev/null
 psql_run "INSERT INTO rgs_h SELECT g, g % 100, repeat('m', 80) || g
 	FROM generate_series(1, $ROWS) g WHERE g % 3 <> 0;" >/dev/null
 
+# rewrite_of TABLE -> "finished" when compact_rewrite rewrote at least one group
+# within the timeout, "rewrote nothing", or the error it raised. A timeout was the one
+# failure counted, so any other error -- and a rewrite that did nothing at all --
+# "finished": with compact_rewrite returning 0 untouched, all five checks passed.
+rewrite_of() {
+	local out
+	out="$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" \
+		-At -q -c "SET statement_timeout = '$LIMIT';" -c "SELECT pgcolumnar.compact_rewrite('$1', 0.2);" 2>&1)"
+	case "$out" in
+		0) echo "rewrote nothing" ;;
+		''|*[!0-9]*) echo "no reading: $(head -1 <<<"$out")" ;;
+		*) echo finished ;;
+	esac
+}
+
+# The comparisons below agree whatever the rewrite did if the table is empty, so the
+# fixture is held to its survivors first.
+check_num "premise: the fixture holds the rows that survive the delete" \
+	"$(q "SELECT count(*) FROM rgs;")" "$((ROWS - ROWS / 3))"
+
+# The index exists before the rewrite, so the rewrite has to maintain it: each moved
+# row gets its entry under its new row number from rewrite_one_group itself. Built
+# after the rewrite, the index was fresh over the rewritten table, and a rewrite that
+# inserted no index entries at all passed every check.
+psql_run "CREATE INDEX rgs_id ON rgs (id);" >/dev/null
 before="$(q "SELECT count(*) || '/' || md5(string_agg(id || ':' || k || ':' || v, ',' ORDER BY id)) FROM rgs;" | tail -1)"
 
 # --- 1. it finishes at all -----------------------------------------------------
 
-err="$(psql_run "SET statement_timeout = '$LIMIT';
-	SELECT pgcolumnar.compact_rewrite('rgs', 0.2);" 2>&1 || true)"
-
-check "a rewrite of a group over the cache cap finishes" \
-	"$(echo "$err" | grep -ciE 'statement timeout|canceling statement')" "0"
+check "a rewrite of a group over the cache cap finishes" "$(rewrite_of rgs)" "finished"
 
 # --- 2. and moved exactly the right rows ---------------------------------------
 
 # The rewrite must be content-preserving. Checking the digest rather than the
 # count, because a rewrite that dropped and re-added the same number of rows
 # would pass a count.
 after="$(q "SELECT count(*) || '/' || md5(string_agg(id || ':' || k || ':' || v, ',' ORDER BY id)) FROM rgs;" | tail -1)"
-check "the rewrite preserved every surviving row" "$after" "$before"
+check_text "the rewrite preserved every surviving row" "$after" "$before"
 
 check "and it agrees with a heap mirror" \
 	"$(q "SELECT count(*) FROM rgs r FULL JOIN rgs_h h USING (id)
@@ -82,18 +103,27 @@ psql_run "INSERT INTO rgs_s SELECT g, g % 100, repeat('m', 80) || g
 	FROM generate_series(1, 20000) g;" >/dev/null
 psql_run "DELETE FROM rgs_s WHERE id % 3 = 0;" >/dev/null
 small_before="$(q "SELECT count(*) || '/' || coalesce(sum(id), 0) FROM rgs_s;" | tail -1)"
-psql_run "SET statement_timeout = '$LIMIT';
-	SELECT pgcolumnar.compact_rewrite('rgs_s', 0.2);" >/dev/null 2>&1
-check "a group under the cap still rewrites correctly" \
-	"$(q "SELECT count(*) || '/' || coalesce(sum(id), 0) FROM rgs_s;" | tail -1)" "$small_before"
+small_rewrite="$(rewrite_of rgs_s)"
+small_after="$(q "SELECT count(*) || '/' || coalesce(sum(id), 0) FROM rgs_s;" | tail -1)"
+# q reads a failure as empty, so each side names its own missing reading: two failed
+# reads must not agree as "finished ".
+check_text "a group under the cap still rewrites correctly" \
+	"$small_rewrite ${small_after:-no reading after}" "finished ${small_before:-no reading before}"
 
 # --- 4. the rows are still reachable through an index --------------------------
 
-# A rewrite reinserts every row's index entries under new row numbers. If the
-# streaming read handed the writer rows in a different order or skipped one, an
-# index lookup is where it shows.
-psql_run "CREATE INDEX rgs_id ON rgs (id);" >/dev/null
-check "an index lookup finds a rewritten row" \
+# A rewrite reinserts every row's index entries under new row numbers, into the
+# index built before it. If it skipped them, or the streaming read handed the writer
+# rows in a different order or skipped one, an index lookup is where it shows.
+#
+# The lookup is held to a plan that reads rgs_id, so a planner change cannot quietly
+# turn it into a sequential-scan comparison. Not a premise: pgcolumnar's own
+# set_rel_pathlist_hook prices index paths on its tables, so this reads more than core.
+check_text "the index lookup's plan reads rgs_id" \
+	"$(q "SET enable_seqscan = off; SET pgcolumnar.enable_custom_scan = off;
+		EXPLAIN (COSTS OFF) SELECT v FROM rgs WHERE id = $((ROWS / 2 + 1));" |
+		grep -o 'rgs_id' | head -1)" "rgs_id"
+check_text "an index lookup finds a rewritten row" \
 	"$(q "SET enable_seqscan = off; SET pgcolumnar.enable_custom_scan = off;
 		SELECT v FROM rgs WHERE id = $((ROWS / 2 + 1));" | tail -1)" \
 	"$(q "SELECT v FROM rgs_h WHERE id = $((ROWS / 2 + 1));" | tail -1)"