From d0c9e2c354925baadf9302e8fa8259d3f8e2f3da Mon Sep 17 00:00:00 2001
From: "Joshua D. Drake" <jd@commandprompt.com>
Date: Wed, 21 Oct 2026 21:25:43 +0000
Subject: [PATCH] test: iceberg_malformed gains its pytest twin, and a crash or
 an empty read no longer passes as a refusal (#1341)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The pytest twin of test/iceberg_malformed.sh, one test per check by the shell's
name, plus two that mirror its skip names. Two fixes to the shell, measured on
PG17 at origin/main:

- Every refusal arm read only XX001, which later parse steps also raise. With
  the S_ISREG guard in PgColumnarOpenLocalRegularFile removed, a FIFO opens
  O_NONBLOCK and reads as an empty file, and four of the seven FIFO arms still
  passed. Each arm now also requires its refusal's own message.
- Every "backend still up" check was a bare SELECT 1, which answers again after
  crash recovery. With a NULL path segfaulting the backend and the probe 3s
  later, both survival checks passed. They now read the server log's count of
  killed backends since their own arm, and wait out a recovery so the arms after
  a crash are not collateral.

Seven source mutations each redden the same names in shell and twin. The
array-valued "fields" arm stays never: on a build without asserts the schema
walker still refuses through av_schema_err with the guard removed. Census 3853
-> 3868 never, 2243 -> 2259 dated, 6127 rows; suites_not_covered 105 -> 104;
cluster_tests 2677 -> 2710; TESTS.md §265.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 test/check_ledger.tsv                 |  31 ++
 test/check_ledger_budget.txt          |  22 +-
 test/iceberg_malformed.sh             | 138 +++++---
 test/pytest/TESTS.md                  |  45 +++
 test/pytest/expected_tests.txt        |   4 +-
 test/pytest/test_compare_to_bash.py   |   1 +
 test/pytest/test_iceberg_malformed.py | 446 ++++++++++++++++++++++++++
 test/pytest/unported_suites.txt       |   1 -
 8 files changed, 642 insertions(+), 46 deletions(-)
 create mode 100644 test/pytest/test_iceberg_malformed.py

diff --git a/test/check_ledger.tsv b/test/check_ledger.tsv
index a9fdef60b..10900f8bb 100644
--- a/test/check_ledger.tsv
+++ b/test/check_ledger.tsv
@@ -2560,6 +2560,37 @@ iceberg_fdw_projection	iceberg_fdw_projection	one column : FDW projection matche
 iceberg_fdw_projection	iceberg_fdw_projection	select one, filter another : FDW projection matches iceberg_scan	15;16;17;18;19	2026-10-21	the Iceberg FDW's projection mask ignores columns only the scan clauses reference
 iceberg_fdw_projection	iceberg_fdw_projection	three columns : FDW projection matches iceberg_scan	15;16;17;18;19	never	-
 iceberg_fdw_projection	iceberg_fdw_projection	whole row : FDW projection matches iceberg_scan	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	a NULL data-file path is refused in iceberg_data_files (XX001, not a crash)	15;16;17;18;19	2026-10-21	columnar_iceberg.c: a manifest entry with a NULL file_path is not refused
+iceberg_malformed	iceberg_malformed	a NULL data-file path is refused in iceberg_scan (XX001, not a crash)	15;16;17;18;19	2026-10-21	columnar_iceberg.c: a manifest entry with a NULL file_path is not refused
+iceberg_malformed	iceberg_malformed	a NULL manifest_path is refused, not a crash (XX001)	15;16;17;18;19	2026-10-21	columnar_iceberg.c: ice_open_path does not refuse a NULL recorded path
+iceberg_malformed	iceberg_malformed	a dangling current-schema-id is refused (XX001), not silently stale-read	15;16;17;18;19	2026-10-21	columnar_iceberg.c: a dangling current-schema-id resolves to the legacy top-level schema
+iceberg_malformed	iceberg_malformed	a manifest-list that is a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	a null manifest-list min_sequence_number decodes as NULL, not 0 (#686)	15;16;17;18;19	2026-10-21	columnar_avro.c: read_manifest_list decodes a null min_sequence_number as 0
+iceberg_malformed	iceberg_malformed	a null manifest-list sequence_number decodes as NULL, not 0	15;16;17;18;19	2026-10-21	columnar_avro.c: read_manifest_list decodes a null sequence_number as 0
+iceberg_malformed	iceberg_malformed	an array-valued schema fields element is refused, not crashed (XX001)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the FIFO refusals	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the dangling-schema refusal	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the lister refusal	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the malformed-schema refusal	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the manifest-list FIFO refusal	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the null-manifest_path refusal	15;16;17;18;19	2026-10-21	columnar_iceberg.c: ice_open_path does not refuse a NULL recorded path
+iceberg_malformed	iceberg_malformed	backend still up after the null-sequence refusal	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	backend still up after the scan refusal	15;16;17;18;19	2026-10-21	columnar_iceberg.c: a manifest entry with a NULL file_path is not refused
+iceberg_malformed	iceberg_malformed	control: a concrete manifest seq scans all 5 rows (seq-3 delete is older than seq-5 data)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a legacy-only-schema table still reads (5 rows)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a present file_path lists one data file	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a present manifest-list sequence_number decodes to its value	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a present min_sequence_number decodes to its value (#686)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a resolvable current-schema-id reads the correct amount column	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: a well-formed field object still decodes (0 entries)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	control: the legacy table binds amount correctly (10..50)	15;16;17;18;19	never	-
+iceberg_malformed	iceberg_malformed	iceberg_current_snapshot on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	iceberg_data_files on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	iceberg_scan on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	iceberg_scan refuses a null manifest seq an ADDED entry must inherit (XX001, not a silent wrong row)	15;16;17;18;19	2026-10-21	columnar_iceberg.c: an added entry inherits a null manifest sequence number as 0
+iceberg_malformed	iceberg_malformed	read_avro_manifest on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	read_manifest_list on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
+iceberg_malformed	iceberg_malformed	read_parquet on a FIFO is refused, not a hang (XX001)	15;16;17;18;19	2026-10-21	columnar_objstore.c: PgColumnarOpenLocalRegularFile drops its S_ISREG guard, so a FIFO opens O_NONBLOCK and reads as an empty file
 iceberg_name_mapping_memory	iceberg_name_mapping_memory	and the mapped values are still correct	15;16;17;18;19	2026-10-21	the name-mapping parse binds every name but region, so region reads NULL on every row;the name-mapping parse ignores every entry's names, so nothing binds
 iceberg_name_mapping_memory	iceberg_name_mapping_memory	core's own parse is the larger part of the scan's peak	15;16;17;18;19	2026-10-21	the name-mapping parse ignores every entry's names, so nothing binds
 iceberg_name_mapping_memory	iceberg_name_mapping_memory	the large arm produced three readings	15;16;17;18;19	2026-10-21	the name-mapping parse ignores every entry's names, so nothing binds
diff --git a/test/check_ledger_budget.txt b/test/check_ledger_budget.txt
index 07074f0c9..66efc310d 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 105
+suites_not_covered 104
 #
 # checks_never_observed_red -- A CENSUS. NOT a ceiling, and it must not become
 #   one.
@@ -5374,4 +5374,22 @@ suites_not_covered 105
 #     never         3850 + 3 = 3853
 #     dated         2239 + 4 = 2243
 #     3853 + 2243 == 6096
-checks_never_observed_red 3853
+# #1341, iceberg_malformed seeded, with its twin (OffgridwithJD): 3853 -> 3868,
+# and suites_not_covered 105 -> 104. The suite had no rows. It enters with 31
+# checks from five-major runs (all green on each of 15-19, shell and twin alike), after
+# two fixes measured on PG17 at origin/main. Every refusal arm read only XX001, which
+# later parse steps also raise: with the regular-file guard removed, a FIFO opened
+# O_NONBLOCK reads as an empty file and four of the seven FIFO arms still passed. Each
+# arm now also names its refusal's message. And every "backend still up" check was a
+# bare SELECT 1, which answers again after crash recovery: with a NULL path segfaulting
+# the backend and the probe 3s later, both survival checks passed. They now read the
+# server log's crash count since their own arm, and wait out a recovery. 16 are dated
+# by seven mutations, each reddening the same names in both twins; 15 arrive never.
+# The array-valued "fields" arm stays never: on a build without asserts the schema
+# walker still refuses through av_schema_err, so dropping the object check passes all 31.
+#
+#     ledger rows   6096 + 31 = 6127
+#     never         3853 + 15 = 3868
+#     dated         2243 + 16 = 2259
+#     3868 + 2259 == 6127
+checks_never_observed_red 3868
diff --git a/test/iceberg_malformed.sh b/test/iceberg_malformed.sh
index 872dd228a..588f62063 100755
--- a/test/iceberg_malformed.sh
+++ b/test/iceberg_malformed.sh
@@ -5,8 +5,10 @@
 # whoever authored the table, and is called by an honest pg_read_server_files
 # role that named only the trusted metadata pointer. A crafted manifest must be
 # REFUSED cleanly, never crash the backend. Each arm has a control so a vacuous
-# pass is visible, and a "backend still up" probe after every refusal proves the
-# refusal was an error, not a segfault. Fixtures are hand-crafted
+# pass is visible, each refusal names its own message (the SQLSTATE alone is
+# shared with later parse errors), and a "backend still up" probe after every
+# refusal reads the server log's crash count, so a segfault followed by crash
+# recovery does not pass as a refusal. Fixtures are hand-crafted
 # (test/fixtures/iceberg/warehouse_malformed) because no writer emits these
 # shapes; see gen_malformed_fixture.py.
 #
@@ -16,31 +18,69 @@ set -uo pipefail
 . "$(dirname "${BASH_SOURCE[0]}")/lib.sh"
 pgc_setup "${1:-/usr/local/pg17/bin/pg_config}"
 
-sqlstate_of() {
+# refusal SQL REGEX LABEL -> "<SQLSTATE> (LABEL)" when SQL raised an error whose message
+# matches REGEX, "<SQLSTATE>: <message>" for any other error, or "no error". Every
+# arm here refuses with XX001, and so do later parse steps: a FIFO read as an empty
+# file failed a parse with the same XX001, so the SQLSTATE alone passed with the
+# guard removed. The arm now names the refusal it wants.
+refusal_of() {	# stdin: psql output -> the refusal reading
+	local re="$1" label="$2" line state msg
+	local out; out="$(cat)"
+	line="$(sed -n 's/^ERROR:  \([0-9A-Z]\{5\}\): /\1|/p' <<<"$out" | head -1)"
+	if [ -z "$line" ]; then
+		grep -qE 'server closed the connection|connection to server' <<<"$out" \
+			&& echo "no refusal: the connection was lost" || echo "no error"
+		return
+	fi
+	state="${line%%|*}"; msg="${line#*|}"
+	if grep -qE -- "$re" <<<"$msg"; then echo "$state ($label)"; else echo "$state: $msg"; fi
+}
+refusal() {
 	env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \
-		-d "$PGC_DB" -qtA 2>&1 <<SQLEOF | sed -n 's/^ERROR:  \([0-9A-Z]\{5\}\).*/\1/p' | head -1
-\\set VERBOSITY sqlstate
+		-d "$PGC_DB" -qtA 2>&1 <<SQLEOF | refusal_of "$2" "$3"
+\\set VERBOSITY verbose
 $1;
 SQLEOF
 }
 
 # Run one statement under a HARD wall-clock cap. A cancel-resistant open (a FIFO)
 # cannot be ended by statement_timeout, so the external `timeout` is the only
-# reliable detector. Prints the 5-char SQLSTATE the backend raised, or the
-# literal HANG when psql was KILLed by the cap (exit 124/137). No SQL-level
+# reliable detector. Prints the refusal reading as refusal() does, or the literal
+# HANG when psql was KILLed by the cap (exit 124/137). No SQL-level
 # statement_timeout is set, so a merely-slow legit path is never a false HANG.
 FIFO_CAP=5
-sqlstate_or_hang() {
+FIFO_RE='is not a regular file'
+refusal_or_hang() {
 	local out rc
 	out="$(timeout -s KILL "$FIFO_CAP" env PATH="$PGC_BINDIR:$PATH" psql \
 		-h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -qtA 2>&1 <<SQLEOF
-\\set VERBOSITY sqlstate
+\\set VERBOSITY verbose
 $1;
 SQLEOF
 )"
 	rc=$?
 	if [ "$rc" -eq 124 ] || [ "$rc" -eq 137 ]; then echo HANG; return; fi
-	printf '%s\n' "$out" | sed -n 's/^ERROR:  \([0-9A-Z]\{5\}\).*/\1/p' | head -1
+	printf '%s\n' "$out" | refusal_of "$FIFO_RE" "not a regular file"
+}
+
+# Backends killed by a signal, from the server log: a number, or nothing when the
+# log can't be read. A fresh connection answers again after crash recovery, so
+# "SELECT 1" alone passed after a backend segfaulted on a NULL path, once the probe
+# came 3s later. Every survival check also requires that this count has not moved
+# since just before its own arm ran; when it has, the check waits out the crash
+# recovery, so the arms after it are not collateral.
+crashes() {
+	local _n
+	_n="$(pgc_pg "grep -c 'terminated by signal' '$PGC_LOGFILE'" 2>/dev/null)"
+	case "$_n" in '' | *[!0-9]*) ;; *) echo "$_n" ;; esac
+}
+survived() {	# survived BEFORE -> "1" when the server answers and no backend died since BEFORE
+	local _up _now _i
+	_up="$(q 'SELECT 1')"; _now="$(crashes)"
+	if ! [[ "$1" =~ ^[0-9]+$ && "$_now" =~ ^[0-9]+$ ]]; then echo "no (crash count unreadable)"; return; fi
+	if [ "$_up" = 1 ] && [ "$_now" = "$1" ]; then echo 1; return; fi
+	for _i in $(seq 1 60); do [ "$(q 'SELECT 1' 2>/dev/null)" = 1 ] && break; sleep 0.5; done
+	echo "no ([$_up], $(( _now - $1 )) backend crashes)"
 }
 
 # Release a backend that is blocked in open(2) on a FIFO (main only): opening the
@@ -70,10 +110,12 @@ chmod -R u+rwX "$DEST"
 #     backend (an object-container Assert / OOB heap read). It must now be a
 #     clean malformed-schema error, and the control must still decode.
 # =========================================================================
-check "an array-valued schema \"fields\" element is refused, not crashed (XX001)" \
-	"$(sqlstate_of "SELECT * FROM pgcolumnar.read_avro_manifest('$DEST/arrayfield/evil.avro')")" \
-	"XX001"
-check "backend still up after the malformed-schema refusal" "$(q 'SELECT 1')" "1"
+B="$(crashes)"
+check "an array-valued schema fields element is refused, not crashed (XX001)" \
+	"$(refusal "SELECT * FROM pgcolumnar.read_avro_manifest('$DEST/arrayfield/evil.avro')" \
+		'unsupported or malformed Avro schema in manifest' "malformed schema")" \
+	"XX001 (malformed schema)"
+check "backend still up after the malformed-schema refusal" "$(survived "$B")" "1"
 check "control: a well-formed field object still decodes (0 entries)" \
 	"$(q "SELECT count(*) FROM pgcolumnar.read_avro_manifest('$DEST/arrayfield/good.avro')")" \
 	"0"
@@ -84,12 +126,16 @@ check "control: a well-formed field object still decodes (0 entries)" \
 #     be a clean DATA_CORRUPTED refusal; the control (file_path present) reads.
 # =========================================================================
 NP="$DEST/nullpath/db/t/metadata/t.metadata.json"
+B="$(crashes)"
 check "a NULL data-file path is refused in iceberg_data_files (XX001, not a crash)" \
-	"$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_data_files('$NP')")" "XX001"
-check "backend still up after the lister refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal "SELECT * FROM pgcolumnar.iceberg_data_files('$NP')" \
+		'has no data file path' "no data file path")" "XX001 (no data file path)"
+check "backend still up after the lister refusal" "$(survived "$B")" "1"
+B="$(crashes)"
 check "a NULL data-file path is refused in iceberg_scan (XX001, not a crash)" \
-	"$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$NP') AS t(id bigint)")" "XX001"
-check "backend still up after the scan refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$NP') AS t(id bigint)" \
+		'has no data file path' "no data file path")" "XX001 (no data file path)"
+check "backend still up after the scan refusal" "$(survived "$B")" "1"
 check "control: a present file_path lists one data file" \
 	"$(q "SELECT count(*) FROM pgcolumnar.iceberg_data_files('$DEST/nullpath_ctl/db/t/metadata/t.metadata.json')")" \
 	"1"
