diff --git a/test/check_ledger.tsv b/test/check_ledger.tsv
index dbcc327ba..ebf54e47a 100644
--- a/test/check_ledger.tsv
+++ b/test/check_ledger.tsv
@@ -497,6 +497,26 @@ column_extent	column_extent	the extent of ty.s is the data's min and max, in the
 column_extent	column_extent	the extent of ty.ts is the data's min and max, in the type's order	15;16;17;18;19	2026-10-14	a writer that finds no complete extent does not recompute one;no extent row is ever written (PgColumnarExtentPreCommit returns at once)
 column_extent	column_extent	while the column that was there from the start still has one	15;16;17;18;19	2026-10-14	a writer that finds no complete extent does not recompute one;no extent row is ever written (PgColumnarExtentPreCommit returns at once)
 column_extent	column_extent	without the extent rows the planner falls back to one zone_map scan	15;16;17;18;19	2026-10-13	the planner's zone-map fallback removed: with no complete extent it reads nothing and bounds nothing
+corruption	corruption	alive after bad descriptor	15;16;17;18;19	2026-10-16	columnar_reader.c: the unrecognized-descriptor refusal replaced by a backend crash
+corruption	corruption	alive after bloom corruption	15;16;17;18;19	2026-10-16	columnar_bloom.c: the bloom length check removed
+corruption	corruption	alive after byte_length	15;16;17;18;19	2026-10-16	columnar_reader.c: the row group span refusal replaced by a backend crash
+corruption	corruption	alive after column_index=-1	15;16;17;18;19	never	-
+corruption	corruption	alive after page_length	15;16;17;18;19	2026-10-16	columnar_reader.c: the chunk-inside-its-group refusal replaced by a backend crash
+corruption	corruption	alive after row_count	15;16;17;18;19	2026-10-16	columnar_reader.c: the no-nulls row accounting and the value-stream bound both removed
+corruption	corruption	alive after value_count	15;16;17;18;19	2026-10-16	columnar_reader.c: the chunk-inside-its-group refusal replaced by a backend crash
+corruption	corruption	bad descriptor errors	15;16;17;18;19	2026-10-16	columnar_reader.c: the unrecognized-descriptor refusal replaced by a backend crash
+corruption	corruption	byte_length corruption errors	15;16;17;18;19	2026-10-16	columnar_reader.c: the row group span refusal replaced by a backend crash;columnar_reader.c: the row group's span checked against its chunks no longer
+corruption	corruption	corrupt bloom equality still correct	15;16;17;18;19	never	-
+corruption	corruption	corrupt bloom: a value the table holds is still found	15;16;17;18;19	2026-10-16	columnar_bloom.c: a malformed bloom header read as absent rather than as no filter;columnar_bloom.c: the bloom length check removed
+corruption	corruption	page_length corruption errors	15;16;17;18;19	2026-10-16	columnar_reader.c: the chunk-inside-its-group refusal replaced by a backend crash
+corruption	corruption	premise: the bloom filter corruption changed a catalog row	15;16;17;18;19	never	-
+corruption	corruption	premise: the byte_length corruption changed a catalog row	15;16;17;18;19	never	-
+corruption	corruption	premise: the column_index corruption changed a catalog row	15;16;17;18;19	never	-
+corruption	corruption	premise: the encoding_descriptor corruption changed a catalog row	15;16;17;18;19	never	-
+corruption	corruption	premise: the page_length corruption changed a catalog row	15;16;17;18;19	never	-
+corruption	corruption	premise: the row_count corruption changed a catalog row	15;16;17;18;19	never	-
+corruption	corruption	premise: the value_count corruption changed a catalog row	15;16;17;18;19	never	-
+corruption	corruption	row_count corruption errors	15;16;17;18;19	2026-10-16	columnar_reader.c: the no-nulls row accounting and the value-stream bound both removed
 debug_hook_privilege	debug_hook_privilege	and that refusal is the owner gate too	15;16;17;18;19	2026-10-14	the debug set-metapage-version hook without its owner check
 debug_hook_privilege	debug_hook_privilege	and the refusal is the owner gate, not an EXECUTE denial	15;16;17;18;19	2026-10-14	pgcolumnar_debug_advance_reserved_offset without its owner check
 debug_hook_privilege	debug_hook_privilege	backend alive	15;16;17;18;19	never	-
@@ -2116,7 +2136,7 @@ inheritance	inheritance	SELECT * from an INHERITS parent plans an Append	15;16;1
 inheritance	inheritance	a columnar materialized view is still recognised as columnar	15;16;17;18;19	2026-09-24	the predicate narrowed to `relkind == RELKIND_RELATION` instead of RELKIND_HAS_STORAGE -- the obvious wrong fix, which also excludes RELKIND_MATVIEW and stops a columnar materialized view being recognised at all 31 call sites
 inheritance	inheritance	a large child is still visible through the parent	15;16;17;18;19	never	-
 inheritance	inheritance	a partitioned parent still returns every row	15;16;17;18;19	never	-
-inheritance	inheritance	add_projection on the parent is refused rather than aborting the backend	15;16;17;18;19	never	-
+inheritance	inheritance	add_projection on the parent is refused rather than aborting the backend	15;16;17;18;19	2026-10-16	columnar_metadata.c: PgColumnarIsColumnarRelation answers from relam alone again, the #1259 predicate reverted
 inheritance	inheritance	and it still returns its rows	15;16;17;18;19	never	-
 inheritance	inheritance	and the cluster is still up after it	15;16;17;18;19	never	-
 inheritance	inheritance	and the cluster is still up after that too	15;16;17;18;19	never	-
@@ -2125,7 +2145,7 @@ inheritance	inheritance	control: add_projection on the LEAF still works	15;16;17
 inheritance	inheritance	control: get_storage_id on the LEAF still returns an id	15;16;17;18;19	never	-
 inheritance	inheritance	count(*) from an INHERITS parent plans an Append	15;16;17;18;19	never	-
 inheritance	inheritance	filtered rows match the heap	15;16;17;18;19	never	-
-inheritance	inheritance	get_storage_id on the parent is refused rather than reading storage	15;16;17;18;19	never	-
+inheritance	inheritance	get_storage_id on the parent is refused rather than reading storage	15;16;17;18;19	2026-10-16	columnar_metadata.c: PgColumnarIsColumnarRelation answers from relam alone again, the #1259 predicate reverted
 inheritance	inheritance	premise: a materialized view is relkind m	15;16;17;18;19	never	-
 inheritance	inheritance	premise: and it carries the columnar access method	15;16;17;18;19	never	-
 inheritance	inheritance	premise: and it is partitioned	15;16;17;18;19	never	-
@@ -2226,6 +2246,34 @@ native_agg	native_agg	sum(int4)	15;16;17;18;19	never	-
 native_agg	native_agg	sum(int4,nulls)	15;16;17;18;19	never	-
 native_agg	native_agg	sum/min/max uses the metadata agg node	15;16;17;18;19	never	-
 native_agg	native_agg	the deletes removed rows at both ends, the same rows from each	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	a second concurrent reader was also dropped by the crash	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	crashA: kill lands mid-write (backend not yet in pg_sleep)	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	crashA: the INSERT had written pages before the kill (the relation grew)	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	crashB: kill lands after the stripes flushed (backend in pg_sleep)	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	crashB: the INSERT had written pages before the kill (the relation grew)	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	crashC: kill lands mid-write (backend not yet in pg_sleep)	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	crashC: the INSERT had written pages before the kill (the relation grew)	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	mid-write: cluster accepts new connections	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	mid-write: committed content in c matches the heap mirror	15;16;17;18;19	2026-10-16	columnar_storage.c: a stripe data page written with no WAL record
+native_backend_crash	native_backend_crash	mid-write: postmaster survived, same pid	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	mid-write: the crash reinitialized the cluster (a live session was dropped)	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	mid-write: the killed insert left no partial row group	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	mid-write: the killed insert left no visible rows	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	mid-write: the mirror itself is unchanged	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	post-flush: cluster accepts new connections	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	post-flush: committed content in c matches the heap mirror	15;16;17;18;19	2026-10-16	columnar_storage.c: a stripe data page written with no WAL record
+native_backend_crash	native_backend_crash	post-flush: postmaster survived, same pid	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	post-flush: the crash reinitialized the cluster (a live session was dropped)	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	post-flush: the killed insert left no partial row group	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	post-flush: the killed insert left no visible rows	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	post-flush: the mirror itself is unchanged	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	with-reader: cluster accepts new connections	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	with-reader: committed content in c matches the heap mirror	15;16;17;18;19	2026-10-16	columnar_storage.c: a stripe data page written with no WAL record
+native_backend_crash	native_backend_crash	with-reader: postmaster survived, same pid	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	with-reader: the crash reinitialized the cluster (a live session was dropped)	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	with-reader: the killed insert left no partial row group	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	with-reader: the killed insert left no visible rows	15;16;17;18;19	never	-
+native_backend_crash	native_backend_crash	with-reader: the mirror itself is unchanged	15;16;17;18;19	never	-
 native_chunk_length_bound	native_chunk_length_bound	a sequential scan of the same poisoned chunk is refused (XX001)	15;16;17;18;19	never	-
 native_chunk_length_bound	native_chunk_length_bound	an index fetch of a chunk whose page_length is 2^32 too large is refused (XX001)	15;16;17;18;19	never	-
 native_chunk_length_bound	native_chunk_length_bound	backend survived the sequential refusal	15;16;17;18;19	never	-
