test: iceberg_data_files refuses a delete entry inside a data manifest (#1341 item 49)

test/iceberg_data_files.sh

@@ -179,6 +179,87 @@ check "a snapshot with a delete manifest is refused (feature_not_supported 0A000
 	"$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_data_files('$DT/metadata/v1.metadata.json')")" "0A000"
 check "backend still up after the delete refusal" "$(q 'SELECT 1')" "1"
 
+# ---- a delete ENTRY inside a data manifest is refused too (#1341 item 49) --
+# The arm above stops at the manifest LIST. A data manifest (content 0 in the
+# list) can still hold an entry the lister must not return: a position or
+# equality delete file (entry content 1 or 2), or a removed file (status 2,
+# DELETED). Returning a removed file would read rows a commit took out. Each
+# variant below is the same crafted table with one (status, content) pair; the
+# (1, 0) control is listed, so a refusal is the entry's doing, not a fault in
+# the crafted Avro.
+entry_tbl() {  # $1 name, $2 status, $3 content; echoes the metadata path
+	local d="$PGC_WORKDIR/entry_$1/db/t"
+	mkdir -p "$d/metadata"
+	python3 - "$d" "$2" "$3" <<'PY'
+import json, sys, os
+D, STATUS, CONTENT = sys.argv[1], int(sys.argv[2]), int(sys.argv[3])
+LOC = "file:///tmp/pgc_ice_wh/db/t"
+def zz(n):
+    u = (n << 1) ^ (n >> 63) if n < 0 else (n << 1)
+    out = bytearray()
+    while True:
+        b = u & 0x7f; u >>= 7
+        out.append(b | 0x80 if u else b)
+        if not u: break
+    return bytes(out)
+def s(bs): return zz(len(bs)) + bs
+def ocf(schema, recs):
+    sync = b"\x00" * 16
+    meta = zz(2) + s(b"avro.schema") + s(json.dumps(schema).encode()) + s(b"avro.codec") + s(b"null") + zz(0)
+    out = b"Obj\x01" + meta + sync
+    for r in recs: out += zz(1) + zz(len(r)) + r + sync
+    return out
+entry = {"type": "record", "name": "manifest_entry", "fields": [
+    {"name": "status", "type": "int"},
+    {"name": "sequence_number", "type": ["null", "long"]},
+    {"name": "data_file", "type": {"type": "record", "name": "df", "fields": [
+        {"name": "content", "type": "int"},
+        {"name": "file_path", "type": "string"},
+        {"name": "file_format", "type": "string"},
+        {"name": "record_count", "type": "long"},
+        {"name": "file_size_in_bytes", "type": "long"}]}}]}
+ent = zz(STATUS) + (zz(1) + zz(1)) + zz(CONTENT) + s(f"{LOC}/data/f.parquet".encode()) + \
+      s(b"PARQUET") + zz(5) + zz(1000)
+open(os.path.join(D, "metadata", "dm.avro"), "wb").write(ocf(entry, [ent]))
+mfile = {"type": "record", "name": "manifest_file", "fields": [
+    {"name": "manifest_path", "type": "string"},
+    {"name": "manifest_length", "type": "long"},
+    {"name": "partition_spec_id", "type": "int"},
+    {"name": "content", "type": "int"},
+    {"name": "sequence_number", "type": ["null", "long"]},
+    {"name": "min_sequence_number", "type": "long"},
+    {"name": "added_snapshot_id", "type": "long"}]}
+mf = s(f"{LOC}/metadata/dm.avro".encode()) + zz(1000) + zz(0) + zz(0) + \
+     (zz(1) + zz(1)) + zz(1) + zz(1)        # content 0: a DATA manifest
+open(os.path.join(D, "metadata", "ml.avro"), "wb").write(ocf(mfile, [mf]))
+md = {"format-version": 2, "location": LOC, "current-snapshot-id": 1,
+      "current-schema-id": 0,
+      "schemas": [{"schema-id": 0, "type": "struct", "fields": [
+          {"id": 1, "name": "id", "required": False, "type": "long"}]}],
+      "snapshots": [{"snapshot-id": 1, "sequence-number": 1, "timestamp-ms": 0,
+                     "manifest-list": f"{LOC}/metadata/ml.avro",
+                     "summary": {"operation": "append"}, "schema-id": 0}]}
+open(os.path.join(D, "metadata", "v1.metadata.json"), "w").write(json.dumps(md))
+PY
+	chmod -R a+rX "$PGC_WORKDIR/entry_$1"
+	echo "$d/metadata/v1.metadata.json"
+}
+EC="$(entry_tbl ctl 1 0)"
+check "control: a data manifest with one added data entry lists that one file" \
+	"$(q "SELECT count(*) FROM pgcolumnar.iceberg_data_files('$EC')")" "1"
+# status 0 is EXISTING: a live file in a manifest that a later commit rewrote or
+# merged. A lister that refused every entry but ADDED would pass every other arm
+# here and fail an ordinary table after its second append.
+check "an EXISTING (status 0) data entry is listed, as an ADDED one is" \
+	"$(q "SELECT count(*) FROM pgcolumnar.iceberg_data_files('$(entry_tbl kept 0 0)')")" "1"
+check "a position-delete entry in a data manifest is refused (0A000)" \
+	"$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_data_files('$(entry_tbl pos 1 1)')")" "0A000"
+check "an equality-delete entry in a data manifest is refused (0A000)" \
+	"$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_data_files('$(entry_tbl eq 1 2)')")" "0A000"
+check "a DELETED (status 2) data entry is refused, not listed (0A000)" \
+	"$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_data_files('$(entry_tbl gone 2 0)')")" "0A000"
+check "backend still up after the entry refusals" "$(q 'SELECT 1')" "1"
+
 # ---- no current snapshot: zero rows ----------------------------------------
 NOSNAP="$PGC_WORKDIR/no_current.metadata.json"
 python3 - "$MD" "$NOSNAP" <<'PY'