@@ -101,9 +147,11 @@ check "control: a present file_path lists one data file" \
 #     author-controlled, so this is reachable. Refuse it as DATA_CORRUPTED.
 # =========================================================================
 NMP="$DEST/nullmanifestpath/db/t/metadata/t.metadata.json"
+B="$(crashes)"
 check "a NULL manifest_path is refused, not a crash (XX001)" \
-	"$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_data_files('$NMP')")" "XX001"
-check "backend still up after the null-manifest_path refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal "SELECT * FROM pgcolumnar.iceberg_data_files('$NMP')" \
+		'a manifest path in .* is null' "null manifest path")" "XX001 (null manifest path)"
+check "backend still up after the null-manifest_path refusal" "$(survived "$B")" "1"
 
 # =========================================================================
 # #1  a non-regular file (FIFO) named on any Iceberg-reachable open must be
@@ -113,31 +161,32 @@ check "backend still up after the null-manifest_path refusal" "$(q 'SELECT 1')"
 FIFODIR="$PGC_WORKDIR/fifo"; mkdir -p "$FIFODIR"
 FIFO="$FIFODIR/x"; mkfifo "$FIFO"
 
+B="$(crashes)"
 # Site A -- ice_slurp_text, FIFO as the metadata argument (three entry points)
 check "iceberg_current_snapshot on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.iceberg_current_snapshot('$FIFO')")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.iceberg_current_snapshot('$FIFO')")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 check "iceberg_data_files on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.iceberg_data_files('$FIFO')")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.iceberg_data_files('$FIFO')")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 check "iceberg_scan on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.iceberg_scan('$FIFO') AS t(id bigint)")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.iceberg_scan('$FIFO') AS t(id bigint)")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 
 # Site C -- av_slurp_file, FIFO as the argument (both entry points)
 check "read_avro_manifest on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.read_avro_manifest('$FIFO')")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.read_avro_manifest('$FIFO')")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 check "read_manifest_list on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.read_manifest_list('$FIFO')")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.read_manifest_list('$FIFO')")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 
 # Site D -- pq_source_open_cfg via the single-file parquet API (sibling)
 check "read_parquet on a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.read_parquet('$FIFO') AS t(id bigint)")" \
-	"XX001"; fifo_release "$FIFO"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.read_parquet('$FIFO') AS t(id bigint)")" \
+	"XX001 (not a regular file)"; fifo_release "$FIFO"
 