@@ -2352,6 +2400,9 @@ native_join_runtime_filter	native_join_runtime_filter	a (SELECT false) gate over
 native_join_runtime_filter	native_join_runtime_filter	a (SELECT false) gate over a heap dimension equals heap	15;16;17;18;19	2026-10-15	the join runtime filter took a join carrying a gating qual and failed on the Result over it (the tree before #1437)
 native_join_runtime_filter	native_join_runtime_filter	a (SELECT true) gate over a columnar dimension equals heap	15;16;17;18;19	2026-10-15	the join runtime filter took a join carrying a gating qual and failed on the Result over it (the tree before #1437)
 native_join_runtime_filter	native_join_runtime_filter	a (SELECT true) gate over a heap dimension equals heap	15;16;17;18;19	2026-10-15	the join runtime filter took a join carrying a gating qual and failed on the Result over it (the tree before #1437)
+native_join_runtime_filter	native_join_runtime_filter	an empty build from a large dimension answers as heap	15;16;17;18;19	never	-
+native_join_runtime_filter	native_join_runtime_filter	an empty build on one rescan does not empty the next	15;16;17;18;19	2026-10-16	the rescan kept the first execution's runtime filter state: prepared not reset on rescan (mutation, #1447)
+native_join_runtime_filter	native_join_runtime_filter	an empty build reads no fact chunk group	15;16;17;18;19	2026-10-16	an inner join whose build was empty ran on, and the fact scan rejected every row one at a time (the tree before #1447)
 native_join_runtime_filter	native_join_runtime_filter	baseline core Hash Join	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	baseline reads all groups	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	build rows omit NULL	15;16;17;18;19	never	-
@@ -2377,8 +2428,12 @@ native_join_runtime_filter	native_join_runtime_filter	join runtime filter defaul
 native_join_runtime_filter	native_join_runtime_filter	plan has build tap	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	plan has runtime coordinator	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	plan retains core Hash Join	15;16;17;18;19	never	-
+native_join_runtime_filter	native_join_runtime_filter	premise: and its build is empty	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	premise: heap's answer under a true gate over a columnar dimension holds 200 rows	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	premise: heap's answer under a true gate over a heap dimension holds 200 rows	15;16;17;18;19	never	-
+native_join_runtime_filter	native_join_runtime_filter	premise: heap's empty-then-full rescan answer holds the 200 keys	15;16;17;18;19	never	-
+native_join_runtime_filter	native_join_runtime_filter	premise: the empty-then-full rescan runs under the coordinator	15;16;17;18;19	never	-
+native_join_runtime_filter	native_join_runtime_filter	premise: the large-dimension empty build runs under the coordinator	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	premise: with no gate, the columnar-dimension join has the coordinator	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	premise: with no gate, the heap-dimension join has the coordinator	15;16;17;18;19	never	-
 native_join_runtime_filter	native_join_runtime_filter	projection join equals heap	15;16;17;18;19	never	-
@@ -3562,10 +3617,3 @@ vm_privilege	vm_privilege	vm entry point is covered by this boundary test: vm_se
 vm_privilege	vm_privilege	vm_is_visible refuses a heap relation even for its owner (#748)	15;16;17;18;19	never	-
 vm_privilege	vm_privilege	vm_selftest did not report success to the unprivileged role	15;16;17;18;19	2026-10-14	vm_selftest without its owner check
 vm_privilege	vm_privilege	vm_selftest refuses a heap relation even for its owner (#748)	15;16;17;18;19	never	-
-native_join_runtime_filter	native_join_runtime_filter	premise: the large-dimension empty build runs under the coordinator	15;16;17;18;19	never	-
-native_join_runtime_filter	native_join_runtime_filter	premise: and its build is empty	15;16;17;18;19	never	-
-native_join_runtime_filter	native_join_runtime_filter	an empty build reads no fact chunk group	15;16;17;18;19	2026-10-16	an inner join whose build was empty ran on, and the fact scan rejected every row one at a time (the tree before #1447)
-native_join_runtime_filter	native_join_runtime_filter	an empty build from a large dimension answers as heap	15;16;17;18;19	never	-
-native_join_runtime_filter	native_join_runtime_filter	premise: the empty-then-full rescan runs under the coordinator	15;16;17;18;19	never	-
-native_join_runtime_filter	native_join_runtime_filter	premise: heap's empty-then-full rescan answer holds the 200 keys	15;16;17;18;19	never	-
-native_join_runtime_filter	native_join_runtime_filter	an empty build on one rescan does not empty the next	15;16;17;18;19	2026-10-16	the rescan kept the first execution's runtime filter state: prepared not reset on rescan (mutation, #1447)
diff --git a/test/check_ledger_budget.txt b/test/check_ledger_budget.txt
index e4c69dd22..e3e62de0d 100644
--- a/test/check_ledger_budget.txt
+++ b/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 221
+suites_not_covered 219
 #
 # checks_never_observed_red -- A CENSUS. NOT a ceiling, and it must not become
 #   one.
@@ -2886,4 +2886,51 @@ suites_not_covered 221
 #     never         2639 + 5 = 2644
 #     dated          925 + 2 = 927
 #     2644 + 927 == 3571
-checks_never_observed_red 2644
+#
+# #1341, native_backend_crash seeded, with its twin (OffgridwithJD): 2644 -> 2669,
+# and suites_not_covered 221 -> 220. The suite had no rows. It enters with all 28
+# checks from five-major runs (28/28 on each of 15-19). 3 are dated by a stripe data
+# page written with no WAL record: each scenario's "committed content in c matches
+# the heap mirror", which the reinit's crash recovery cannot rebuild. Twenty-five
+# arrive never. The no-visible-rows and no-partial-row-group arms are MVCC's: no
+# mutation of the extension leaves an aborted transaction's catalog rows visible.
+# The sentinel, second-reader, kill-phase and relation-grew arms go red only under
+# harness mutations (a signal that kills nothing, an INSERT whose stripes buffer
+# whole), which observe the suite rather than the code. The rest hold of the heap
+# mirror, the postmaster pid and a connection.
+#
+#     ledger rows   3571 + 28 = 3599
+#     never         2644 + 25 = 2669
+#     dated          927 +  3 = 930
+#     2669 + 930 == 3599
+#
+# #1341, corruption seeded, with its twin (OffgridwithJD): 2669 -> 2678, and
+# suites_not_covered 220 -> 219. The suite had no rows. It enters with all 20 checks
+# from five-major runs (20/20 on each of 15-19). 11 are dated by seven mutations of
+# the reader and the bloom probe: the group-span check removed, the span, the
+# chunk-bound and the descriptor refusals each replaced by a backend crash, the
+# row-count accounting and the value-stream bound removed together, a malformed
+# bloom header read as "absent", and the bloom length check removed. Each corruption
+# has a second guard, so removing one refusal alone mostly leaves its arm green; a
+# refusal turned into a crash is what the new XX001 and crash-count arms catch.
+# Nine arrive never: the seven touched-a-row premises, which redden only when the
+# harness's UPDATEs match nothing, column_index's liveness, which no single bound
+# reddens, and the absent-value bloom lookup, which cannot see a false negative.
+#
+#     ledger rows   3599 + 20 = 3619
+#     never         2669 +  9 = 2678
+#     dated          930 + 11 = 941
+#     2678 + 941 == 3619
+#
+# #1341 item 29, inheritance re-dated (OffgridwithJD): 2678 -> 2676. Its two #1259
+# refusal arms counted any error as a refusal, and with the #1259 predicate
+# reverted a release build fails with 58P01 opening relfilenode 0, which is the bug:
+# the suite stayed 26/26. They now require 42809, and the revert reddens both, so
+# both leave never. parquet_import, phase5 and drop_cleanup gain partial twins but
+# are not ledger-covered suites, so they owe no rows here.
+#
+#     ledger rows   3619
+#     never         2678 - 2 = 2676
+#     dated          941 + 2 = 943
+#     2676 + 943 == 3619
+checks_never_observed_red 2676
diff --git a/test/corruption.sh b/test/corruption.sh
index 99d8a1c06..f34c1614d 100755
--- a/test/corruption.sh
+++ b/test/corruption.sh
@@ -21,11 +21,39 @@ set -uo pipefail
 
 pgc_setup "${1:-/usr/local/pg17/bin/pg_config}"
 
-alive() { [ "$(q 'SELECT 1;')" = "1" ] && echo yes || echo no; }
-errors() { if psql_run "$1" >/dev/null 2>&1; then echo no; else echo yes; fi; }
+# A BACKEND CRASH PASSED THESE CHECKS (#1341). `errors` said yes to any failure of
+# psql, and a crash is one. `alive` then connected after the postmaster's crash
+# recovery, so it passed whenever recovery finished first: with the bad-descriptor
+# refusal replaced by abort(), the suite was 12/12 once `alive` waited 3s. So a
+# refusal must now be a clean XX001, which a crash never prints, and `alive` also
+# requires that no backend has died since the scenario started, read from the
+# server log rather than from timing.
+crashes() {
+	# grep -c prints 0 itself when nothing matches, and then exits 1
+	pgc_pg "grep -c 'terminated by signal' '$PGC_LOGFILE'" 2>/dev/null || true
+}
+alive() {	# since this scenario's mkc, which notes the count
+	[ "$(q 'SELECT 1;')" = "1" ] && [ "$(crashes)" = "$CRASHES_BEFORE" ] && echo yes \
+		|| echo "no ($(( $(crashes) - CRASHES_BEFORE )) backend crashes)"
+}
+refused() {	# refused SQL -> the SQLSTATE of the ERROR it raised, or what it printed instead
+	local out
+	out="$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" \
+		-qAtX -v VERBOSITY=verbose -c "$1" 2>&1)"
+	sed -nE 's/^ERROR:  ([0-9A-Z]{5}): .*/\1/p' <<<"$out" | head -1 | grep . \
+		|| echo "no error ($(head -1 <<<"$out"))"
+}
 sid() { q "SELECT pgcolumnar.get_storage_id('c');"; }
+# corrupt FIELD "UPDATE ..." : run the corrupting UPDATE and assert it changed a
+# catalog row. An UPDATE that matched nothing would leave the "must not crash"
+# scenarios passing on a table nobody corrupted.
+corrupt() {
+	check "premise: the $1 corruption changed a catalog row" \
+		"$(q "WITH u AS ($2 RETURNING 1) SELECT count(*) > 0 FROM u;")" "t"
+}
 
 mkc() {
+	CRASHES_BEFORE="$(crashes)"
 	psql_run "DROP TABLE IF EXISTS c;" >/dev/null 2>&1
 	psql_run "CREATE TABLE c (id int, r bigint, t text) USING pgcolumnar;"
 	# r random -> poorly compressible; id monotonic; t low-cardinality -> dictionary
@@ -38,51 +66,57 @@ mkc() {
 # backend surviving.
 echo "-- row_group.byte_length corruption must error, not crash"
 mkc
-psql_run "UPDATE pgcolumnar.row_group SET byte_length = byte_length + 123456
-          WHERE storage_id = $(sid);"
-check "byte_length corruption errors"   "$(errors 'SELECT count(*), sum(r) FROM c;')" "yes"
+corrupt "byte_length" "UPDATE pgcolumnar.row_group SET byte_length = byte_length + 123456
+          WHERE storage_id = $(sid)"
+check "byte_length corruption errors"   "$(refused 'SELECT count(*), sum(r) FROM c;')" "XX001"
 check "alive after byte_length"         "$(alive)" "yes"
 
 echo "-- row_group.row_count corruption must error, not crash"
 mkc
-psql_run "UPDATE pgcolumnar.row_group SET row_count = row_count + 100000
-          WHERE storage_id = $(sid);"
-check "row_count corruption errors"     "$(errors 'SELECT sum(r) FROM c;')" "yes"
+corrupt "row_count" "UPDATE pgcolumnar.row_group SET row_count = row_count + 100000
+          WHERE storage_id = $(sid)"
+check "row_count corruption errors"     "$(refused 'SELECT sum(r) FROM c;')" "XX001"
 check "alive after row_count"           "$(alive)" "yes"
 
 echo "-- column_chunk.page_length corruption must error, not crash"
 mkc
-psql_run "UPDATE pgcolumnar.column_chunk SET page_length = page_length + 123456
-          WHERE storage_id = $(sid);"
-check "page_length corruption errors"   "$(errors 'SELECT sum(r) FROM c;')" "yes"
+corrupt "page_length" "UPDATE pgcolumnar.column_chunk SET page_length = page_length + 123456
+          WHERE storage_id = $(sid)"
+check "page_length corruption errors"   "$(refused 'SELECT sum(r) FROM c;')" "XX001"
 check "alive after page_length"         "$(alive)" "yes"
 
 echo "-- column_chunk.value_count corruption must not crash (degrades safely)"
 mkc
-psql_run "UPDATE pgcolumnar.column_chunk SET value_count = value_count + 100000
-          WHERE storage_id = $(sid);"
+corrupt "value_count" "UPDATE pgcolumnar.column_chunk SET value_count = value_count + 100000
+          WHERE storage_id = $(sid)"
 q "SELECT sum(r) FROM c;" >/dev/null 2>&1
 check "alive after value_count"         "$(alive)" "yes"
 
 echo "-- corrupt encoding descriptor must error, not crash"
 mkc
-psql_run "UPDATE pgcolumnar.column_chunk SET encoding_descriptor = '\\xffffffff'::bytea
-          WHERE storage_id = $(sid) AND column_index = 1;"
-check "bad descriptor errors"           "$(errors 'SELECT * FROM c;')" "yes"
+corrupt "encoding_descriptor" "UPDATE pgcolumnar.column_chunk SET encoding_descriptor = '\\xffffffff'::bytea
+          WHERE storage_id = $(sid) AND column_index = 1"
+check "bad descriptor errors"           "$(refused 'SELECT * FROM c;')" "XX001"
 check "alive after bad descriptor"      "$(alive)" "yes"
 
 echo "-- column_index = -1 (negative index guard) must not crash"
 mkc
-psql_run "UPDATE pgcolumnar.column_chunk SET column_index = -1
-          WHERE storage_id = $(sid) AND column_index = 1;"
+corrupt "column_index" "UPDATE pgcolumnar.column_chunk SET column_index = -1
+          WHERE storage_id = $(sid) AND column_index = 1"
 q "SELECT count(*) FROM c;" >/dev/null 2>&1
 check "alive after column_index=-1"     "$(alive)" "yes"
 
 echo "-- corrupt bloom header (huge nbits, short buffer) is treated as no filter"
 mkc
-psql_run "UPDATE pgcolumnar.bloom SET filter = '\\x0000004006'::bytea
-          WHERE storage_id = $(sid);"
+BLOOM_HIT="$(q 'SELECT r FROM c WHERE id = 7777;')"
+corrupt "bloom filter" "UPDATE pgcolumnar.bloom SET filter = '\\x0000004006'::bytea
+          WHERE storage_id = $(sid)"
 check "corrupt bloom equality still correct" "$(q 'SELECT count(*) FROM c WHERE r = -999;')" "0"
+# An absent value is 0 rows whether the corrupt filter is ignored or wrongly says
+# "absent", so the arm above cannot see a false negative, which is the bloom's one
+# way to lose rows (#1341). A value the table holds must still be found.
+check "corrupt bloom: a value the table holds is still found" \
+	"$(q "SELECT count(*) FROM c WHERE r = $BLOOM_HIT;")" "1"
 check "alive after bloom corruption"         "$(alive)" "yes"
 
 pgc_summary
diff --git a/test/inheritance.sh b/test/inheritance.sh
index 01af0bb6c..a9998f2d2 100755
--- a/test/inheritance.sh
+++ b/test/inheritance.sh
@@ -143,19 +143,37 @@ check "premise: so it has no storage of its own" \
 check "premise: while its leaf is an ordinary relation" \
 	"$(q "SELECT relkind::text FROM pg_class WHERE relname = 'inh_pp1'")" "r"
 
-# EACH CALL IS FOLLOWED BY A LIVENESS CHECK taken from a new backend, because a
-# crashed cluster restarts and a later query succeeds -- a delayed check reads
-# "up" for a backend that aborted.
-inh_up() { psql_run "SELECT 1;" >/dev/null 2>&1 && echo up || echo DOWN; }
+# EACH CALL IS FOLLOWED BY A LIVENESS CHECK, and the liveness is read from the
+# server log, not from a new backend (#1341). A crashed cluster restarts and a
+# later query succeeds, so a connection reads "up" for a backend that aborted
+# once crash recovery has finished.
+#
+# AND A REFUSAL IS 42809, NOT ANY ERROR (#1341). With the #1259 predicate
+# reverted, a release build does not abort: each call opens relfilenode 0 and
+# fails with 58P01 "could not open file .../0/0", which is the bug, and the arms
+# counted it as refused. inheritance stayed 26/26. CI runs release builds.
+inh_crashes() {
+	# grep -c prints 0 itself when nothing matches, and then exits 1
+	pgc_pg "grep -c 'terminated by signal' '$PGC_LOGFILE'" 2>/dev/null || true
+}
+INH_CRASHES="$(inh_crashes)"
+inh_up() {
+	[ "$(q 'SELECT 1;')" = 1 ] && [ "$(inh_crashes)" = "$INH_CRASHES" ] && echo up \
+		|| echo "DOWN ($(( $(inh_crashes) - INH_CRASHES )) backend crashes)"
+}
+inh_sqlstate() {	# inh_sqlstate SQL -> the SQLSTATE of the ERROR it raised, or "accepted"
+	local out
+	out="$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" \
+		-qAtX -v VERBOSITY=verbose -c "$1" 2>&1)" && { echo accepted; return; }
+	sed -nE 's/^ERROR:  ([0-9A-Z]{5}): .*/\1/p' <<<"$out" | head -1 | grep . || echo "no SQLSTATE ($(head -1 <<<"$out"))"
+}
 
 check "get_storage_id on the parent is refused rather than reading storage" \
-	"$(psql_run "SELECT pgcolumnar.get_storage_id('inh_pp'::regclass);" >/dev/null 2>&1 \
-		&& echo accepted || echo refused)" "refused"
+	"$(inh_sqlstate "SELECT pgcolumnar.get_storage_id('inh_pp'::regclass);")" "42809"
 check "and the cluster is still up after it" "$(inh_up)" "up"
 
 check "add_projection on the parent is refused rather than aborting the backend" \
-	"$(psql_run "SELECT pgcolumnar.add_projection('inh_pp'::regclass, 'p1', ARRAY['a']);" \
-		>/dev/null 2>&1 && echo accepted || echo refused)" "refused"
+	"$(inh_sqlstate "SELECT pgcolumnar.add_projection('inh_pp'::regclass, 'p1', ARRAY['a']);")" "42809"
 check "and the cluster is still up after that too" "$(inh_up)" "up"
 
 # THE CONTROL THAT THE CALLS WORK AT ALL. Without it "refused" is satisfied by a
diff --git a/test/native_backend_crash.sh b/test/native_backend_crash.sh
index 9f46f430f..07769259d 100755
--- a/test/native_backend_crash.sh
+++ b/test/native_backend_crash.sh
@@ -103,8 +103,9 @@ open_sentinel() {
 # backend is parked in the trailing pg_sleep, between the last flush and the
 # commit that never comes. The phase is asserted, not assumed, from wait_event.
 crash_one_backend() {
-	local rows="$1" marker="$2" phase="$3" bg pid i we src
+	local rows="$1" marker="$2" phase="$3" bg pid i we src size0
 	open_sentinel || echo "-- sentinel did not connect"
+	size0="$(q "SELECT pg_relation_size('c');")"
 	env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \
 		-d "$PGC_DB" -c "BEGIN; INSERT INTO c /* $marker */ SELECT g, 'k'||g, g
 			FROM generate_series(1, $rows) g; SELECT pg_sleep(60);" >/dev/null 2>&1 &
@@ -131,8 +132,18 @@ crash_one_backend() {
 			sleep 0.1
 		done
 	else
-		sleep 0.5
+		# Until the write has reached the relation. A fixed 0.5s stood here, and
+		# nothing showed the INSERT had written a page by then: with its stripes
+		# buffered whole (stripe_row_limit above the row count) the kill landed on
+		# a write that had produced nothing, and every mid-write arm still passed
+		# (#1341). So the kill now waits for the relation to grow, and asserts it.
+		for i in $(seq 1 600); do
+			[ "$(q "SELECT pg_relation_size('c') > $size0;")" = t ] && break
+			sleep 0.1
+		done
 	fi
+	check "$marker: the INSERT had written pages before the kill (the relation grew)" \
+		"$(q "SELECT pg_relation_size('c') > $size0;")" "t"
 	we="$(q "SELECT coalesce(wait_event,'') FROM pg_stat_activity WHERE pid = $pid;")"
 	if [ "$phase" = sleep ]; then
 		check "$marker: kill lands after the stripes flushed (backend in pg_sleep)" "$we" "PgSleep"
diff --git a/test/parquet_import.sh b/test/parquet_import.sh
index 21df5b963..11e4c6716 100755
--- a/test/parquet_import.sh
+++ b/test/parquet_import.sh
@@ -281,14 +281,28 @@ else
 	check_num "control: a HEAP target still imports, so the guard is about storage" \
 		"$(q "SELECT pgcolumnar.import_parquet('pq_heap', '$PQFILE');")" "50"
 
-	expect_error "importing into the partitioned parent is refused" \
-		"SELECT pgcolumnar.import_parquet('pq_parent', '$PQFILE');"
+	# A REFUSAL IS 42809, AND THE CRASH IS READ FROM THE SERVER LOG (#1341). With
+	# the guard removed the import SIGSEGVs, and "is refused" passed on the crash,
+	# since a dropped connection is an error too; only the liveness check went red,
+	# and only because it reached the server before crash recovery finished. A
+	# check that connects after recovery reads a live cluster.
+	pq_crashes() {
+		# grep -c prints 0 itself when nothing matches, and then exits 1
+		pgc_pg "grep -c 'terminated by signal' '$PGC_LOGFILE'" 2>/dev/null || true
+	}
+	pq_crashes0="$(pq_crashes)"
+	pq_out="$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" \
+		-qAtX -v VERBOSITY=verbose -c "SELECT pgcolumnar.import_parquet('pq_parent', '$PQFILE');" 2>&1)" \
+		&& pq_state=accepted \
+		|| pq_state="$(sed -nE 's/^ERROR:  ([0-9A-Z]{5}): .*/\1/p' <<<"$pq_out" | head -1 | grep . \
+			|| echo "no SQLSTATE ($(head -1 <<<"$pq_out"))")"
+	check "importing into the partitioned parent is refused" "$pq_state" "42809"
 	# `q`, not `psql_run`: psql_run runs -q with no -At and prints nothing, so this
 	# arm returned empty on a LIVE cluster and could never pass. It read as a
 	# correct red against the unfixed build and would have stayed red against the
 	# fix -- a discriminator that discriminates nothing.
 	check "and the cluster is still up, so the refusal was not an abort" \
-		"$(q "SELECT 1;")" "1"
+		"$(q "SELECT 1;")|$(( $(pq_crashes) - pq_crashes0 )) backend crashes" "1|0 backend crashes"
 fi
 
 pgc_summary
diff --git a/test/phase5.sh b/test/phase5.sh
index 497afe39c..7fc82a451 100755
--- a/test/phase5.sh
+++ b/test/phase5.sh
@@ -94,36 +94,11 @@ run_pg "createdb -p $PORT p5"
 PSQL="psql -p $PORT -d p5 -qAtX -v ON_ERROR_STOP=1"
 q() { run_pg "$PSQL -c \"$1\""; }
 
-fail=0
-# RECORDS RATHER THAN ONLY PRINTING (#965). This suite emitted human PASS lines
-# and no RESULT records, so every mechanism built on the record vocabulary -- the
-# ledger, the census, checks_never_observed_red, the red-observation record,
-# duplicate-name detection -- was blind to all of them. The ledger did not merely
-# return nothing on this log, it REFUSED it: "no RESULT records, so there is
-# nothing to reconcile".
-#
-# `pgc_record` takes the DISPLAY whole, so the human output below is byte-for-byte
-# what it was. NOT lib.sh's own `check`: that composes its own display and would
-# drop the `: $got` suffix, which is the measured value rather than a label.
-#
-# `fail` is still set, so this suite's exit logic is untouched. Its verdict line
-# stays for the same reason: under `set -euo pipefail` a failing command aborts the
-# suite, and the verdict is what distinguishes finished from stopped.
-#
-# The `checks run:` line is READ BY NOTHING YET. The matrix gates reconciliation on
-# the ACCOUNTING line via `pgc_log_shows_accounting`, and this suite emits none
-# because it emits no accounting line. The line is still correct and wanted --
-# it is the total the records reconcile against -- so the remaining step is a gate
-# flip rather than new work (@jdatcmd, #969 review).
-check() {
-	local name="$1" got="$2" want="$3"
-	if [ "$got" = "$want" ]; then
-		pgc_record PASS "$name" "PASS  $name: $got"
-	else
-		pgc_record FAIL "$name" "FAIL  $name: got [$got] want [$want]"
-		fail=1
-	fi
-}
+# lib.sh's `check` records every result (#965), and since #1341 this suite uses
+# it rather than a local wrapper around pgc_record. The local one printed the
+# measured value on a PASS too. lib.sh's prints it on a FAIL, which is when it is
+# read, and a pytest twin's grading can read the names, which it could not before.
+# pgc_summary at the end prints the accounting line the matrix reconciles.
 
 # assert an EXPLAIN plan contains / omits text
 assert_plan() {
@@ -132,11 +107,10 @@ assert_plan() {
 	local plan
 	plan="$(run_pg "$PSQL -c \"$sql\"")"
 	if grep -q "$want" <<<"$plan" && { [ -z "$notwant" ] || ! grep -q "$notwant" <<<"$plan"; }; then
-		pgc_record PASS "$name" "PASS  $name"
+		check "$name" "matches" "matches"
 	else
-		pgc_record FAIL "$name" "FAIL  $name: plan was:
+		pgc_fail "$name" "plan was:
 $(echo "$plan" | sed 's/^/        /')"
-		fail=1
 	fi
 }
 
@@ -399,14 +373,5 @@ assert_plan "disable custom scan -> seq scan" \
 check "seq-scan fallback returns correct rows" \
 	"$(q "SET pgcolumnar.enable_custom_scan=off; SELECT count(*) FROM t WHERE a BETWEEN 12000 AND 12100;")" "101"
 
-echo
-if [ "$fail" = "0" ]; then
-	echo "checks run: $PGC_CHECKS"
-	echo "PHASE 5 TEST PASSED"
-else
-	echo "checks run: $PGC_CHECKS"
-	echo "PHASE 5 TEST FAILED"
-	echo "---- server log tail ----"
-	run_pg "tail -60 '$LOGFILE'" || true
-fi
-exit $fail
+[ "$PGC_FAILED" = 0 ] || { echo "---- server log tail ----"; run_pg "tail -60 '$LOGFILE'" || true; }
+pgc_summary
diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md
index 0c8cd1f85..9212ee1a5 100644
--- a/test/pytest/TESTS.md
+++ b/test/pytest/TESTS.md
@@ -182,6 +182,12 @@ behaviour, the source of that number is named.
 - [134. test_unique_conc.py: concurrent inserters of one unique key serialize](#134-test_unique_concpy-concurrent-inserters-of-one-unique-key-serialize)
 - [135. test_update_conc.py: two writers of one row serialize](#135-test_update_concpy-two-writers-of-one-row-serialize)
 - [136. test_recovery.py: committed writes survive a crash, uncommitted ones leave nothing](#136-test_recoverypy-committed-writes-survive-a-crash-uncommitted-ones-leave-nothing)
+- [137. test_native_backend_crash.py: one backend dies and committed data is intact](#137-test_native_backend_crashpy-one-backend-dies-and-committed-data-is-intact)
+- [138. test_corruption.py: a corrupt catalog row is refused cleanly and no backend dies](#138-test_corruptionpy-a-corrupt-catalog-row-is-refused-cleanly-and-no-backend-dies)
+- [139. test_inheritance.py: a partitioned columnar parent is refused, a columnar matview is not](#139-test_inheritancepy-a-partitioned-columnar-parent-is-refused-a-columnar-matview-is-not)
+- [140. test_parquet_import.py: a partitioned columnar parent is refused by import_parquet](#140-test_parquet_importpy-a-partitioned-columnar-parent-is-refused-by-import_parquet)
+- [141. test_phase5.py: set_options and reset_options agree](#141-test_phase5py-set_options-and-reset_options-agree)
+- [142. test_drop_cleanup.py: a columnar matview's options row goes with its DROP](#142-test_drop_cleanuppy-a-columnar-matviews-options-row-goes-with-its-drop)
 
 ## 1. How to read a test in here
 
@@ -7296,3 +7302,78 @@ A private cluster, stopped with `-m immediate` for each crash. Each crash is pro
 | `test_the_pair_through_three_crashes` | rows committed after a checkpoint survive; an open transaction's checkpointed INSERT leaves no row; committed deletes survive; each against the heap pair |
 | `test_a_cleared_visibility_bit_stays_cleared` | a bit set and checkpointed, then cleared by a DELETE, stays cleared after the crash; a clean block's bit stays set; the rows match heap |
 | `test_a_projection_survives_a_crash` | a projection copy committed after the checkpoint replays, and reproduces the base's rows and a scan through it |
+
+## 137. test_native_backend_crash.py: one backend dies and committed data is intact
+
+The pytest twin of `test/native_backend_crash.sh` (#217, ported for #1341), every check by the shell's name. A SIGSEGV in one backend makes the postmaster drop every session and reinitialize, with crash recovery, while it stays up itself. Committed columnar rows must then match a heap mirror, and an INSERT killed before its commit must leave no row and no row group.
+
+The kill lands mid-write, once the INSERT has written pages to the relation and before it reaches its trailing `pg_sleep`, and after every stripe has flushed, while it sleeps; both are asserted. A sentinel session opened before each kill must be dropped by the reinit, which proves the crash happened. A private cluster, since the reinit drops every session on the server.
+
+### Every test
+
+| test | what it holds |
+| --- | --- |
+| `test_a_kill_mid_write` | a 3,000,000-row INSERT killed after it has grown the relation: the cluster reinitialized, the postmaster kept its pid, and c holds exactly its committed rows and row groups |
+| `test_a_kill_after_the_stripes_flushed` | the same for a 100,000-row INSERT killed in its trailing `pg_sleep` |
+| `test_a_kill_with_a_second_reader` | the mid-write kill with a second parked reader, which the reinit drops too |
+
+## 138. test_corruption.py: a corrupt catalog row is refused cleanly and no backend dies
+
+The pytest twin of `test/corruption.sh` (ported for #1341), every check by the shell's name. The reader trusts the lengths, counts, offsets and descriptors the catalog stores. Each test corrupts one field on a fresh table and asserts that a read raises XX001, or, for a field the reader tolerates, that it does not crash.
+
+A crash is read from the server log, counted from the test's start, since a client error cannot tell a refusal from a crash and a liveness probe passes once crash recovery has finished. Each corrupting UPDATE must change a catalog row. The bloom test also looks up a value the table holds, since an absent value answers 0 even when a corrupt filter wrongly says "absent". A private cluster, since a crash drops every session.
+
+### Every test
+
+| test | what it holds |
+| --- | --- |
+| `test_a_corrupt_byte_length` | `row_group.byte_length` inflated: the read raises XX001, and no backend died |
+| `test_a_corrupt_row_count` | `row_group.row_count` inflated: XX001, no crash |
+| `test_a_corrupt_page_length` | `column_chunk.page_length` inflated: XX001, no crash |
+| `test_a_corrupt_value_count_degrades_safely` | `column_chunk.value_count` inflated: the read may answer or refuse, and no backend died |
+| `test_a_corrupt_encoding_descriptor` | a descriptor of `\xffffffff`: XX001, no crash |
+| `test_a_negative_column_index` | `column_index = -1`: no crash |
+| `test_a_corrupt_bloom_header_is_no_filter` | a huge nbits over a short buffer: an absent value is 0 rows, a held value is still found, no crash |
+
+## 139. test_inheritance.py: a partitioned columnar parent is refused, a columnar matview is not
+
+Part of the pytest twin of `test/inheritance.sh` (#1259, ported for #1341): its #1259 arms only, by the shell's names, and the pair is declared INCOMPLETE. From PostgreSQL 17 a partitioned table may carry relam = pgcolumnar and has no storage. `get_storage_id` and `add_projection` on it must be refused with 42809, not any error: with the fix reverted a release build fails with 58P01 opening relfilenode 0. Liveness is read from the server log. PostgreSQL 15 and 16 cannot build the fixture, so that test declines there.
+
+### Every test
+
+| test | what it holds |
+| --- | --- |
+| `test_a_partitioned_parent_is_refused_and_its_leaf_is_not` | the parent is refused with 42809 by both calls and no backend crashed; its leaf is accepted by both |
+| `test_a_columnar_materialized_view_is_still_columnar` | a columnar matview is still recognised and returns its rows |
+
+## 140. test_parquet_import.py: a partitioned columnar parent is refused by import_parquet
+
+Part of the pytest twin of `test/parquet_import.sh` (#1263, ported for #1341): its #1263 arms only, by the shell's names, and the pair is declared INCOMPLETE. Handed a partitioned parent, import_parquet used to SIGSEGV. The file is exported first, so the import reaches the relation; a leaf and a heap target must still import; the parent must be refused with 42809, and no backend may crash, read from the server log. It declines on PostgreSQL 15 and 16.
+
+### Every test
+
+| test | what it holds |
+| --- | --- |
+| `test_a_partitioned_parent_is_refused` | the file exists, a leaf and a heap target import 50 rows each, and the parent is refused with 42809 and no crash |
+
+## 141. test_phase5.py: set_options and reset_options agree
+
+Part of the pytest twin of `test/phase5.sh` (#1265, ported for #1341): its #1265 arms only, by the shell's names, and the pair is declared INCOMPLETE. reset_options had no guard, so it reported success where set_options refused. The property is parity on a columnar table, a heap table, a partitioned parent and a columnar matview, with an ordinary columnar table as the control both must accept, and the settled verdicts named: both accept the matview and refuse the partitioned parent.
+
+### Every test
+
+| test | what it holds |
+| --- | --- |
+| `test_set_and_reset_options_agree` | the four fixtures' kinds, the control, parity on each, and the matview and partitioned-parent verdicts |
+
+## 142. test_drop_cleanup.py: a columnar matview's options row goes with its DROP
+
+Part of the pytest twin of `test/drop_cleanup.sh` (#1265, ported for #1341): its #1265 arms only, by the shell's names, and the pair is declared INCOMPLETE. The drop hook skipped anything that was not an ordinary table, so a matview's options row outlived it. The row is planted directly, so the hook is the only subject. REFRESH must keep the row, and an unrelated table's row is the control against a hook that deletes every row.
+
+### Every test
+
+| test | what it holds |
+| --- | --- |
+| `test_a_columnar_matview_takes_its_options_row_with_it` | a planted options row is gone after DROP MATERIALIZED VIEW |
+| `test_refresh_keeps_a_matviews_options_row` | a REFRESH that rewrote the matview leaves its set_options row in place |
+| `test_an_unrelated_options_row_survives` | another table's row survives a matview's DROP, and an ordinary columnar table still takes its own |
diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt
index 76ea8a89f..a6a77ccba 100644
--- a/test/pytest/expected_tests.txt
+++ b/test/pytest/expected_tests.txt
@@ -741,4 +741,10 @@ guard_tests 438
 # tests, by collection over the cluster files on this tree.
 # cluster_tests 641 -> 643 for #1447's two tests in test_join_runtime_filter.py, by
 # collection over the cluster files on this tree.
-cluster_tests 643
+# cluster_tests 643 -> 646 for #1341's test_native_backend_crash.py, the twin of
+# native_backend_crash.sh: three tests, by collection over the cluster files on this tree.
+# cluster_tests 646 -> 653 for #1341's test_corruption.py, the twin of corruption.sh:
+# seven tests, by collection over the cluster files on this tree.
+# cluster_tests 653 -> 660 for #1341 item 29's partial twins: test_inheritance.py (2),
+# test_parquet_import.py (1), test_phase5.py (1) and test_drop_cleanup.py (3), by collection.
+cluster_tests 660
diff --git a/test/pytest/expected_unrunnable.txt b/test/pytest/expected_unrunnable.txt
index 731240aae..135205390 100644
--- a/test/pytest/expected_unrunnable.txt
+++ b/test/pytest/expected_unrunnable.txt
@@ -71,3 +71,10 @@
 17 test_temporal.py::test_for_portion_of
 18 test_temporal.py::test_for_portion_of
 19 test_temporal.py::test_for_portion_of
+
+# PostgreSQL 15 and 16: a partitioned table cannot carry an access method before 17,
+# so the #1259 and #1263 partitioned-parent fixtures cannot exist (#1341 item 29).
+15 test_inheritance.py::test_a_partitioned_parent_is_refused_and_its_leaf_is_not
+16 test_inheritance.py::test_a_partitioned_parent_is_refused_and_its_leaf_is_not
+15 test_parquet_import.py::test_a_partitioned_parent_is_refused
+16 test_parquet_import.py::test_a_partitioned_parent_is_refused
diff --git a/test/pytest/test_compare_to_bash.py b/test/pytest/test_compare_to_bash.py
index 15b5b7cc7..4140fbf32 100644
--- a/test/pytest/test_compare_to_bash.py
+++ b/test/pytest/test_compare_to_bash.py
@@ -101,6 +101,7 @@
             "catalog_scan_index",
             "collation_change",
             "column_extent",
+            "corruption",
             "debug_hook_privilege",
             "differential",
             "encode_post_codec",
@@ -116,6 +117,7 @@
             "latest_reading",
             "maintenance_crash",
             "maintenance_rollback",
+            "native_backend_crash",
             "native_chunk_length_bound",
             "native_delete_vector_index",
             "native_delete_visibility_paths",
@@ -177,6 +179,20 @@
 # about the PORT rather than a licence: the standing arm does not grade it, so the
 # reason is the only thing standing between a declared gap and a forgotten one.
 INCOMPLETE = {
+    # #1341 item 29 ported the arms four fixes added with shell arms only. Each
+    # pair below holds that fix's arms and nothing else of its shell suite.
+    "inheritance": "the #1259 arms (a partitioned columnar parent is refused with "
+                   "42809 and nothing crashes, its leaf and a columnar matview are "
+                   "accepted) are ported; the inheritance-tree arms are not",
+    "parquet_import": "the #1263 arms (a partitioned columnar parent is refused with "
+                      "42809 and nothing crashes, a leaf and a heap target import) "
+                      "are ported; the page-version, type and round-trip arms are not",
+    "phase5": "the #1265 arms (set_options and reset_options agree on every relkind, "
+              "accept a columnar matview and refuse a partitioned parent) are "
+              "ported; the projection, pushdown, compression and vacuum arms are not",
+    "drop_cleanup": "the #1265 arms (a columnar matview's options row goes with its "
+                    "DROP and survives its REFRESH, with an unrelated row as the "
+                    "control) are ported; the storage and catalog cleanup arms are not",
     # #1433 ported the arms it added: cidr, varchar and a domain over each prune
     # as inet and text do, with heap's answer first. The rest of the shell suite
     # (range, equality, bigint, the #477 and #483 arms, the partitioned table) is
diff --git a/test/pytest/test_corruption.py b/test/pytest/test_corruption.py
new file mode 100644
index 000000000..63cae400c
--- /dev/null
+++ b/test/pytest/test_corruption.py
@@ -0,0 +1,179 @@
+"""A corrupt pgcolumnar catalog row is refused cleanly, or degrades safely, and no
+backend dies (#1341). The pytest twin of `test/corruption.sh`, every check by the
+shell's name.
+
+The reader trusts the lengths, counts, offsets and descriptors the catalog stores.
+Each scenario corrupts one of those fields on a fresh table and asserts that a read
+raises XX001 (data_corrupted), or for the fields the reader tolerates, that it does
+not crash. A crash is read from the server log ("terminated by signal"), counted
+from the scenario's start. A client error alone cannot tell a refusal from a crash,
+and a liveness probe passes once crash recovery has finished. Each corrupting UPDATE
+must change a catalog row, or its scenario would pass on a table nobody corrupted.
+
+A PRIVATE CLUSTER, because a crash would drop every session on the server.
+"""
+import pathlib
+import shutil
+import tempfile
+
+import psycopg
+import pytest
+
+from pgc_vacuity import query_error
+
+NO_SUCH_STORAGE = False     # True points every corruption at a storage id that does not exist
+
+
+@pytest.fixture(scope="module")
+def cor(pgc_cluster):
+    """-> the private cluster, started, with the extension created."""
+    from pgc_cluster import Cluster, pick_port
+
+    work = pathlib.Path(tempfile.mkdtemp(prefix="pgc-corrupt-"))
+    work.chmod(0o777)
+    datadir = work / "data"
+    datadir.mkdir()
+    cluster = Cluster(pgc_cluster.pg_config, "corrupt", datadir, pick_port(16))
+    try:
+        cluster.initdb()
+        cluster.start()
+        with psycopg.connect(cluster.dsn(), autocommit=True) as c:
+            c.execute("CREATE EXTENSION IF NOT EXISTS pgcolumnar")
+        yield cluster
+    finally:
+        cluster.stop()
+        shutil.rmtree(work, ignore_errors=True)
+
+
+def _one(cluster, sql):
+    """-> the first column of the first row, or a distinct failure sentinel."""
+    try:
+        with psycopg.connect(cluster.dsn(), autocommit=True, connect_timeout=5) as c:
+            return c.execute(sql).fetchone()[0]
+    except psycopg.Error as exc:
+        return query_error(str(exc).splitlines()[0])
+
+
+def _crashes(cluster):
+    try:
+        return (cluster.datadir / "server.log").read_text(errors="replace").count("terminated by signal")
+    except OSError:
+        return 0
+
+
+class _Scenario:
+    """A fresh table c, the crash count at its start, and the probes the shell runs."""
+
+    def __init__(self, cluster):
+        self.cluster = cluster
+        self.crashes = _crashes(cluster)
+        with psycopg.connect(cluster.dsn(), autocommit=True) as c:
+            c.execute("DROP TABLE IF EXISTS c")
+            c.execute("CREATE TABLE c (id int, r bigint, t text) USING pgcolumnar")
+            # r random: poorly compressible; id monotonic; t low-cardinality: dictionary
+            c.execute("INSERT INTO c SELECT g, (random() * 9e18)::bigint, 'v' || (g % 5) "
+                      "FROM generate_series(1, 20000) g")
+        self.sid = 0 if NO_SUCH_STORAGE else _one(cluster, "SELECT pgcolumnar.get_storage_id('c')")
+
+    def corrupt(self, update):
+        """`update`, with {sid} filled. -> whether it changed a catalog row, as text."""
+        return str(_one(self.cluster, f"WITH u AS ({update.format(sid=self.sid)} RETURNING 1) "
+                                      "SELECT count(*) > 0 FROM u"))
+
+    def refused(self, sql):
+        """-> the SQLSTATE sql raised, or what happened instead. A crash ends the
+        connection with no SQLSTATE at all."""
+        try:
+            with psycopg.connect(self.cluster.dsn(), autocommit=True) as c:
+                row = c.execute(sql).fetchone()
+            return f"no error ({row})"
+        except psycopg.Error as exc:
+            return exc.sqlstate or f"no error ({str(exc).splitlines()[0]})"
+
+    def read(self, sql):
+        """Run sql for its effect on the backend; its answer is not the subject."""
+        try:
+            with psycopg.connect(self.cluster.dsn(), autocommit=True) as c:
+                c.execute(sql).fetchall()
+        except psycopg.Error:
+            pass
+
+    def alive(self):
+        crashed = _crashes(self.cluster) - self.crashes
+        return "yes" if _one(self.cluster, "SELECT 1") == 1 and crashed == 0 else f"no ({crashed} backend crashes)"
+
+
+def test_a_corrupt_byte_length(cor, expect):
+    s = _Scenario(cor)
+    field = "byte_length"
+    expect.text(s.corrupt("UPDATE pgcolumnar.row_group SET byte_length = byte_length + 123456 "
+                          "WHERE storage_id = {sid}"), "True",
+                f"premise: the {field} corruption changed a catalog row")
+    expect.text(s.refused("SELECT count(*), sum(r) FROM c"), "XX001", "byte_length corruption errors")
+    expect.text(s.alive(), "yes", "alive after byte_length")
+
+
+def test_a_corrupt_row_count(cor, expect):
+    s = _Scenario(cor)
+    field = "row_count"
+    expect.text(s.corrupt("UPDATE pgcolumnar.row_group SET row_count = row_count + 100000 "
+                          "WHERE storage_id = {sid}"), "True",
+                f"premise: the {field} corruption changed a catalog row")
+    expect.text(s.refused("SELECT sum(r) FROM c"), "XX001", "row_count corruption errors")
+    expect.text(s.alive(), "yes", "alive after row_count")
+
+
+def test_a_corrupt_page_length(cor, expect):
+    s = _Scenario(cor)
+    field = "page_length"
+    expect.text(s.corrupt("UPDATE pgcolumnar.column_chunk SET page_length = page_length + 123456 "
+                          "WHERE storage_id = {sid}"), "True",
+                f"premise: the {field} corruption changed a catalog row")
+    expect.text(s.refused("SELECT sum(r) FROM c"), "XX001", "page_length corruption errors")
+    expect.text(s.alive(), "yes", "alive after page_length")
+
+
+def test_a_corrupt_value_count_degrades_safely(cor, expect):
+    s = _Scenario(cor)
+    field = "value_count"
+    expect.text(s.corrupt("UPDATE pgcolumnar.column_chunk SET value_count = value_count + 100000 "
+                          "WHERE storage_id = {sid}"), "True",
+                f"premise: the {field} corruption changed a catalog row")
+    s.read("SELECT sum(r) FROM c")
+    expect.text(s.alive(), "yes", "alive after value_count")
+
+
+def test_a_corrupt_encoding_descriptor(cor, expect):
+    s = _Scenario(cor)
+    field = "encoding_descriptor"
+    expect.text(s.corrupt("UPDATE pgcolumnar.column_chunk SET encoding_descriptor = '\\xffffffff'::bytea "
+                          "WHERE storage_id = {sid} AND column_index = 1"), "True",
+                f"premise: the {field} corruption changed a catalog row")
+    expect.text(s.refused("SELECT * FROM c"), "XX001", "bad descriptor errors")
+    expect.text(s.alive(), "yes", "alive after bad descriptor")
+
+
+def test_a_negative_column_index(cor, expect):
+    s = _Scenario(cor)
+    field = "column_index"
+    expect.text(s.corrupt("UPDATE pgcolumnar.column_chunk SET column_index = -1 "
+                          "WHERE storage_id = {sid} AND column_index = 1"), "True",
+                f"premise: the {field} corruption changed a catalog row")
+    s.read("SELECT count(*) FROM c")
+    expect.text(s.alive(), "yes", "alive after column_index=-1")
+
+
+def test_a_corrupt_bloom_header_is_no_filter(cor, expect):
+    """A huge nbits over a short buffer. Looking up an absent value answers 0 whether
+    the filter is ignored or wrongly says "absent", so a value the table holds is
+    looked up too: a false negative is the bloom's one way to lose rows."""
+    s = _Scenario(cor)
+    hit = _one(cor, "SELECT r FROM c WHERE id = 7777")
+    field = "bloom filter"
+    expect.text(s.corrupt("UPDATE pgcolumnar.bloom SET filter = '\\x0000004006'::bytea "
+                          "WHERE storage_id = {sid}"), "True",
+                f"premise: the {field} corruption changed a catalog row")
+    expect.num(_one(cor, "SELECT count(*) FROM c WHERE r = -999"), 0, "corrupt bloom equality still correct")
+    expect.num(_one(cor, f"SELECT count(*) FROM c WHERE r = {hit}"), 1,
+               "corrupt bloom: a value the table holds is still found")
+    expect.text(s.alive(), "yes", "alive after bloom corruption")
diff --git a/test/pytest/test_drop_cleanup.py b/test/pytest/test_drop_cleanup.py
new file mode 100644
index 000000000..90487df6b
--- /dev/null
+++ b/test/pytest/test_drop_cleanup.py
@@ -0,0 +1,72 @@
+"""Dropping a columnar materialized view takes its options row with it, and REFRESH
+keeps it (#1265, #1341). Part of the pytest twin of `test/drop_cleanup.sh`: these
+arms only, by the shell's names, and the pair is declared INCOMPLETE.
+
+The drop hook returned before it examined anything that was not an ordinary table, so
+an options row describing a columnar matview outlived the relation. The row is planted
+directly, not through set_options: the subject is the drop hook, and set_options has a
+relkind guard of its own, which while it refused a matview would leave no row and let
+"0 rows afterwards" pass on a hook that does nothing. Widening the hook means a
+refresh that dropped the relation would discard the user's options, so REFRESH is
+asserted too, and an unrelated table's row is the control against a hook that deletes
+every row.
+"""
+
+
+def _one(conn, sql):
+    with conn.cursor() as cur:
+        cur.execute(sql)
+        return cur.fetchone()[0]
+
+
+def _run(conn, *sqls):
+    with conn.cursor() as cur:
+        for sql in sqls:
+            cur.execute(sql)
+
+
+def test_a_columnar_matview_takes_its_options_row_with_it(pgc_conn, expect):
+    conn = pgc_conn
+    _run(conn, "CREATE TABLE dcl_mv_src (a int)", "INSERT INTO dcl_mv_src VALUES (1), (2)",
+         "CREATE MATERIALIZED VIEW dcl_mv USING pgcolumnar AS SELECT a FROM dcl_mv_src")
+    expect.text(_one(conn, "SELECT c.relkind::text || am.amname::text FROM pg_class c JOIN pg_am am "
+                           "ON am.oid = c.relam WHERE c.oid = 'dcl_mv'::regclass"), "mpgcolumnar",
+                "premise: the matview is relkind m on the columnar access method")
+    _run(conn, "INSERT INTO pgcolumnar.options (regclass, stripe_row_limit) VALUES ('dcl_mv'::regclass, 123)")
+    expect.num(_one(conn, "SELECT count(*) FROM pgcolumnar.options WHERE regclass = 'dcl_mv'::regclass"), 1,
+               "premise: the matview carries an options row for the hook to clear")
+    oid = _one(conn, "SELECT 'dcl_mv'::regclass::oid")
+    _run(conn, "DROP MATERIALIZED VIEW dcl_mv")
+    expect.num(_one(conn, f"SELECT count(*) FROM pgcolumnar.options WHERE regclass = {oid}"), 0,
+               "dropping a columnar matview takes its options row with it")
+
+
+def test_refresh_keeps_a_matviews_options_row(pgc_conn, expect):
+    conn = pgc_conn
+    _run(conn, "CREATE TABLE dcl_mv_ref_src (a int)", "INSERT INTO dcl_mv_ref_src VALUES (1), (2)",
+         "CREATE MATERIALIZED VIEW dcl_mv_ref USING pgcolumnar AS SELECT a FROM dcl_mv_ref_src",
+         "SELECT pgcolumnar.set_options('dcl_mv_ref', stripe_row_limit => 100000)")
+    expect.text(_one(conn, "SELECT stripe_row_limit::text FROM pgcolumnar.options "
+                           "WHERE regclass = 'dcl_mv_ref'::regclass"), "100000",
+                "premise: the matview holds the options row set through set_options")
+    _run(conn, "INSERT INTO dcl_mv_ref_src VALUES (3)", "REFRESH MATERIALIZED VIEW dcl_mv_ref")
+    expect.num(_one(conn, "SELECT count(*) FROM dcl_mv_ref"), 3, "premise: the refresh actually rewrote the matview")
+    expect.text(_one(conn, "SELECT stripe_row_limit::text FROM pgcolumnar.options "
+                           "WHERE regclass = 'dcl_mv_ref'::regclass"), "100000",
+                "REFRESH keeps the matview's options row")
+
+
+def test_an_unrelated_options_row_survives(pgc_conn, expect):
+    """Without it a hook that deletes every options row passes the arms above."""
+    conn = pgc_conn
+    _run(conn, "CREATE TABLE dcl_opt_keep (a int) USING pgcolumnar",
+         "INSERT INTO pgcolumnar.options (regclass, stripe_row_limit) VALUES ('dcl_opt_keep'::regclass, 123)",
+         "CREATE TABLE dcl_other_src (a int)",
+         "CREATE MATERIALIZED VIEW dcl_other USING pgcolumnar AS SELECT a FROM dcl_other_src",
+         "DROP MATERIALIZED VIEW dcl_other")
+    expect.num(_one(conn, "SELECT count(*) FROM pgcolumnar.options WHERE regclass = 'dcl_opt_keep'::regclass"), 1,
+               "control: an unrelated options row survives that drop")
+    oid = _one(conn, "SELECT 'dcl_opt_keep'::regclass::oid")
+    _run(conn, "DROP TABLE dcl_opt_keep")
+    expect.num(_one(conn, f"SELECT count(*) FROM pgcolumnar.options WHERE regclass = {oid}"), 0,
+               "control: an ordinary columnar table still takes its own options row")
diff --git a/test/pytest/test_inheritance.py b/test/pytest/test_inheritance.py
new file mode 100644
index 000000000..2f2379f3b
--- /dev/null
+++ b/test/pytest/test_inheritance.py
@@ -0,0 +1,100 @@
+"""A partitioned parent carrying the columnar access method has no storage, and is
+refused rather than read; a columnar materialized view is still columnar (#1259,
+#1341). Part of the pytest twin of `test/inheritance.sh`: these arms only, by the
+shell's names, and the pair is declared INCOMPLETE.
+
+From PostgreSQL 17 a partitioned table may carry relam = pgcolumnar as the default for
+its future partitions. It has no storage of its own, relfilenode 0, and
+PgColumnarIsColumnarRelation used to answer "yes" from relam alone. Every caller that
+then read storage handed smgropen a zero relfilenumber: an abort on an assert build,
+and on a release build a failure to open ".../0/0", 58P01. So a refusal here is
+42809 (wrong_object_type), not any error, and liveness is read from the server log,
+since a crashed cluster restarts and a later connection succeeds.
+"""
+import psycopg
+
+
+def _crashes(pgc_cluster):
+    try:
+        return (pgc_cluster.datadir / "server.log").read_text(errors="replace").count("terminated by signal")
+    except OSError:
+        return 0
+
+
+def _sqlstate(conn, sql):
+    """-> 'accepted', or the SQLSTATE the statement raised."""
+    try:
+        with conn.cursor() as cur:
+            cur.execute(sql)
+        return "accepted"
+    except psycopg.Error as exc:
+        return exc.sqlstate or f"no SQLSTATE ({str(exc).splitlines()[0]})"
+
+
+def _one(conn, sql):
+    with conn.cursor() as cur:
+        cur.execute(sql)
+        return cur.fetchone()[0]
+
+
+def test_a_partitioned_parent_is_refused_and_its_leaf_is_not(pgc_conn, pgc_cluster, expect):
+    conn = pgc_conn
+    major = int(_one(conn, "SHOW server_version_num")) // 10000
+    if major < 17:
+        # PostgreSQL 15 and 16 refuse PARTITION BY ... USING outright, so the fixture
+        # cannot exist and every "is refused" arm would pass on a missing relation.
+        expect.cannot_run("UNSUPPORTED_MAJOR",
+                          f"a partitioned table cannot carry an access method before PostgreSQL 17; this server is {major}",
+                          name="get_storage_id on the parent is refused rather than reading storage")
+        return
+    with conn.cursor() as cur:
+        cur.execute("CREATE TABLE inh_pp (a int, b text) PARTITION BY RANGE (a) USING pgcolumnar")
+        cur.execute("CREATE TABLE inh_pp1 PARTITION OF inh_pp FOR VALUES FROM (0) TO (100)")
+        cur.execute("INSERT INTO inh_pp SELECT g, 'x' || g FROM generate_series(0, 99) g")
+    expect.text(_one(conn, "SELECT am.amname::text FROM pg_class c JOIN pg_am am ON am.oid = c.relam "
+                           "WHERE c.oid = 'inh_pp'::regclass"), "pgcolumnar",
+                "premise: the parent carries the columnar access method")
+    expect.text(_one(conn, "SELECT relkind::text FROM pg_class WHERE oid = 'inh_pp'::regclass"), "p",
+                "premise: and it is partitioned")
+    expect.num(_one(conn, "SELECT relfilenode::int FROM pg_class WHERE oid = 'inh_pp'::regclass"), 0,
+               "premise: so it has no storage of its own")
+    expect.text(_one(conn, "SELECT relkind::text FROM pg_class WHERE oid = 'inh_pp1'::regclass"), "r",
+                "premise: while its leaf is an ordinary relation")
+
+    crashes = _crashes(pgc_cluster)
+
+    def up():
+        alive = _sqlstate(conn, "SELECT 1") == "accepted"
+        crashed = _crashes(pgc_cluster) - crashes
+        return "up" if alive and crashed == 0 else f"DOWN ({crashed} backend crashes)"
+
+    expect.text(_sqlstate(conn, "SELECT pgcolumnar.get_storage_id('inh_pp'::regclass)"), "42809",
+                "get_storage_id on the parent is refused rather than reading storage")
+    expect.text(up(), "up", "and the cluster is still up after it")
+    expect.text(_sqlstate(conn, "SELECT pgcolumnar.add_projection('inh_pp'::regclass, 'p1', ARRAY['a'])"),
+                "42809", "add_projection on the parent is refused rather than aborting the backend")
+    expect.text(up(), "up", "and the cluster is still up after that too")
+
+    # Without these a function that refuses everything passes the arms above.
+    expect.text(_sqlstate(conn, "SELECT pgcolumnar.get_storage_id('inh_pp1'::regclass)"), "accepted",
+                "control: get_storage_id on the LEAF still returns an id")
+    expect.text(_sqlstate(conn, "SELECT pgcolumnar.add_projection('inh_pp1'::regclass, 'p1', ARRAY['a'])"),
+                "accepted", "control: add_projection on the LEAF still works")
+
+
+def test_a_columnar_materialized_view_is_still_columnar(pgc_conn, expect):
+    """The arm against the obvious wrong fix: relkind == RELKIND_RELATION would exclude
+    a matview too, and a matview can be columnar. RELKIND_HAS_STORAGE admits it."""
+    conn = pgc_conn
+    with conn.cursor() as cur:
+        cur.execute("CREATE TABLE inh_src (a int)")
+        cur.execute("INSERT INTO inh_src SELECT generate_series(1, 10)")
+        cur.execute("CREATE MATERIALIZED VIEW inh_mv USING pgcolumnar AS SELECT * FROM inh_src")
+    expect.text(_one(conn, "SELECT relkind::text FROM pg_class WHERE oid = 'inh_mv'::regclass"), "m",
+                "premise: a materialized view is relkind m")
+    expect.text(_one(conn, "SELECT am.amname::text FROM pg_class c JOIN pg_am am ON am.oid = c.relam "
+                           "WHERE c.oid = 'inh_mv'::regclass"), "pgcolumnar",
+                "premise: and it carries the columnar access method")
+    expect.text(_sqlstate(conn, "SELECT pgcolumnar.get_storage_id('inh_mv'::regclass)"), "accepted",
+                "a columnar materialized view is still recognised as columnar")
+    expect.num(_one(conn, "SELECT count(*) FROM inh_mv"), 10, "and it still returns its rows")
diff --git a/test/pytest/test_native_backend_crash.py b/test/pytest/test_native_backend_crash.py
new file mode 100644
index 000000000..1c7b8aed4
--- /dev/null
+++ b/test/pytest/test_native_backend_crash.py
@@ -0,0 +1,235 @@
+"""One backend dies, the postmaster lives, and committed columnar data is intact
+(#217, #1341). The pytest twin of `test/native_backend_crash.sh`, every check by the
+shell's name.
+
+A SIGSEGV in one backend makes the postmaster drop every session and reinitialize
+shared memory, with crash recovery, while the postmaster itself stays up. Columnar
+data goes through the buffer manager and WAL, so committed rows must match a heap
+mirror afterwards, and an INSERT killed before its commit must leave no row and no
+row group. The kill lands at two points: mid-write, once the INSERT has written
+pages to the relation and before it reaches its trailing pg_sleep, and after every
+stripe has flushed, while the backend sleeps. Both are asserted, not assumed.
+
+The write is uncommitted either way, so "no rows survived" cannot tell a crash from
+a clean abort. A sentinel session, opened before each kill, must be dropped by the
+reinit: that is what proves the crash happened.
+
+A PRIVATE CLUSTER, because the reinit drops every session on the server.
+"""
+import os
+import pathlib
+import shutil
+import signal
+import tempfile
+import threading
+import time
+
+import psycopg
+import pytest
+
+from pgc_vacuity import query_error
+
+BIG_STRIPE = None   # a stripe_row_limit for the inserting session; None keeps the default
+
+
+@pytest.fixture(scope="module")
+def nbc(pgc_cluster):
+    """-> the private cluster, started, with the extension and the h/c pair at 5000 rows."""
+    from pgc_cluster import Cluster, pick_port
+
+    work = pathlib.Path(tempfile.mkdtemp(prefix="pgc-nbcrash-"))
+    work.chmod(0o777)
+    datadir = work / "data"
+    datadir.mkdir()
+    cluster = Cluster(pgc_cluster.pg_config, "nbcrash", datadir, pick_port(14))
+    try:
+        cluster.initdb()
+        cluster.start()
+        with psycopg.connect(cluster.dsn(), autocommit=True) as c:
+            c.execute("CREATE EXTENSION IF NOT EXISTS pgcolumnar")
+            # the postmaster must reinitialize after a backend crash, not shut down
+            c.execute("ALTER SYSTEM SET restart_after_crash = on")
+            c.execute("SELECT pg_reload_conf()")
+            c.execute("CREATE TABLE h (id int, v text, x float8)")
+            c.execute("CREATE TABLE c (id int, v text, x float8) USING pgcolumnar")
+            c.execute("INSERT INTO h SELECT g, 'b' || g, g * 2.5 FROM generate_series(1, 5000) g")
+            c.execute("INSERT INTO c SELECT g, 'b' || g, g * 2.5 FROM generate_series(1, 5000) g")
+        yield cluster
+    finally:
+        cluster.stop()
+        shutil.rmtree(work, ignore_errors=True)
+
+
+def _one(cluster, sql):
+    """-> the first column of the first row, or a distinct failure sentinel."""
+    try:
+        with psycopg.connect(cluster.dsn(), autocommit=True, connect_timeout=5) as c:
+            return c.execute(sql).fetchone()[0]
+    except psycopg.Error as exc:
+        return query_error(str(exc).splitlines()[0])
+
+
+def _rows(cluster, sql):
+    try:
+        with psycopg.connect(cluster.dsn(), autocommit=True, connect_timeout=5) as c:
+            return c.execute(sql).fetchall()
+    except psycopg.Error as exc:
+        return query_error(str(exc).splitlines()[0])
+
+
+def _row_groups(cluster):
+    return _one(cluster, "SELECT count(*) FROM pgcolumnar.row_group r JOIN pgcolumnar.storage s "
+                         "ON s.storage_id = r.storage_id WHERE s.relation_oid = 'c'::regclass")
+
+
+def _postmaster_pid(cluster):
+    return (cluster.datadir / "postmaster.pid").read_text().splitlines()[0]
+
+
+class _Bystander:
+    """A session parked in pg_sleep on its own thread. -> .dropped once the server
+    ended it, which a plain exit of another backend never does."""
+
+    def __init__(self, cluster, seconds):
+        self.conn = psycopg.connect(cluster.dsn(), autocommit=True)
+        self.pid = self.conn.info.backend_pid
+        self.error = None
+        self.thread = threading.Thread(target=self._run, args=(seconds,), daemon=True)
+        self.thread.start()
+
+    def _run(self, seconds):
+        try:
+            self.conn.execute(f"SELECT pg_sleep({seconds})")
+        except psycopg.Error as exc:
+            self.error = exc
+
+    def dropped(self, within):
+        self.thread.join(within)
+        return not self.thread.is_alive() and self.error is not None
+
+    def close(self, cluster):
+        _end(cluster, self.conn, self.thread, self.pid)
+
+
+def _end(cluster, conn, thread, pid):
+    """Close a session whose statement may still run on its thread. A live one is
+    terminated first: closing a connection under a running statement crashes the
+    driver, and a session a failed check left behind holds its locks into the next
+    test."""
+    if thread.is_alive():
+        _one(cluster, f"SELECT pg_terminate_backend({pid})")
+        thread.join(10)
+    try:
+        conn.close()
+    except psycopg.Error:
+        pass
+
+
+def _wait_up(cluster):
+    """Twelve successful connections in a row, 0.25s apart: a 3s window a reinit's
+    brief accepting blip cannot fake. -> True once it holds."""
+    ok = 0
+    for _ in range(1200):
+        if _one(cluster, "SELECT 1") == 1:
+            ok += 1
+            if ok >= 12:
+                return True
+        else:
+            ok = 0
+        time.sleep(0.25)
+    return False
+
+
+def _crash_one_backend(cluster, expect, rows, marker, phase):
+    """An INSERT into c, held open by a trailing pg_sleep so it never commits, SIGSEGVed
+    in `phase` ('insert': mid-write, 'sleep': after every stripe flushed).
+    -> whether the sentinel was dropped by the reinit."""
+    sentinel = _Bystander(cluster, 120)
+    size0 = _one(cluster, "SELECT pg_relation_size('c')")
+    writer = psycopg.connect(cluster.dsn(), autocommit=True)
+    pid = writer.info.backend_pid
+
+    def write():
+        try:
+            if BIG_STRIPE:
+                writer.execute(f"SET pgcolumnar.stripe_row_limit = {BIG_STRIPE}")
+            writer.execute("BEGIN")
+            writer.execute(f"INSERT INTO c /* {marker} */ SELECT g, 'k' || g, g FROM generate_series(1, {rows}) g")
+            writer.execute("SELECT pg_sleep(60)")
+        except psycopg.Error:
+            pass
+    thread = threading.Thread(target=write, daemon=True)
+    thread.start()
+    try:
+        return _kill_when_ready(cluster, expect, marker, phase, pid, size0, thread, sentinel)
+    finally:
+        _end(cluster, writer, thread, pid)
+        sentinel.close(cluster)
+
+
+def _kill_when_ready(cluster, expect, marker, phase, pid, size0, thread, sentinel):
+    def wait_event():
+        return _one(cluster, f"SELECT coalesce(wait_event, '') FROM pg_stat_activity WHERE pid = {pid}")
+
+    for _ in range(600):
+        if phase == "sleep":
+            if wait_event() == "PgSleep":
+                break
+        elif _one(cluster, f"SELECT pg_relation_size('c') > {size0}") is True:
+            break
+        time.sleep(0.1)
+    expect.text(str(_one(cluster, f"SELECT pg_relation_size('c') > {size0}")), "True",
+                f"{marker}: the INSERT had written pages before the kill (the relation grew)")
+    we = wait_event()
+    if phase == "sleep":
+        expect.text(we, "PgSleep", f"{marker}: kill lands after the stripes flushed (backend in pg_sleep)")
+    else:
+        expect.text("yes" if we != "PgSleep" else "no", "yes",
+                    f"{marker}: kill lands mid-write (backend not yet in pg_sleep)")
+
+    os.kill(pid, signal.SIGSEGV)
+    thread.join(30)
+    _wait_up(cluster)
+    return sentinel.dropped(5)
+
+
+def _assert_intact(cluster, expect, label, dropped, base):
+    expect.text("yes" if dropped else "no", "yes",
+                f"{label}: the crash reinitialized the cluster (a live session was dropped)")
+    expect.text(_postmaster_pid(cluster), base["pm"], f"{label}: postmaster survived, same pid")
+    expect.num(_one(cluster, "SELECT 1"), 1, f"{label}: cluster accepts new connections")
+    expect.row_set(_rows(cluster, "SELECT id, v, x FROM c"), base["content"],
+                   f"{label}: committed content in c matches the heap mirror")
+    expect.row_set(_rows(cluster, "SELECT id, v, x FROM h"), base["mirror"],
+                   f"{label}: the mirror itself is unchanged")
+    expect.num(_one(cluster, "SELECT count(*) FROM c"), 5000, f"{label}: the killed insert left no visible rows")
+    expect.num(_row_groups(cluster), base["groups"], f"{label}: the killed insert left no partial row group")
+
+
+@pytest.fixture(scope="module")
+def base(nbc):
+    """The committed baseline, read once before any crash."""
+    return {"groups": _row_groups(nbc), "content": _rows(nbc, "SELECT id, v, x FROM c"),
+            "mirror": _rows(nbc, "SELECT id, v, x FROM h"), "pm": _postmaster_pid(nbc)}
+
+
+def test_a_kill_mid_write(nbc, base, expect):
+    dropped = _crash_one_backend(nbc, expect, 3000000, "crashA", "insert")
+    _assert_intact(nbc, expect, "mid-write", dropped, base)
+
+
+def test_a_kill_after_the_stripes_flushed(nbc, base, expect):
+    dropped = _crash_one_backend(nbc, expect, 100000, "crashB", "sleep")
+    _assert_intact(nbc, expect, "post-flush", dropped, base)
+
+
+def test_a_kill_with_a_second_reader(nbc, base, expect):
+    """The reinit drops every session, so a second parked reader goes too."""
+    reader = _Bystander(nbc, 90)
+    try:
+        dropped = _crash_one_backend(nbc, expect, 3000000, "crashC", "insert")
+        expect.text("yes" if reader.dropped(5) else "no", "yes",
+                    "a second concurrent reader was also dropped by the crash")
+    finally:
+        reader.close(nbc)
+    _assert_intact(nbc, expect, "with-reader", dropped, base)
diff --git a/test/pytest/test_parquet_import.py b/test/pytest/test_parquet_import.py
new file mode 100644
index 000000000..4e3787916
--- /dev/null
+++ b/test/pytest/test_parquet_import.py
@@ -0,0 +1,75 @@
+"""import_parquet refuses a partitioned columnar parent rather than segfaulting (#1263,
+#1341). Part of the pytest twin of `test/parquet_import.sh`: these arms only, by the
+shell's names, and the pair is declared INCOMPLETE.
+
+import_parquet never asked whether its target had storage. Handed a partitioned parent,
+relam pgcolumnar and relfilenode 0, it dereferenced its way to SIGSEGV, on a release
+build too. The file must exist before the call, or the import fails at the OPEN and
+never reaches the relation. A heap target must still import: the guard is about
+storage, not about being columnar. A refusal is 42809, and the crash is read from the
+server log, since a dropped connection is an error too and a later one succeeds.
+"""
+import psycopg
+
+
+def _crashes(pgc_cluster):
+    try:
+        return (pgc_cluster.datadir / "server.log").read_text(errors="replace").count("terminated by signal")
+    except OSError:
+        return 0
+
+
+def _one(conn, sql, params=None):
+    with conn.cursor() as cur:
+        cur.execute(sql, params)
+        return cur.fetchone()[0]
+
+
+def _sqlstate(conn, sql, params=None):
+    """-> 'accepted', or the SQLSTATE the statement raised."""
+    try:
+        with conn.cursor() as cur:
+            cur.execute(sql, params)
+        return "accepted"
+    except psycopg.Error as exc:
+        return exc.sqlstate or f"no SQLSTATE ({str(exc).splitlines()[0]})"
+
+
+def test_a_partitioned_parent_is_refused(pgc_conn, pgc_cluster, server_dir, expect):
+    conn = pgc_conn
+    major = int(_one(conn, "SHOW server_version_num")) // 10000
+    if major < 17:
+        expect.cannot_run("UNSUPPORTED_MAJOR",
+                          f"a partitioned table cannot carry an access method before PostgreSQL 17; this server is {major}",
+                          name="importing into the partitioned parent is refused")
+        return
+    with conn.cursor() as cur:
+        cur.execute("CREATE TABLE pq_parent (id int, v text) PARTITION BY RANGE (id)")
+        cur.execute("ALTER TABLE pq_parent SET ACCESS METHOD pgcolumnar")
+        cur.execute("CREATE TABLE pq_leaf PARTITION OF pq_parent FOR VALUES FROM (0) TO (100)")
+        cur.execute("INSERT INTO pq_parent SELECT g, 'r' || g FROM generate_series(0, 49) g")
+    expect.text(_one(conn, "SELECT c.relkind::text || am.amname::text FROM pg_class c "
+                           "JOIN pg_am am ON am.oid = c.relam WHERE c.oid = 'pq_parent'::regclass"),
+                "ppgcolumnar", "premise: the parent is partitioned and carries the columnar access method")
+    expect.num(_one(conn, "SELECT relfilenode::int FROM pg_class WHERE oid = 'pq_parent'::regclass"), 0,
+               "premise: so it has no storage of its own")
+
+    path = str(server_dir / "pq_leaf.parquet")
+    _sqlstate(conn, "SELECT pgcolumnar.export_parquet('pq_leaf', %s)", (path,))
+    exists = (server_dir / "pq_leaf.parquet").exists() and (server_dir / "pq_leaf.parquet").stat().st_size > 0
+    expect.text("yes" if exists else "no", "yes",
+                "premise: a parquet file exists, so the import reaches the relation")
+
+    expect.num(_one(conn, "SELECT pgcolumnar.import_parquet('pq_leaf', %s)", (path,)), 50,
+               "control: importing that file into the LEAF still works")
+    with conn.cursor() as cur:
+        cur.execute("CREATE TABLE pq_heap (id int, v text)")
+    expect.num(_one(conn, "SELECT pgcolumnar.import_parquet('pq_heap', %s)", (path,)), 50,
+               "control: a HEAP target still imports, so the guard is about storage")
+
+    crashes = _crashes(pgc_cluster)
+    expect.text(_sqlstate(conn, "SELECT pgcolumnar.import_parquet('pq_parent', %s)", (path,)), "42809",
+                "importing into the partitioned parent is refused")
+    alive = _sqlstate(conn, "SELECT 1") == "accepted"
+    expect.text(f"{'1' if alive else ''}|{_crashes(pgc_cluster) - crashes} backend crashes", "1|0 backend crashes",
+                "and the cluster is still up, so the refusal was not an abort")
diff --git a/test/pytest/test_phase5.py b/test/pytest/test_phase5.py
new file mode 100644
index 000000000..ae3715d2a
--- /dev/null
+++ b/test/pytest/test_phase5.py
@@ -0,0 +1,80 @@
+"""set_options and reset_options reach the same verdict on every kind of relation, and
+both accept a columnar materialized view (#1265, #1341). Part of the pytest twin of
+`test/phase5.sh`: these arms only, by the shell's names, and the pair is declared
+INCOMPLETE.
+
+reset_options had no guard at all, so it reported success for a heap table and a
+partitioned parent while set_options refused the same relation. The property is
+parity, not a relkind list. Parity alone is satisfied by both refusing, so the
+settled verdict on a matview (accept) and on a partitioned parent (refuse) are
+asserted by name too, and an ordinary columnar table is the control both must accept.
+"""
+import psycopg
+
+
+def _verdict(conn, sql):
+    try:
+        with conn.cursor() as cur:
+            cur.execute(sql)
+        return "accepted"
+    except psycopg.Error:
+        return "refused"
+
+
+def _set(conn, rel):
+    return _verdict(conn, f"SELECT pgcolumnar.set_options('{rel}', chunk_group_row_limit => 1000)")
+
+
+def _reset(conn, rel):
+    return _verdict(conn, f"SELECT pgcolumnar.reset_options('{rel}', chunk_group_row_limit => true)")
+
+
+def _agree(conn, rel):
+    s, r = _set(conn, rel), _reset(conn, rel)
+    return "agree" if s == r else f"set={s} reset={r}"
+
+
+def _one(conn, sql):
+    with conn.cursor() as cur:
+        cur.execute(sql)
+        return cur.fetchone()[0]
+
+
+def test_set_and_reset_options_agree(pgc_conn, expect):
+    conn = pgc_conn
+    with conn.cursor() as cur:
+        cur.execute("CREATE TABLE o_par_col (a int) USING pgcolumnar")
+        cur.execute("CREATE TABLE o_par_heap (a int)")
+        cur.execute("CREATE TABLE o_par_part (a int) PARTITION BY RANGE (a)")
+        cur.execute("CREATE TABLE o_par_mvsrc (a int)")
+        cur.execute("INSERT INTO o_par_mvsrc VALUES (1), (2)")
+        cur.execute("CREATE MATERIALIZED VIEW o_par_mv USING pgcolumnar AS SELECT a FROM o_par_mvsrc")
+
+    expect.text(_one(conn, "SELECT c.relkind::text || am.amname::text FROM pg_class c JOIN pg_am am "
+                           "ON am.oid = c.relam WHERE c.oid = 'o_par_col'::regclass"), "rpgcolumnar",
+                "premise: the columnar fixture is relkind r on the columnar access method")
+    expect.text(_one(conn, "SELECT c.relkind::text || (am.amname <> 'pgcolumnar')::text FROM pg_class c "
+                           "JOIN pg_am am ON am.oid = c.relam WHERE c.oid = 'o_par_heap'::regclass"), "rtrue",
+                "premise: the heap fixture is relkind r and is NOT columnar")
+    expect.text(_one(conn, "SELECT c.relkind::text || c.relfilenode::text FROM pg_class c "
+                           "WHERE c.oid = 'o_par_part'::regclass"), "p0",
+                "premise: the partitioned fixture is relkind p with no storage")
+    expect.text(_one(conn, "SELECT c.relkind::text || am.amname::text FROM pg_class c JOIN pg_am am "
+                           "ON am.oid = c.relam WHERE c.oid = 'o_par_mv'::regclass"), "mpgcolumnar",
+                "premise: the matview fixture is relkind m on the columnar access method")
+
+    # Without it "agree" is satisfied by a build where both refuse everything.
+    expect.text(f"{_set(conn, 'o_par_col')}/{_reset(conn, 'o_par_col')}", "accepted/accepted",
+                "control: both entry points accept an ordinary columnar table")
+
+    expect.text(_agree(conn, "o_par_col"), "agree", "set_options and reset_options agree about a columnar table")
+    expect.text(_agree(conn, "o_par_heap"), "agree", "set_options and reset_options agree about a heap table")
+    expect.text(_agree(conn, "o_par_part"), "agree",
+                "set_options and reset_options agree about a partitioned parent")
+    expect.text(_agree(conn, "o_par_mv"), "agree", "set_options and reset_options agree about a columnar matview")
+
+    expect.text(f"{_set(conn, 'o_par_mv')}/{_reset(conn, 'o_par_mv')}", "accepted/accepted",
+                "both entry points accept a columnar matview (#1265, settled)")
+    # Widening to matviews is not widening to everything without storage.
+    expect.text(f"{_set(conn, 'o_par_part')}/{_reset(conn, 'o_par_part')}", "refused/refused",
+                "control: a partitioned parent is still refused by both")
diff --git a/test/suites_without_accounting.txt b/test/suites_without_accounting.txt
index 515d4b41a..39b824f88 100644
--- a/test/suites_without_accounting.txt
+++ b/test/suites_without_accounting.txt
@@ -20,6 +20,5 @@ concurrency
 phase2
 phase3
 phase4
-phase5
 phase6
 smoke