test/pytest/TESTS.md

@@ -177,6 +177,7 @@ behaviour, the source of that number is named.
 - [129. test_savepoint_flush.py: a savepoint rolled back after a flush](#129-test_savepoint_flushpy-a-savepoint-rolled-back-after-a-flush)
 - [130. test_rls_query_paths.py: row-level security holds on every read path](#130-test_rls_query_pathspy-row-level-security-holds-on-every-read-path)
 - [131. test_type_matrix.py: types the differential matrix never stored answer as heap](#131-test_type_matrixpy-types-the-differential-matrix-never-stored-answer-as-heap)
+- [132. test_iceberg_data_files.py: a delete entry in a data manifest is refused](#132-test_iceberg_data_filespy-a-delete-entry-in-a-data-manifest-is-refused)
 
 ## 1. How to read a test in here
 
@@ -7196,3 +7197,15 @@ The pytest twin of `test/type_matrix.sh` (#1341), every check by the shell's nam
 | test | what it holds |
 | --- | --- |
 | `test_types_the_matrix_never_stored_answer_as_heap` | the 21 columns against heap: rows, counts, sixteen ranges, the three non-btree matches, and the special values |
+
+## 132. test_iceberg_data_files.py: a delete entry in a data manifest is refused
+
+A partial pytest twin of `test/iceberg_data_files.sh` (#1341 item 49). It ports only the arms that item added, by the shell's names; the rest of the shell suite is declared INCOMPLETE in `test_compare_to_bash.py`. The lister refuses a delete manifest in the manifest list without opening it, but a data manifest can still hold an entry it must not list: a position or equality delete file (entry content 1 or 2), or a removed file (status 2, DELETED). Each table is crafted Avro, the same one-entry data manifest with one (status, content) pair, because no available writer emits these shapes. The (1, 0) control must be listed, so a refusal of the others comes from the entry and not from a fault in the crafted Avro. An EXISTING (0, 0) entry must be listed too, so a lister that keeps only ADDED entries fails.
+
+### Every test
+
+| test | what it holds |
+| --- | --- |
+| `test_the_control_entry_is_listed` | a data manifest with one added data entry lists that one file |
+| `test_an_existing_entry_is_listed` | an EXISTING (status 0) entry is listed too, so a lister that keeps only ADDED entries fails |
+| `test_a_delete_entry_is_refused` | a position-delete entry, an equality-delete entry and a DELETED entry are each refused with 0A000, and the backend stays up |

test/pytest/expected_tests.txt

@@ -729,4 +729,6 @@ guard_tests 438
 # test_native_agg.py, by collection over the cluster files on this tree.
 # cluster_tests 603 -> 604 for #1437's test_a_gating_qual_over_the_join_matches_heap in
 # test_join_runtime_filter.py, by collection over the cluster files on this tree.
-cluster_tests 604
+# cluster_tests 604 -> 609 for #1341's test_iceberg_data_files.py, five tests (two
+# listed controls and three parametrized refusals), by collection over the cluster files.
+cluster_tests 609

test/pytest/test_compare_to_bash.py

@@ -174,6 +174,13 @@
 # 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 49 ported the arms it added: a delete or removed entry inside a
+    # data manifest is refused, with a control that is listed. The rest of the
+    # shell suite (the pyiceberg oracle, path rebasing and its escapes, the
+    # delete-manifest refusal, no snapshot, the privilege refusal) is not.
+    "iceberg_data_files": "the delete-entry refusals and their control are ported; "
+                          "the oracle, rebasing, path-escape, delete-manifest, "
+                          "no-snapshot and privilege arms are not",
     # #1341's megabyte values: every arm is ported except the opt-in chunk over 1 GB
     # (PGC_RUN_OVER_1GB=1), which needs several GB of memory and which this layer,
     # having no SKIP verdict for an arm a run chooses not to make, does not carry.

test/pytest/test_iceberg_data_files.py

@@ -0,0 +1,140 @@
+"""`iceberg_data_files` refuses a delete entry inside a data manifest (#1341 item 49).
+
+The lister refuses a snapshot with deletes in two places. A delete manifest in the
+manifest LIST is refused without being opened. A DATA manifest (content 0 in the
+list) can still hold an entry that must not be listed: a position or equality delete
+file (entry content 1 or 2), or a removed file (status 2, DELETED). Listing a removed
+file would hand a caller rows that a commit took out.
+
+No available writer emits these shapes, so each table here is crafted Avro: the same
+one-entry data manifest with one (status, content) pair. The (1, 0) control is listed,
+so a refusal of the others is the entry's doing, not a fault in the crafted Avro.
+An EXISTING (0, 0) entry is listed too, so a lister that keeps only ADDED entries fails.
+
+This ports only the entry arms of `test/iceberg_data_files.sh`; the rest of that suite
+is declared INCOMPLETE in `test_compare_to_bash.py`.
+"""
+import json
+import os
+import stat
+
+import pytest
+
+LOC = "file:///tmp/pgc_ice_wh/db/t"     # the recorded root the reader rebases from
+
+
+def _zz(n):
+    u = (n << 1) ^ (n >> 63) if n < 0 else (n << 1)
+    out = bytearray()
+    while True:
+        b = u & 0x7f
+        u >>= 7
+        out.append(b | 0x80 if u else b)
+        if not u:
+            return bytes(out)
+
+
+def _s(bs):
+    return _zz(len(bs)) + bs
+
+
+def _ocf(schema, recs):
+    sync = b"\x00" * 16
+    meta = (_zz(2) + _s(b"avro.schema") + _s(json.dumps(schema).encode())
+            + _s(b"avro.codec") + _s(b"null") + _zz(0))
+    out = b"Obj\x01" + meta + sync
+    for r in recs:
+        out += _zz(1) + _zz(len(r)) + r + sync
+    return out
+
+
+ENTRY = {"type": "record", "name": "manifest_entry", "fields": [
+    {"name": "status", "type": "int"},
+    {"name": "sequence_number", "type": ["null", "long"]},
+    {"name": "data_file", "type": {"type": "record", "name": "df", "fields": [
+        {"name": "content", "type": "int"},
+        {"name": "file_path", "type": "string"},
+        {"name": "file_format", "type": "string"},
+        {"name": "record_count", "type": "long"},
+        {"name": "file_size_in_bytes", "type": "long"}]}}]}
+
+MFILE = {"type": "record", "name": "manifest_file", "fields": [
+    {"name": "manifest_path", "type": "string"},
+    {"name": "manifest_length", "type": "long"},
+    {"name": "partition_spec_id", "type": "int"},
+    {"name": "content", "type": "int"},
+    {"name": "sequence_number", "type": ["null", "long"]},
+    {"name": "min_sequence_number", "type": "long"},
+    {"name": "added_snapshot_id", "type": "long"}]}
+
+
+def _open_to_the_server(root):
+    """Make `root`, its parents and everything under it readable by the server user."""
+    for parent in [root] + list(root.parents):
+        try:
+            os.chmod(parent, os.stat(parent).st_mode | stat.S_IROTH | stat.S_IXOTH)
+        except (PermissionError, FileNotFoundError):
+            break
+    for where, _dirs, files in os.walk(root):
+        os.chmod(where, 0o755)
+        for f in files:
+            os.chmod(os.path.join(where, f), 0o644)
+
+
+def _entry_table(root, status, content):
+    """-> the metadata.json path of a table whose one data manifest holds one entry."""
+    md = root / "db" / "t" / "metadata"
+    md.mkdir(parents=True)
+    ent = (_zz(status) + (_zz(1) + _zz(1)) + _zz(content)
+           + _s(f"{LOC}/data/f.parquet".encode()) + _s(b"PARQUET") + _zz(5) + _zz(1000))
+    (md / "dm.avro").write_bytes(_ocf(ENTRY, [ent]))
+    mf = (_s(f"{LOC}/metadata/dm.avro".encode()) + _zz(1000) + _zz(0)
+          + _zz(0)                          # content 0: a DATA manifest
+          + (_zz(1) + _zz(1)) + _zz(1) + _zz(1))
+    (md / "ml.avro").write_bytes(_ocf(MFILE, [mf]))
+    meta = {"format-version": 2, "location": LOC, "current-snapshot-id": 1,
+            "current-schema-id": 0,
+            "schemas": [{"schema-id": 0, "type": "struct", "fields": [
+                {"id": 1, "name": "id", "required": False, "type": "long"}]}],
+            "snapshots": [{"snapshot-id": 1, "sequence-number": 1, "timestamp-ms": 0,
+                           "manifest-list": f"{LOC}/metadata/ml.avro",
+                           "summary": {"operation": "append"}, "schema-id": 0}]}
+    (md / "v1.metadata.json").write_text(json.dumps(meta))
+    _open_to_the_server(root)
+    return md / "v1.metadata.json"
+
+
+def _listed(conn, path):
+    with conn.cursor() as cur:
+        cur.execute("SELECT count(*) FROM pgcolumnar.iceberg_data_files(%s)", (str(path),))
+        return cur.fetchone()[0]
+
+
+def test_the_control_entry_is_listed(pgc_conn, tmp_path, expect):
+    """The crafted Avro reads: one added data entry lists one file."""
+    expect.num(_listed(pgc_conn, _entry_table(tmp_path / "ctl", 1, 0)), 1,
+               "control: a data manifest with one added data entry lists that one file")
+
+
+def test_an_existing_entry_is_listed(pgc_conn, tmp_path, expect):
+    """Status 0 is EXISTING: a live file in a manifest a later commit rewrote or
+    merged. A lister that refused every entry but ADDED would pass the other arms
+    and fail an ordinary table after its second append."""
+    expect.num(_listed(pgc_conn, _entry_table(tmp_path / "kept", 0, 0)), 1,
+               "an EXISTING (status 0) data entry is listed, as an ADDED one is")
+
+
+@pytest.mark.parametrize("status, content, name", [
+    (1, 1, "a position-delete entry in a data manifest is refused (0A000)"),
+    (1, 2, "an equality-delete entry in a data manifest is refused (0A000)"),
+    (2, 0, "a DELETED (status 2) data entry is refused, not listed (0A000)"),
+])
+def test_a_delete_entry_is_refused(pgc_conn, tmp_path, expect, status, content, name):
+    import psycopg
+
+    path = _entry_table(tmp_path / "t", status, content)
+    with pytest.raises(psycopg.Error) as err:
+        pgc_conn.execute("SELECT * FROM pgcolumnar.iceberg_data_files(%s)", (str(path),))
+    expect.sqlstate(err.value, "0A000", name)
+    expect.num(int(pgc_conn.execute("SELECT 1").fetchone()[0]), 1,
+               "backend still up after the entry refusals")