-check "backend still up after the FIFO refusals" "$(q 'SELECT 1')" "1"
+check "backend still up after the FIFO refusals" "$(survived "$B")" "1"
 
 # Site B -- ice_slurp_bin, FIFO named as the manifest-list INSIDE metadata.json.
 # realpath() on the FIFO succeeds and passes ice_open_path's containment, so only
@@ -153,10 +202,11 @@ cat > "$MDJSON" <<JSON
    "schema-id":0,"manifest-list":"file://$MLROOT/db/t/metadata/ml.avro",
    "summary":{"operation":"append"}}]}
 JSON
+B="$(crashes)"
 check "a manifest-list that is a FIFO is refused, not a hang (XX001)" \
-	"$(sqlstate_or_hang "SELECT * FROM pgcolumnar.iceberg_data_files('$MDJSON')")" \
-	"XX001"; fifo_release "$MLFIFO"
-check "backend still up after the manifest-list FIFO refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal_or_hang "SELECT * FROM pgcolumnar.iceberg_data_files('$MDJSON')")" \
+	"XX001 (not a regular file)"; fifo_release "$MLFIFO"
+check "backend still up after the manifest-list FIFO refusal" "$(survived "$B")" "1"
 
 # =========================================================================
 # #4  a manifest_file.sequence_number that is the union's NULL branch decoded
@@ -182,10 +232,12 @@ check "control: a present min_sequence_number decodes to its value (#686)" \
 
 NS="$DEST/nullseq/db/t/metadata/t.metadata.json"
 NSC="$DEST/nullseq_ctl/db/t/metadata/t.metadata.json"
+B="$(crashes)"
 check "iceberg_scan refuses a null manifest seq an ADDED entry must inherit (XX001, not a silent wrong row)" \
-	"$(sqlstate_of "SELECT count(*) FROM pgcolumnar.iceberg_scan('$NS') AS t(id bigint, region text, amount int)")" \
-	"XX001"
-check "backend still up after the null-sequence refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal "SELECT count(*) FROM pgcolumnar.iceberg_scan('$NS') AS t(id bigint, region text, amount int)" \
+		'has a null sequence number that an added entry' "null sequence to inherit")" \
+	"XX001 (null sequence to inherit)"
+check "backend still up after the null-sequence refusal" "$(survived "$B")" "1"
 check "control: a concrete manifest seq scans all 5 rows (seq-3 delete is older than seq-5 data)" \
 	"$(q "SELECT count(*) FROM pgcolumnar.iceberg_scan('$NSC') AS t(id bigint, region text, amount int)")" \
 	"5"
@@ -197,10 +249,12 @@ check "control: a concrete manifest seq scans all 5 rows (seq-3 delete is older
 #     schema and misprojects every row). Asymmetric with ice_current_snapshot.
 # =========================================================================
 DS="$DEST/danglingschema/db/t/metadata/t.metadata.json"
+B="$(crashes)"
 check "a dangling current-schema-id is refused (XX001), not silently stale-read" \
-	"$(sqlstate_of "SELECT amount FROM pgcolumnar.iceberg_scan('$DS') AS t(amount int)")" \
-	"XX001"
-check "backend still up after the dangling-schema refusal" "$(q 'SELECT 1')" "1"
+	"$(refusal "SELECT amount FROM pgcolumnar.iceberg_scan('$DS') AS t(amount int)" \
+		'is not present in the schemas array' "dangling schema id")" \
+	"XX001 (dangling schema id)"
+check "backend still up after the dangling-schema refusal" "$(survived "$B")" "1"
 # control: the SAME data with a RESOLVABLE current-schema-id reads the correct
 # column (amount = field id 3 = 10..50). Proves only the dangling id flips it.
 DSC="$DEST/danglingschema_ctl/db/t/metadata/t.metadata.json"
diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md
index 74f2e000b..95399115e 100644
--- a/test/pytest/TESTS.md
+++ b/test/pytest/TESTS.md
@@ -310,6 +310,7 @@ behaviour, the source of that number is named.
 - [262. test_parallel_flush_optin.py: parallel_flush stays opt-in, and the dispatch metric cannot separate the shapes](#262-test_parallel_flush_optinpy-parallel_flush-stays-opt-in-and-the-dispatch-metric-cannot-separate-the-shapes)
 - [263. test_parquet_export_stats.py: the Parquet files we export carry correct per-row-group statistics](#263-test_parquet_export_statspy-the-parquet-files-we-export-carry-correct-per-row-group-statistics)
 - [264. test_catalog_filter_reads.py: stats() and sort_status() read a table's storage id once](#264-test_catalog_filter_readspy-stats-and-sort_status-read-a-tables-storage-id-once)
+- [265. test_iceberg_malformed.py: the Iceberg reader refuses crafted metadata cleanly, never crashing or hanging](#265-test_iceberg_malformedpy-the-iceberg-reader-refuses-crafted-metadata-cleanly-never-crashing-or-hanging)
 
 ## 1. How to read a test in here
 
@@ -10270,3 +10271,47 @@ The pytest twin of `test/catalog_filter_reads.sh` (#1637), every check by the sh
 | `test_stats_reports_only_the_tables_groups` | control: stats() reports t's 3 row groups and no other table's |
 | `test_sort_status_reads_the_storage_id_once` | sort_status() reads the storage id once, not once per catalog row |
 | `test_sort_status_counts_only_the_tables_groups` | control: sort_status() counts t's 3 row groups and no other table's |
+
+## 265. test_iceberg_malformed.py: the Iceberg reader refuses crafted metadata cleanly, never crashing or hanging
+
+The twin of `test/iceberg_malformed.sh` (#388 phase 4a, #644). The hand-crafted fixtures in `test/fixtures/iceberg/warehouse_malformed` are refused with the refusal each arm names, not merely its SQLSTATE: an array-valued Avro `fields` element, a NULL data-file path in the lister and the scan, a NULL manifest_path, a FIFO at every Iceberg-reachable open (a HANG after 5s), a null manifest sequence number an added entry must inherit, and a dangling current-schema-id. Null manifest-list sequence numbers decode as NULL, and every refusal has a control that still reads.
+
+### Every test
+
+One module fixture copies the fixtures into `server_dir`, builds the FIFOs there, and records every reading from a private database. A FIFO read runs in a thread with a 5s deadline. Each survival reading compares the server log's crash count with the one taken just before its arm, and waits out a crash recovery. The first two tests mirror the shell's two skip names.
+
+| test | what it holds |
+| --- | --- |
+| `test_the_malformed_iceberg_fixtures_are_present` | premise: the committed malformed fixtures are not empty |
+| `test_python_reads_the_malformed_metadata` | premise: the dangling-schema metadata parses as JSON |
+| `test_an_array_valued_fields_element_is_refused` | an array-valued schema fields element is refused, not crashed (XX001) |
+| `test_the_backend_survives_the_malformed_schema` | backend still up after the malformed-schema refusal |
+| `test_a_well_formed_field_object_still_decodes` | control: a well-formed field object still decodes (0 entries) |
+| `test_the_lister_refuses_a_null_data_file_path` | a NULL data-file path is refused in iceberg_data_files (XX001, not a crash) |
+| `test_the_backend_survives_the_lister_refusal` | backend still up after the lister refusal |
+| `test_the_scan_refuses_a_null_data_file_path` | a NULL data-file path is refused in iceberg_scan (XX001, not a crash) |
+| `test_the_backend_survives_the_scan_refusal` | backend still up after the scan refusal |
+| `test_a_present_file_path_lists_one_data_file` | control: a present file_path lists one data file |
+| `test_a_null_manifest_path_is_refused` | a NULL manifest_path is refused, not a crash (XX001) |
+| `test_the_backend_survives_the_null_manifest_path` | backend still up after the null-manifest_path refusal |
+| `test_iceberg_current_snapshot_refuses_a_fifo` | iceberg_current_snapshot on a FIFO is refused, not a hang (XX001) |
+| `test_iceberg_data_files_refuses_a_fifo` | iceberg_data_files on a FIFO is refused, not a hang (XX001) |
+| `test_iceberg_scan_refuses_a_fifo` | iceberg_scan on a FIFO is refused, not a hang (XX001) |
+| `test_read_avro_manifest_refuses_a_fifo_as_iceberg_input` | read_avro_manifest on a FIFO is refused, not a hang (XX001) |
+| `test_read_manifest_list_refuses_a_fifo` | read_manifest_list on a FIFO is refused, not a hang (XX001) |
+| `test_read_parquet_refuses_a_fifo_beside_iceberg` | read_parquet on a FIFO is refused, not a hang (XX001) |
+| `test_the_backend_survives_the_fifo_refusals` | backend still up after the FIFO refusals |
+| `test_a_manifest_list_that_is_a_fifo_is_refused` | a manifest-list that is a FIFO is refused, not a hang (XX001) |
+| `test_the_backend_survives_the_manifest_list_fifo` | backend still up after the manifest-list FIFO refusal |
+| `test_a_null_manifest_list_sequence_number_reads_null` | a null manifest-list sequence_number decodes as NULL, not 0 |
+| `test_a_present_manifest_list_sequence_number_reads_its_value` | control: a present manifest-list sequence_number decodes to its value |
+| `test_a_null_min_sequence_number_reads_null` | a null manifest-list min_sequence_number decodes as NULL, not 0 (#686) |
+| `test_a_present_min_sequence_number_reads_its_value` | control: a present min_sequence_number decodes to its value (#686) |
+| `test_the_scan_refuses_a_null_sequence_an_added_entry_inherits` | iceberg_scan refuses a null manifest seq an ADDED entry must inherit (XX001, not a silent wrong row) |
+| `test_the_backend_survives_the_null_sequence_refusal` | backend still up after the null-sequence refusal |
+| `test_a_concrete_manifest_sequence_scans_every_row` | control: a concrete manifest seq scans all 5 rows (seq-3 delete is older than seq-5 data) |
+| `test_a_dangling_current_schema_id_is_refused` | a dangling current-schema-id is refused (XX001), not silently stale-read |
+| `test_the_backend_survives_the_dangling_schema_refusal` | backend still up after the dangling-schema refusal |
+| `test_a_resolvable_current_schema_id_reads_amount` | control: a resolvable current-schema-id reads the correct amount column |
+| `test_a_legacy_only_schema_table_still_reads` | control: a legacy-only-schema table still reads (5 rows) |
+| `test_the_legacy_table_binds_amount` | control: the legacy table binds amount correctly (10..50) |
diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt
index c4c6e9a81..cffb8bff7 100644
--- a/test/pytest/expected_tests.txt
+++ b/test/pytest/expected_tests.txt
@@ -1025,6 +1025,8 @@ guard_tests 474
 # 5 tests, one per check, by collection over the cluster files.
 # cluster_tests 2508 -> 2670 for #1341's test_parquet_export_stats.py, the twin of parquet_export_stats.sh:
 # 162 tests, one per check, by collection over the cluster files.
+# cluster_tests 2677 -> 2710 for #1341's test_iceberg_malformed.py, the twin of iceberg_malformed.sh:
+# 31 checks in 33 tests, by collection over the cluster files.
 # cluster_tests 2670 -> 2677 for #1637's test_catalog_filter_reads.py, the twin of catalog_filter_reads.sh:
 # 7 tests, one per check, by collection over the cluster files.
-cluster_tests 2677
+cluster_tests 2710
diff --git a/test/pytest/test_compare_to_bash.py b/test/pytest/test_compare_to_bash.py
index f6c63c000..d89e8525b 100644
--- a/test/pytest/test_compare_to_bash.py
+++ b/test/pytest/test_compare_to_bash.py
@@ -124,6 +124,7 @@
             "iceberg_name_mapping_memory",
             "iceberg_catalog",
             "iceberg_scan",
+            "iceberg_malformed",
             "concurrency",
             "isolation",
             "index_only",
diff --git a/test/pytest/test_iceberg_malformed.py b/test/pytest/test_iceberg_malformed.py
new file mode 100644
index 000000000..5b966150b
--- /dev/null
+++ b/test/pytest/test_iceberg_malformed.py
@@ -0,0 +1,446 @@
+"""The Iceberg reader refuses crafted metadata, manifests and manifest lists cleanly,
+never crashing or hanging the backend (#388 phase 4a, #644). The pytest twin of
+`test/iceberg_malformed.sh`, every check by the shell's name (#1341).
+
+The hand-crafted fixtures under test/fixtures/iceberg/warehouse_malformed are copied
+into `server_dir`, so the recorded root differs from the actual one and every read
+goes through the same path rebasing the honest read path uses. Each refusal is read
+as "<SQLSTATE> (<label>)" only when the error's primary message is the refusal under
+test: every arm refuses with XX001, and so do later parse steps, so a FIFO read as an
+empty file passed on its SQLSTATE alone with the regular-file guard removed. A FIFO
+read that is still running after 5s is a HANG; its write end is then opened, so the
+backend is released.
+
+A BACKEND THAT DIED DOES NOT PASS AS ONE THAT SURVIVED. A fresh connection answers
+again after crash recovery, so every survival check also reads the server log's count
+of killed backends since the fixture began, and an unreadable log fails it.
+
+ON A PRIVATE DATABASE; each test asserts one reading (#1339).
+"""
+import json
+import os
+import pathlib
+import re
+import shutil
+import threading
+import time
+
+import psycopg
+import pytest
+
+from pgc_vacuity import query_error
+
+DB = "pgc_iceberg_malformed"
+WHM = pathlib.Path(__file__).resolve().parents[1] / "fixtures" / "iceberg" / "warehouse_malformed"
+NEEDED = ("nullpath/db/t/metadata/t.metadata.json", "nullseq_ml/ml.avro", "nullminseq_ml/ml.avro",
+          "nullmanifestpath/db/t/metadata/t.metadata.json", "danglingschema/db/t/metadata/t.metadata.json")
+# the refusal each arm wants, matched against the primary message as the shell does
+SCHEMA = (r"unsupported or malformed Avro schema in manifest", "malformed schema")
+NOPATH = (r"has no data file path", "no data file path")
+NULLMP = (r"a manifest path in .* is null", "null manifest path")
+FIFO_REFUSAL = (r"is not a regular file", "not a regular file")
+NULLSEQ = (r"has a null sequence number that an added entry", "null sequence to inherit")
+DANGLING = (r"is not present in the schemas array", "dangling schema id")
+
+
+def _message(exc):
+    return str(exc).splitlines()[0] if str(exc) else type(exc).__name__
+
+
+def _value(dsn, sql):
+    """-> the first column of the first row as the shell's psql prints it, or the
+    failure as a query_error()."""
+    try:
+        with psycopg.connect(dsn, autocommit=True) as conn:
+            row = conn.execute(sql).fetchone()
+    except psycopg.Error as exc:
+        return query_error(_message(exc))
+    if row is None or row[0] is None:
+        return query_error(f"no value from {sql}")
+    v = row[0]
+    return "t" if v is True else "f" if v is False else str(v)
+
+
+def _reading(exc, want):
+    """-> "<SQLSTATE> (<label>)" when the primary message is the wanted refusal,
+    "<SQLSTATE>: <message>" for any other error."""
+    if not exc.sqlstate:
+        return "no refusal: the connection was lost"
+    msg = exc.diag.message_primary or ""
+    regex, label = want
+    return f"{exc.sqlstate} ({label})" if re.search(regex, msg) else f"{exc.sqlstate}: {msg}"
+
+
+def _refusal(dsn, sql, want):
+    try:
+        with psycopg.connect(dsn, autocommit=True) as conn:
+            conn.execute(sql).fetchall()
+    except psycopg.Error as exc:
+        return _reading(exc, want)
+    return "no error"
+
+
+def _refusal_or_hang(dsn, sql, fifo):
+    """-> the refusal reading, or HANG when the statement is still running after 5s;
+    a hung backend is released by opening the FIFO's write end."""
+    result = {}
+
+    def run():
+        result["v"] = _refusal(dsn, sql, FIFO_REFUSAL)
+
+    t = threading.Thread(target=run, daemon=True)
+    t.start()
+    t.join(5)
+    if t.is_alive():
+        try:
+            os.close(os.open(fifo, os.O_WRONLY | os.O_NONBLOCK))
+        except OSError:
+            pass
+        t.join(10)
+        return "HANG"
+    return result.get("v", query_error("the statement left no outcome"))
+
+
+def _crashes(cluster):
+    try:
+        return (cluster.datadir / "server.log").read_text(errors="replace").count("terminated by signal")
+    except OSError as exc:
+        return query_error(f"crash count unreadable: {exc}")
+
+
+def _survived(dsn, cluster, before):
+    """-> "1" when the server answers and no backend died since `before`, read just
+    before the arm ran; after a crash, waits out the recovery so the arms after it
+    are not collateral."""
+    up, now = _value(dsn, "SELECT 1"), _crashes(cluster)
+    if not isinstance(before, int) or not isinstance(now, int):
+        return query_error(f"crash count unreadable: {before} / {now}")
+    if up == "1" and now == before:
+        return "1"
+    for _ in range(60):
+        if _value(dsn, "SELECT 1") == "1":
+            break
+        time.sleep(0.5)
+    return f"no ([{up if up == '1' else ''}], {now - before} backend crashes)"
+
+
+def _fixtures_present():
+    return all((WHM / f).is_file() for f in NEEDED)
+
+
+@pytest.fixture(scope="module")
+def imf(pgc_cluster, server_dir):
+    if not _fixtures_present():
+        yield None
+        return
+    dest = server_dir / "malformed"
+    shutil.copytree(WHM, dest)
+    for p in [dest, *dest.rglob("*")]:
+        p.chmod(0o777 if p.is_dir() else 0o666)
+    fifodir = server_dir / "fifo"
+    fifodir.mkdir()
+    fifodir.chmod(0o777)
+    fifo = fifodir / "x"
+    os.mkfifo(fifo)
+    os.chmod(fifo, 0o666)
+    mlroot = fifodir / "tbl"
+    (mlroot / "db" / "t" / "metadata").mkdir(parents=True)
+    for p in [mlroot, *mlroot.rglob("*")]:
+        p.chmod(0o777)
+    mlfifo = mlroot / "db" / "t" / "metadata" / "ml.avro"
+    os.mkfifo(mlfifo)
+    os.chmod(mlfifo, 0o666)
+    mdjson = mlroot / "db" / "t" / "metadata" / "t.metadata.json"
+    mdjson.write_text(json.dumps({
+        "format-version": 2, "location": f"file://{mlroot}/db/t",
+        "current-snapshot-id": 4242, "current-schema-id": 0,
+        "schemas": [{"schema-id": 0, "type": "struct", "fields": []}],
+        "snapshots": [{"snapshot-id": 4242, "timestamp-ms": 0, "sequence-number": 1, "schema-id": 0,
+                       "manifest-list": f"file://{mlroot}/db/t/metadata/ml.avro",
+                       "summary": {"operation": "append"}}]}))
+    mdjson.chmod(0o666)
+
+    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:
+        dsn = pgc_cluster.dsn(dbname=DB)
+        with psycopg.connect(dsn, autocommit=True) as c:
+            c.execute("CREATE EXTENSION IF NOT EXISTS pgcolumnar")
+        crashes = lambda: _crashes(pgc_cluster)  # noqa: E731
+        alive = lambda b: _survived(dsn, pgc_cluster, b)  # noqa: E731
+        md = lambda d: dest / d / "db" / "t" / "metadata" / "t.metadata.json"  # noqa: E731
+        scan3 = "AS t(id bigint, region text, amount int)"
+        r = {}
+        b = crashes()
+        r["array refused"] = _refusal(dsn, f"SELECT * FROM pgcolumnar.read_avro_manifest('{dest}/arrayfield/evil.avro')", SCHEMA)
+        r["array up"] = alive(b)
+        r["array control"] = _value(dsn, f"SELECT count(*) FROM pgcolumnar.read_avro_manifest('{dest}/arrayfield/good.avro')")
+        b = crashes()
+        r["np lister"] = _refusal(dsn, f"SELECT * FROM pgcolumnar.iceberg_data_files('{md('nullpath')}')", NOPATH)
+        r["np lister up"] = alive(b)
+        b = crashes()
+        r["np scan"] = _refusal(dsn, f"SELECT * FROM pgcolumnar.iceberg_scan('{md('nullpath')}') AS t(id bigint)", NOPATH)
+        r["np scan up"] = alive(b)
+        r["np control"] = _value(dsn, f"SELECT count(*) FROM pgcolumnar.iceberg_data_files('{md('nullpath_ctl')}')")
+        b = crashes()
+        r["nmp"] = _refusal(dsn, f"SELECT * FROM pgcolumnar.iceberg_data_files('{md('nullmanifestpath')}')", NULLMP)
+        r["nmp up"] = alive(b)
+        b = crashes()
+        for key, sql in (("fifo snapshot", f"SELECT * FROM pgcolumnar.iceberg_current_snapshot('{fifo}')"),
+                         ("fifo lister", f"SELECT * FROM pgcolumnar.iceberg_data_files('{fifo}')"),
+                         ("fifo scan", f"SELECT * FROM pgcolumnar.iceberg_scan('{fifo}') AS t(id bigint)"),
+                         ("fifo manifest", f"SELECT * FROM pgcolumnar.read_avro_manifest('{fifo}')"),
+                         ("fifo list", f"SELECT * FROM pgcolumnar.read_manifest_list('{fifo}')"),
+                         ("fifo parquet", f"SELECT * FROM pgcolumnar.read_parquet('{fifo}') AS t(id bigint)")):
+            r[key] = _refusal_or_hang(dsn, sql, fifo)
+        r["fifo up"] = alive(b)
+        b = crashes()
+        r["fifo ml"] = _refusal_or_hang(dsn, f"SELECT * FROM pgcolumnar.iceberg_data_files('{mdjson}')", mlfifo)
+        r["fifo ml up"] = alive(b)
+        r["seq null"] = _value(dsn, f"SELECT sequence_number IS NULL FROM pgcolumnar.read_manifest_list('{dest}/nullseq_ml/ml.avro')")
+        r["seq ctl"] = _value(dsn, f"SELECT sequence_number FROM pgcolumnar.read_manifest_list('{dest}/nullseq_ml/ml_ctl.avro')")
+        r["minseq null"] = _value(dsn, f"SELECT min_sequence_number IS NULL FROM pgcolumnar.read_manifest_list('{dest}/nullminseq_ml/ml.avro')")
+        r["minseq ctl"] = _value(dsn, f"SELECT min_sequence_number FROM pgcolumnar.read_manifest_list('{dest}/nullminseq_ml/ml_ctl.avro')")
+        b = crashes()
+        r["seq refused"] = _refusal(dsn, f"SELECT count(*) FROM pgcolumnar.iceberg_scan('{md('nullseq')}') {scan3}", NULLSEQ)
+        r["seq up"] = alive(b)
+        r["seq control"] = _value(dsn, f"SELECT count(*) FROM pgcolumnar.iceberg_scan('{md('nullseq_ctl')}') {scan3}")
+        b = crashes()
+        r["dangling"] = _refusal(dsn, f"SELECT amount FROM pgcolumnar.iceberg_scan('{md('danglingschema')}') AS t(amount int)", DANGLING)
+        r["dangling up"] = alive(b)
+        r["dangling control"] = _value(dsn, "SELECT string_agg(amount::text, ',' ORDER BY amount) "
+                                            f"FROM pgcolumnar.iceberg_scan('{md('danglingschema_ctl')}') AS t(amount int)")
+        r["legacy rows"] = _value(dsn, f"SELECT count(*) FROM pgcolumnar.iceberg_scan('{md('legacyschema')}') {scan3}")
+        r["legacy amount"] = _value(dsn, "SELECT string_agg(amount::text, ',' ORDER BY amount) "
+                                         f"FROM pgcolumnar.iceberg_scan('{md('legacyschema')}') {scan3}")
+        yield r
+    finally:
+        with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a:
+            a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)")
+
+
+def _declined(imf, expect):
+    if imf is not None:
+        return False
+    expect.cannot_run("ABSENT_FIXTURE", "the malformed iceberg fixtures are missing")
+    return True
+
+
+def test_the_malformed_iceberg_fixtures_are_present(expect):
+    """The shell skips the whole suite without the committed fixtures, under this name."""
+    if not _fixtures_present():
+        expect.cannot_run("MISSING_DEPENDENCY", "the committed malformed Iceberg fixtures are not in this tree",
+                          name="malformed iceberg fixtures are missing")
+        return
+    expect.at_least((WHM / NEEDED[0]).stat().st_size, 1,
+                    "premise: the committed malformed fixtures are not empty")
+
+
+def test_python_reads_the_malformed_metadata(expect):
+    """The shell's other skip name: it needs python3 with json."""
+    try:
+        import json as _json
+    except ImportError:
+        expect.cannot_run("MISSING_DEPENDENCY", "json is not importable", name="python3 is needed")
+        return
+    expect.at_least(len(_json.loads((WHM / NEEDED[4]).read_text())), 1,
+                    "premise: the dangling-schema metadata parses as JSON")
+
+
+def test_an_array_valued_fields_element_is_refused(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["array refused"], "XX001 (malformed schema)",
+                "an array-valued schema fields element is refused, not crashed (XX001)")
+
+
+def test_the_backend_survives_the_malformed_schema(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["array up"], "1", "backend still up after the malformed-schema refusal")
+
+
+def test_a_well_formed_field_object_still_decodes(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["array control"], "0", "control: a well-formed field object still decodes (0 entries)")
+
+
+def test_the_lister_refuses_a_null_data_file_path(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["np lister"], "XX001 (no data file path)",
+                "a NULL data-file path is refused in iceberg_data_files (XX001, not a crash)")
+
+
+def test_the_backend_survives_the_lister_refusal(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["np lister up"], "1", "backend still up after the lister refusal")
+
+
+def test_the_scan_refuses_a_null_data_file_path(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["np scan"], "XX001 (no data file path)",
+                "a NULL data-file path is refused in iceberg_scan (XX001, not a crash)")
+
+
+def test_the_backend_survives_the_scan_refusal(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["np scan up"], "1", "backend still up after the scan refusal")
+
+
+def test_a_present_file_path_lists_one_data_file(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["np control"], "1", "control: a present file_path lists one data file")
+
+
+def test_a_null_manifest_path_is_refused(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["nmp"], "XX001 (null manifest path)", "a NULL manifest_path is refused, not a crash (XX001)")
+
+
+def test_the_backend_survives_the_null_manifest_path(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["nmp up"], "1", "backend still up after the null-manifest_path refusal")
+
+
+def test_iceberg_current_snapshot_refuses_a_fifo(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo snapshot"], "XX001 (not a regular file)",
+                "iceberg_current_snapshot on a FIFO is refused, not a hang (XX001)")
+
+
+def test_iceberg_data_files_refuses_a_fifo(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo lister"], "XX001 (not a regular file)",
+                "iceberg_data_files on a FIFO is refused, not a hang (XX001)")
+
+
+def test_iceberg_scan_refuses_a_fifo(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo scan"], "XX001 (not a regular file)", "iceberg_scan on a FIFO is refused, not a hang (XX001)")
+
+
+def test_read_avro_manifest_refuses_a_fifo_as_iceberg_input(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo manifest"], "XX001 (not a regular file)",
+                "read_avro_manifest on a FIFO is refused, not a hang (XX001)")
+
+
+def test_read_manifest_list_refuses_a_fifo(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo list"], "XX001 (not a regular file)",
+                "read_manifest_list on a FIFO is refused, not a hang (XX001)")
+
+
+def test_read_parquet_refuses_a_fifo_beside_iceberg(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo parquet"], "XX001 (not a regular file)", "read_parquet on a FIFO is refused, not a hang (XX001)")
+
+
+def test_the_backend_survives_the_fifo_refusals(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo up"], "1", "backend still up after the FIFO refusals")
+
+
+def test_a_manifest_list_that_is_a_fifo_is_refused(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo ml"], "XX001 (not a regular file)",
+                "a manifest-list that is a FIFO is refused, not a hang (XX001)")
+
+
+def test_the_backend_survives_the_manifest_list_fifo(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["fifo ml up"], "1", "backend still up after the manifest-list FIFO refusal")
+
+
+def test_a_null_manifest_list_sequence_number_reads_null(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["seq null"], "t", "a null manifest-list sequence_number decodes as NULL, not 0")
+
+
+def test_a_present_manifest_list_sequence_number_reads_its_value(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["seq ctl"], "5", "control: a present manifest-list sequence_number decodes to its value")
+
+
+def test_a_null_min_sequence_number_reads_null(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["minseq null"], "t", "a null manifest-list min_sequence_number decodes as NULL, not 0 (#686)")
+
+
+def test_a_present_min_sequence_number_reads_its_value(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["minseq ctl"], "7", "control: a present min_sequence_number decodes to its value (#686)")
+
+
+def test_the_scan_refuses_a_null_sequence_an_added_entry_inherits(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["seq refused"], "XX001 (null sequence to inherit)",
+                "iceberg_scan refuses a null manifest seq an ADDED entry must inherit (XX001, not a silent wrong row)")
+
+
+def test_the_backend_survives_the_null_sequence_refusal(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["seq up"], "1", "backend still up after the null-sequence refusal")
+
+
+def test_a_concrete_manifest_sequence_scans_every_row(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["seq control"], "5",
+                "control: a concrete manifest seq scans all 5 rows (seq-3 delete is older than seq-5 data)")
+
+
+def test_a_dangling_current_schema_id_is_refused(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["dangling"], "XX001 (dangling schema id)",
+                "a dangling current-schema-id is refused (XX001), not silently stale-read")
+
+
+def test_the_backend_survives_the_dangling_schema_refusal(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["dangling up"], "1", "backend still up after the dangling-schema refusal")
+
+
+def test_a_resolvable_current_schema_id_reads_amount(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["dangling control"], "10,20,30,40,50",
+                "control: a resolvable current-schema-id reads the correct amount column")
+
+
+def test_a_legacy_only_schema_table_still_reads(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["legacy rows"], "5", "control: a legacy-only-schema table still reads (5 rows)")
+
+
+def test_the_legacy_table_binds_amount(imf, expect):
+    if _declined(imf, expect):
+        return
+    expect.text(imf["legacy amount"], "10,20,30,40,50", "control: the legacy table binds amount correctly (10..50)")
diff --git a/test/pytest/unported_suites.txt b/test/pytest/unported_suites.txt
index 94ddacc08..03d9da374 100644
--- a/test/pytest/unported_suites.txt
+++ b/test/pytest/unported_suites.txt
@@ -21,7 +21,6 @@ harness_selftest
 hilbert_curve
 iceberg_deletes
 iceberg_dv
-iceberg_malformed
 iceberg_name_mapping
 iceberg_objstore
 iceberg_rest