From a3703688359314afca00415287ea8aea6b28d3ff Mon Sep 17 00:00:00 2001 From: "Joshua D. Drake" Date: Sat, 17 Oct 2026 17:05:18 +0000 Subject: [PATCH 1/2] test: iceberg_data_files's twin is a port by name (#1341) The twin ported only the delete-entry arms #1341 item 49 added (5 tests). It now runs the whole shell suite by name, one check per test (21 tests): pyiceberg's committed oracle, the rebasing from the recorded root, the four escapes refused with 22023 (foreign, '..', sibling prefix, symlink), the delete manifest and delete entries refused with 0A000, no current snapshot, and the 42501 privilege refusal. The shell's two skip names are carried as the same declines. Every reading goes through query_error(). Co-Authored-By: Claude Opus 5.5 --- test/pytest/TESTS.md | 23 +- test/pytest/expected_tests.txt | 4 +- test/pytest/test_compare_to_bash.py | 8 +- test/pytest/test_iceberg_data_files.py | 406 +++++++++++++++++++++---- 4 files changed, 363 insertions(+), 78 deletions(-) diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md index 20d5ba327..43e967ba8 100644 --- a/test/pytest/TESTS.md +++ b/test/pytest/TESTS.md @@ -177,7 +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) +- [132. test_iceberg_data_files.py: an Iceberg table's live data files, listed inside its location](#132-test_iceberg_data_filespy-an-iceberg-tables-live-data-files-listed-inside-its-location) - [133. test_native_skip.py: a type binary-coercible to its opclass's type prunes](#133-test_native_skippy-a-type-binary-coercible-to-its-opclasss-type-prunes) - [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) @@ -7337,17 +7337,28 @@ The pytest twin of `test/type_matrix.sh` (#1341), every check by the shell's nam | --- | --- | | `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 +## 132. test_iceberg_data_files.py: an Iceberg table's live data files, listed inside its location -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. +The port of `test/iceberg_data_files.sh`, by name, one check per test (#1341). The `ice` module fixture copies the committed warehouse away from its recorded root (`/tmp/pgc_ice_wh`), so rebasing is exercised for real. It stages every tampered and crafted table once and records every reading. + +- **The oracle is pyiceberg's own view,** committed as `expected_files.json`, so pyiceberg is not needed at test time. +- **A manifest list outside the table's location is refused with 22023, by four routes:** a foreign absolute path, a `..` traversal, a sibling sharing the byte prefix, and a symlink under the location that targets a file outside it. XX001 would mean the backend opened the file. +- **Deletes are refused, not ignored:** a delete manifest in the list, and a delete or removed entry inside a data manifest (#1341 item 49). Each table is crafted Avro, because no available writer emits these shapes. The added and the EXISTING entry must be listed, so a refusal comes from the entry, and a lister that keeps only ADDED entries fails. +- **The privilege** is read under `SET ROLE` to a role without `pg_read_server_files`. + +The shell's two skip names, "iceberg warehouse fixture is missing" and "python3 (json, zlib) is needed", are carried as the same declines. ### 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_the_warehouse_fixture_is_present`, `test_python_json_and_zlib_are_available` | the shell's skip names; when present, the oracle lists data files and zlib round-trips | +| `test_the_data_files_match_pyicebergs_oracle`, `test_every_data_file_is_parquet` | the live data files, by record count, region and file name, all PARQUET | +| `test_returned_paths_are_rebased_under_the_actual_location`, `test_no_returned_path_leaks_the_recorded_root` | the rebasing | +| `test_a_manifest_list_outside_the_location_is_refused`, `test_the_backend_survives_the_traversal_attempts` | the four escapes, refused with 22023 | +| `test_a_snapshot_with_a_delete_manifest_is_refused`, `test_the_backend_survives_the_delete_refusal` | a delete manifest in the list, refused with 0A000 | +| `test_the_control_entry_is_listed`, `test_an_existing_entry_is_listed`, `test_a_delete_entry_is_refused`, `test_the_backend_survives_the_entry_refusals` | an added and an existing entry are listed; a position-delete, an equality-delete and a DELETED entry are refused with 0A000 | +| `test_a_table_with_no_current_snapshot_yields_no_data_files`, `test_an_unprivileged_role_is_refused` | no current snapshot lists nothing; a caller without `pg_read_server_files` gets 42501 | ## 133. test_native_skip.py: a type binary-coercible to its opclass's type prunes diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt index 6140302d9..ab9931b27 100644 --- a/test/pytest/expected_tests.txt +++ b/test/pytest/expected_tests.txt @@ -850,4 +850,6 @@ guard_tests 442 # nineteen tests, one per check, by collection over the cluster files on this tree. # cluster_tests 966 -> 990 for #1341, test_native_format.py ported by name, one check per # test (1 -> 25 tests), by collection over the cluster files. -cluster_tests 990 +# cluster_tests 990 -> 1006 for #1341, test_iceberg_data_files.py ported by name, one check +# per test (5 -> 21 tests), by collection over the cluster files. +cluster_tests 1006 diff --git a/test/pytest/test_compare_to_bash.py b/test/pytest/test_compare_to_bash.py index ee645d5f1..6bc3760e3 100644 --- a/test/pytest/test_compare_to_bash.py +++ b/test/pytest/test_compare_to_bash.py @@ -88,6 +88,7 @@ COMPLETE = ["advisory_lock_class", "range_pruning", "native_format", + "iceberg_data_files", "alter_table_scans", "fdw_plan_privilege", "am_contracts", @@ -241,13 +242,6 @@ "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", - # #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. diff --git a/test/pytest/test_iceberg_data_files.py b/test/pytest/test_iceberg_data_files.py index c79a76a6e..fb3ddfb53 100644 --- a/test/pytest/test_iceberg_data_files.py +++ b/test/pytest/test_iceberg_data_files.py @@ -1,27 +1,50 @@ -"""`iceberg_data_files` refuses a delete entry inside a data manifest (#1341 item 49). +"""`iceberg_data_files` lists an Iceberg table's live data files, inside its location (#388). -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. +The resolver chains a real Iceberg table (written by pyiceberg, committed under +test/fixtures/iceberg/warehouse/) from its metadata.json through the manifest list and +manifests to the live data files at the current snapshot. The answer is compared with +pyiceberg's own view, `expected_files.json`. The warehouse is committed at a recorded +root (/tmp/pgc_ice_wh) and copied elsewhere at test time, so the rebasing from the +recorded location to the actual one is exercised for real. -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. +What it refuses: +- a manifest list outside the table's location, by four routes (a foreign absolute + path, a `..` traversal, a sibling that shares the byte prefix, and a symlink under + the location that targets a file outside it), with 22023 and never XX001, which + would mean the backend opened the file; +- a snapshot whose manifest list carries a delete manifest (0A000); +- a delete or removed ENTRY inside a data manifest (0A000), with an added and an + existing entry as controls (#1341 item 49); +- a caller without pg_read_server_files (42501). -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`. +No available writer emits the delete shapes, so those tables are crafted Avro. + +THE PORT OF test/iceberg_data_files.sh, BY NAME (#1341). The two names the shell +records only when it skips ("iceberg warehouse fixture is missing", "python3 (json, +zlib) is needed") are carried as the same declines here. Independent of the shell +suite: neither file reads or runs the other. + +ONE CHECK PER TEST (#1339). The `ice` module fixture stages every table once and +records every reading; each test asserts one. A failed query is a `query_error()`. """ import json import os +import pathlib +import shutil import stat +import psycopg import pytest -LOC = "file:///tmp/pgc_ice_wh/db/t" # the recorded root the reader rebases from +from pgc_vacuity import query_error + +FX = pathlib.Path(__file__).resolve().parent.parent / "fixtures" / "iceberg" +WH = FX / "warehouse" +RECORDED = "/tmp/pgc_ice_wh" +LOC = f"file://{RECORDED}/db/t" # the crafted tables' recorded location + +# ---- crafted Avro -------------------------------------------------------------------- def _zz(n): u = (n << 1) ^ (n >> 63) if n < 0 else (n << 1) @@ -67,22 +90,36 @@ def _ocf(schema, recs): {"name": "min_sequence_number", "type": "long"}, {"name": "added_snapshot_id", "type": "long"}]} +# the delete-manifest list: a reduced manifest_file schema carrying exactly the fields +# the reader projects, with content declared so a delete manifest is expressible +DELETE_LIST = {"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": "long"}, + {"name": "min_sequence_number", "type": "long"}, + {"name": "added_snapshot_id", "type": "long"}, + {"name": "added_files_count", "type": "int"}, + {"name": "existing_files_count", "type": "int"}, + {"name": "deleted_files_count", "type": "int"}, + {"name": "added_rows_count", "type": "long"}, + {"name": "existing_rows_count", "type": "long"}, + {"name": "deleted_rows_count", "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 _metadata(manifest_list): + return {"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": manifest_list, + "summary": {"operation": "append"}, "schema-id": 0}]} def _entry_table(root, status, content): - """-> the metadata.json path of a table whose one data manifest holds one entry.""" + """-> the metadata.json 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) @@ -92,49 +129,290 @@ def _entry_table(root, status, content): + _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(_metadata(f"{LOC}/metadata/ml.avro"))) + return md / "v1.metadata.json" + + +def _delete_list_table(root): + """-> the metadata.json of a table whose manifest list holds a delete manifest.""" + md = root / "db" / "t" / "metadata" + md.mkdir(parents=True) + rec = (_s(f"{LOC}/metadata/delete-m0.avro".encode()) + _zz(100) + _zz(0) + + _zz(1) # content 1: the delete manifest + + _zz(1) + _zz(1) + _zz(1) + + _zz(1) + _zz(0) + _zz(0) + _zz(1) + _zz(0) + _zz(0)) + (md / "delete-list.avro").write_bytes(_ocf(DELETE_LIST, [rec])) + meta = _metadata(f"{LOC}/metadata/delete-list.avro") + meta["snapshots"][0]["summary"]["operation"] = "overwrite" (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 _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: + p = os.path.join(where, f) + if not os.path.islink(p): + os.chmod(p, 0o644) + + +# ---- readings ------------------------------------------------------------------------ + +def _one(conn, sql, *params): + try: + return conn.execute(sql, params or None).fetchone()[0] + except psycopg.Error as exc: + return query_error(exc.diag.message_primary) + + +def _sqlstate(conn, path): + """The SQLSTATE listing `path` fails with, or "ok".""" + try: + conn.execute("SELECT * FROM pgcolumnar.iceberg_data_files(%s)", (str(path),)).fetchall() + return "ok" + except psycopg.Error as exc: + return exc.sqlstate or "?" + + +def _alive(pgc_cluster): + """What the shell's `q 'SELECT 1'` reads: a NEW session still gets an answer.""" + try: + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as c: + return c.execute("SELECT 1").fetchone()[0] + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0]) -def test_the_control_entry_is_listed(pgc_conn, tmp_path, expect): +def _tampered(md, name, manifest_list): + """A variant of the real table's metadata whose current snapshot names + `manifest_list`, written beside it so the resolver derives the same location.""" + m = json.loads(md.read_text()) + m["snapshots"][-1]["manifest-list"] = manifest_list + out = md.parent / f"esc_{name}.metadata.json" + out.write_text(json.dumps(m)) + os.chmod(out, 0o644) + return out + + +@pytest.fixture(scope="module") +def ice(pgc_cluster, tmp_path_factory): + got = {"have fixture": (WH / "db" / "events" / "metadata").is_dir()} + if not got["have fixture"]: + yield got + return + root = tmp_path_factory.mktemp("iceberg_data_files") + dest = root / "relocated" + shutil.copytree(WH / "db", dest / "db") + mdfiles = sorted((dest / "db" / "events" / "metadata").glob("*.metadata.json")) + md = mdfiles[-1] + loc = json.loads(md.read_text())["location"].replace("file://", "") + aroot = os.path.realpath(dest / "db" / "events") + + outside = root / "outside_symlink_target" + outside.write_text("outside\n") + (dest / "db" / "events" / "metadata" / "sneaky.avro").symlink_to(outside) + escapes = { + "a": _tampered(md, "a", "file:///etc/passwd"), + "b": _tampered(md, "b", f"file://{loc}/../../../../../../../etc/hostname"), + "c": _tampered(md, "c", f"file://{loc}EVIL/x.avro"), + "d": _tampered(md, "d", f"file://{loc}/metadata/sneaky.avro"), + } + nosnap = root / "no_current.metadata.json" + m = json.loads(md.read_text()) + m.pop("current-snapshot-id", None) + nosnap.write_text(json.dumps(m)) + dlist = _delete_list_table(root / "deltbl") + entries = {k: _entry_table(root / f"entry_{k}", st, ct) + for k, st, ct in (("ctl", 1, 0), ("kept", 0, 0), ("pos", 1, 1), + ("eq", 1, 2), ("gone", 2, 0))} + _open_to_the_server(root) + + oracle = sorted("%d|%s|%s" % (r["record_count"], r["region"], r["file_basename"]) + for r in json.loads((WH / "expected_files.json").read_text())["data_files"]) + + c = psycopg.connect(pgc_cluster.dsn(), autocommit=True) + try: + lst = "SELECT {} FROM pgcolumnar.iceberg_data_files(%s)" + try: + got["decoded"] = "\n".join(r[0] for r in c.execute( + lst.format("record_count || '|' || split_part(partition,'=',2) || '|' || " + "(regexp_match(file_path,'[^/]+$'))[1]") + " ORDER BY 1", + (str(md),)).fetchall()) + except psycopg.Error as exc: + got["decoded"] = query_error(exc.diag.message_primary) + got["oracle"] = "\n".join(oracle) + got["parquet"] = str(_one(c, lst.format("bool_and(file_format = 'PARQUET')"), str(md))) + got["rebased"] = str(_one(c, lst.format("bool_and(file_path LIKE %s)"), + aroot + "/%", str(md))) + got["leaks"] = _one(c, lst.format("count(*)") + " WHERE file_path LIKE %s", + str(md), RECORDED + "/%") + for k, path in escapes.items(): + got[f"escape {k}"] = _sqlstate(c, path) + got["alive traversal"] = _alive(pgc_cluster) + got["delete list"] = _sqlstate(c, dlist) + got["alive delete"] = _alive(pgc_cluster) + got["entry ctl"] = _one(c, lst.format("count(*)"), str(entries["ctl"])) + got["entry kept"] = _one(c, lst.format("count(*)"), str(entries["kept"])) + for k in ("pos", "eq", "gone"): + got[f"entry {k}"] = _sqlstate(c, entries[k]) + got["alive entries"] = _alive(pgc_cluster) + got["no snapshot"] = _one(c, lst.format("count(*)"), str(nosnap)) + try: + c.execute("DROP ROLE IF EXISTS icedf_unpriv") + c.execute("CREATE ROLE icedf_unpriv LOGIN") + c.execute("SET ROLE icedf_unpriv") + got["unprivileged"] = _sqlstate(c, md) + finally: + c.execute("RESET ROLE") + yield got + finally: + try: + c.execute("DROP ROLE IF EXISTS icedf_unpriv") + except psycopg.Error: + pass + c.close() + + +def _need(ice, expect): + """The shell skips the whole suite without the committed warehouse, under one + name; each check here declines under that name instead.""" + if not ice["have fixture"]: + expect.cannot_run("MISSING_DEPENDENCY", "the committed Iceberg warehouse is not " + "in this tree", name="iceberg warehouse fixture is missing") + return False + return True + + +def test_the_warehouse_fixture_is_present(ice, expect): + """The shell's skip name, carried as the same decline.""" + if _need(ice, expect): + expect.at_least(len([r for r in ice["oracle"].split("\n") if r]), 1, + "premise: the oracle lists data files, so matching it is not vacuous") + + +def test_python_json_and_zlib_are_available(expect): + """The shell's other skip name. The crafted Avro here needs json; zlib is what the + shell's helper imports.""" + try: + import zlib # noqa: F401 + except ImportError: + expect.cannot_run("MISSING_DEPENDENCY", "zlib is not importable", + name="python3 (json, zlib) is needed") + return + expect.text(zlib.decompress(zlib.compress(b"pgc")).decode(), "pgc", + "premise: zlib round-trips") + + +# ---- the oracle and the rebasing ------------------------------------------------------- + +def test_the_data_files_match_pyicebergs_oracle(ice, expect): + if _need(ice, expect): + expect.text(ice["decoded"], ice["oracle"], + "data files == pyiceberg oracle (record_count|region|basename)") + + +def test_every_data_file_is_parquet(ice, expect): + if _need(ice, expect): + expect.text(ice["parquet"], "True", "every data file is PARQUET") + + +def test_returned_paths_are_rebased_under_the_actual_location(ice, expect): + if _need(ice, expect): + expect.text(ice["rebased"], "True", "returned paths are rebased under the actual location") + + +def test_no_returned_path_leaks_the_recorded_root(ice, expect): + if _need(ice, expect): + expect.num(ice["leaks"], 0, "no returned path leaks the recorded root") + + +# ---- a path outside the table location is refused, not read ------------------------ +# +# All must be 22023 ("not under / escapes the table location"), and NOT XX001 (bad Avro +# magic): an XX001 means the backend OPENED and read the file. + +# (name, reading): a literal table, so the twin grader reads its names. +ESCAPES = [ + ("a foreign absolute manifest-list is refused (22023)", "escape a"), + ("a .. traversal manifest-list is refused, not read (22023, not XX001)", "escape b"), + ("a sibling-prefix manifest-list is refused (22023)", "escape c"), + ("a symlink under the location targeting outside is refused (22023, not XX001)", + "escape d"), +] + + +@pytest.mark.parametrize("name, reading", ESCAPES) +def test_a_manifest_list_outside_the_location_is_refused(ice, expect, name, reading): + if _need(ice, expect): + expect.text(ice[reading], "22023", name) + + +def test_the_backend_survives_the_traversal_attempts(ice, expect): + if _need(ice, expect): + expect.num(ice["alive traversal"], 1, "backend still up after the traversal attempts") + + +# ---- deletes are refused, not ignored ------------------------------------------------ + +def test_a_snapshot_with_a_delete_manifest_is_refused(ice, expect): + if _need(ice, expect): + expect.text(ice["delete list"], "0A000", + "a snapshot with a delete manifest is refused (feature_not_supported 0A000)") + + +def test_the_backend_survives_the_delete_refusal(ice, expect): + if _need(ice, expect): + expect.num(ice["alive delete"], 1, "backend still up after the delete refusal") + + +def test_the_control_entry_is_listed(ice, 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") + if _need(ice, expect): + expect.num(ice["entry ctl"], 1, + "control: a data manifest with one added data entry lists that one file") + + +def test_an_existing_entry_is_listed(ice, 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 fail an ordinary table after its + second append.""" + if _need(ice, expect): + expect.num(ice["entry kept"], 1, + "an EXISTING (status 0) data entry is listed, as an ADDED one is") + + +ENTRY_REFUSALS = [ + ("a position-delete entry in a data manifest is refused (0A000)", "entry pos"), + ("an equality-delete entry in a data manifest is refused (0A000)", "entry eq"), + ("a DELETED (status 2) data entry is refused, not listed (0A000)", "entry gone"), +] + + +@pytest.mark.parametrize("name, reading", ENTRY_REFUSALS) +def test_a_delete_entry_is_refused(ice, expect, name, reading): + if _need(ice, expect): + expect.text(ice[reading], "0A000", name) + + +def test_the_backend_survives_the_entry_refusals(ice, expect): + if _need(ice, expect): + expect.num(ice["alive entries"], 1, "backend still up after the entry refusals") + + +# ---- no current snapshot, and the privilege ------------------------------------------ + +def test_a_table_with_no_current_snapshot_yields_no_data_files(ice, expect): + if _need(ice, expect): + expect.num(ice["no snapshot"], 0, "a table with no current snapshot yields no data files") + + +def test_an_unprivileged_role_is_refused(ice, expect): + if _need(ice, expect): + expect.text(ice["unprivileged"], "42501", "an unprivileged role is refused (42501)") From 89614394cb221eaad8847b911aed2bd4e6ac7cf0 Mon Sep 17 00:00:00 2001 From: "Joshua D. Drake" Date: Sat, 17 Oct 2026 19:19:26 +0000 Subject: [PATCH 2/2] test: iceberg_data_files's privilege arm reaches the gate, and a crash fails only its own arm (#1341) The unprivileged role had no USAGE on schema pgcolumnar, so its 42501 was "permission denied for schema pgcolumnar": the function never ran, and removing its pg_read_server_files gate left both suites green. Both twins now grant USAGE on the schema and EXECUTE on the function, and drop the role's grants before dropping it. The twin read everything on one connection, so a backend that died on one refusal closed it, and all 20 tests that use the fixture errored with no check name. Each reading now runs in a session of its own, as the shell's psql per statement does. Co-Authored-By: Claude Opus 5.5 --- test/iceberg_data_files.sh | 27 +++++- test/pytest/test_iceberg_data_files.py | 125 ++++++++++++++++--------- 2 files changed, 109 insertions(+), 43 deletions(-) diff --git a/test/iceberg_data_files.sh b/test/iceberg_data_files.sh index 61c8abd54..f8fac4e37 100755 --- a/test/iceberg_data_files.sh +++ b/test/iceberg_data_files.sh @@ -83,6 +83,18 @@ LOC="$(python3 -c "import json;print(json.load(open('$MD'))['location'].replace( # symlink planted there resolves under it -- writing them elsewhere would derive a # wrong actual root and mask the boundary under a "file not found". MDDIR="$(dirname "$MD")" +# After a survival check, wait (up to 60s) until the server accepts connections +# again. A refusal that crashes the backend fails its own arm and the survival check; +# without the wait, whether the NEXT arm also failed on the recovery depended on how +# long the python step before it took (#1341). +recovered() { + local i + for i in $(seq 240); do + [ "$(q 'SELECT 1')" = "1" ] && return 0 + sleep 0.25 + done +} + esc_case() { # $1 payload manifest-list, writes a metadata variant, echoes its path local payload="$1" out="$MDDIR/esc_$2.metadata.json" python3 - "$MD" "$out" "$payload" <<'PY' @@ -113,6 +125,7 @@ ln -sf "$OUTSIDE" "$DEST/db/events/metadata/sneaky.avro" check "a symlink under the location targeting outside is refused (22023, not XX001)" \ "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_data_files('$(esc_case "file://$LOC/metadata/sneaky.avro" d)')")" "22023" check "backend still up after the traversal attempts" "$(q 'SELECT 1')" "1" +recovered # ---- deletes are refused, not ignored -------------------------------------- # pyiceberg 0.11.1 cannot write merge-on-read deletes, so craft the deny input: @@ -178,6 +191,7 @@ PY 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" +recovered # ---- 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 @@ -259,6 +273,7 @@ check "an equality-delete entry in a data manifest is refused (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" +recovered # ---- no current snapshot: zero rows ---------------------------------------- NOSNAP="$PGC_WORKDIR/no_current.metadata.json" @@ -273,7 +288,16 @@ check "a table with no current snapshot yields no data files" \ "$(q "SELECT count(*) FROM pgcolumnar.iceberg_data_files('$NOSNAP')")" "0" # ---- privilege: needs pg_read_server_files --------------------------------- -psql_run "DROP ROLE IF EXISTS icedf_unpriv; CREATE ROLE icedf_unpriv LOGIN;" +# The role gets USAGE on the schema and EXECUTE on the function, so the refusal is +# the function's own gate. Without them the 42501 is "permission denied for schema +# pgcolumnar", the function never runs, and the arm cannot fail on the gate it names. +DROP_UNPRIV="DO \$\$BEGIN IF EXISTS (SELECT FROM pg_roles WHERE rolname = 'icedf_unpriv') + THEN EXECUTE 'DROP OWNED BY icedf_unpriv'; END IF; END\$\$; +DROP ROLE IF EXISTS icedf_unpriv;" +psql_run "$DROP_UNPRIV +CREATE ROLE icedf_unpriv LOGIN; +GRANT USAGE ON SCHEMA pgcolumnar TO icedf_unpriv; +GRANT EXECUTE ON FUNCTION pgcolumnar.iceberg_data_files TO icedf_unpriv;" check "an unprivileged role is refused (42501)" \ "$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U icedf_unpriv -d "$PGC_DB" \ -qtA 2>&1 < None, or the failed reading.""" try: - conn.execute("SELECT * FROM pgcolumnar.iceberg_data_files(%s)", (str(path),)).fetchall() - return "ok" + with psycopg.connect(dsn, autocommit=True) as conn: + for st in stmts: + conn.execute(st) + return None except psycopg.Error as exc: - return exc.sqlstate or "?" + return _failed(exc) + + +# The unprivileged role reaches the function: USAGE on the schema and EXECUTE on the +# function, or its 42501 is the schema's and the pg_read_server_files gate never runs. +DROP_UNPRIV = ("DO $$BEGIN IF EXISTS (SELECT FROM pg_roles WHERE rolname = 'icedf_unpriv') " + "THEN EXECUTE 'DROP OWNED BY icedf_unpriv'; END IF; END$$", + "DROP ROLE IF EXISTS icedf_unpriv") +MAKE_UNPRIV = DROP_UNPRIV + ("CREATE ROLE icedf_unpriv LOGIN", + "GRANT USAGE ON SCHEMA pgcolumnar TO icedf_unpriv", + "GRANT EXECUTE ON FUNCTION pgcolumnar.iceberg_data_files TO icedf_unpriv") def _alive(pgc_cluster): @@ -190,6 +227,18 @@ def _alive(pgc_cluster): return query_error(str(exc).splitlines()[0]) +def _recovered(pgc_cluster, alive): + """`alive` as read. If the backend did not answer, wait (up to 60s) until the + server accepts connections again before the next reading. The shell spends that + time building its next table, so its later arms run on a recovered server; read + at once, they would each fail on the recovery and not on their own refusal.""" + for _ in range(240): + if alive == 1 or _alive(pgc_cluster) == 1: + break + time.sleep(0.25) + return alive + + def _tampered(md, name, manifest_list): """A variant of the real table's metadata whose current snapshot names `manifest_list`, written beside it so the resolver derives the same location.""" @@ -237,47 +286,38 @@ def ice(pgc_cluster, tmp_path_factory): oracle = sorted("%d|%s|%s" % (r["record_count"], r["region"], r["file_basename"]) for r in json.loads((WH / "expected_files.json").read_text())["data_files"]) - c = psycopg.connect(pgc_cluster.dsn(), autocommit=True) + dsn = pgc_cluster.dsn() + lst = "SELECT {} FROM pgcolumnar.iceberg_data_files(%s)" try: - lst = "SELECT {} FROM pgcolumnar.iceberg_data_files(%s)" - try: + with psycopg.connect(dsn, autocommit=True) as c: got["decoded"] = "\n".join(r[0] for r in c.execute( lst.format("record_count || '|' || split_part(partition,'=',2) || '|' || " "(regexp_match(file_path,'[^/]+$'))[1]") + " ORDER BY 1", (str(md),)).fetchall()) - except psycopg.Error as exc: - got["decoded"] = query_error(exc.diag.message_primary) - got["oracle"] = "\n".join(oracle) - got["parquet"] = str(_one(c, lst.format("bool_and(file_format = 'PARQUET')"), str(md))) - got["rebased"] = str(_one(c, lst.format("bool_and(file_path LIKE %s)"), - aroot + "/%", str(md))) - got["leaks"] = _one(c, lst.format("count(*)") + " WHERE file_path LIKE %s", - str(md), RECORDED + "/%") - for k, path in escapes.items(): - got[f"escape {k}"] = _sqlstate(c, path) - got["alive traversal"] = _alive(pgc_cluster) - got["delete list"] = _sqlstate(c, dlist) - got["alive delete"] = _alive(pgc_cluster) - got["entry ctl"] = _one(c, lst.format("count(*)"), str(entries["ctl"])) - got["entry kept"] = _one(c, lst.format("count(*)"), str(entries["kept"])) - for k in ("pos", "eq", "gone"): - got[f"entry {k}"] = _sqlstate(c, entries[k]) - got["alive entries"] = _alive(pgc_cluster) - got["no snapshot"] = _one(c, lst.format("count(*)"), str(nosnap)) - try: - c.execute("DROP ROLE IF EXISTS icedf_unpriv") - c.execute("CREATE ROLE icedf_unpriv LOGIN") - c.execute("SET ROLE icedf_unpriv") - got["unprivileged"] = _sqlstate(c, md) - finally: - c.execute("RESET ROLE") + except psycopg.Error as exc: + got["decoded"] = _failed(exc) + got["oracle"] = "\n".join(oracle) + got["parquet"] = str(_one(dsn, lst.format("bool_and(file_format = 'PARQUET')"), str(md))) + got["rebased"] = str(_one(dsn, lst.format("bool_and(file_path LIKE %s)"), + aroot + "/%", str(md))) + got["leaks"] = _one(dsn, lst.format("count(*)") + " WHERE file_path LIKE %s", + str(md), RECORDED + "/%") + for k, path in escapes.items(): + got[f"escape {k}"] = _sqlstate(dsn, path) + got["alive traversal"] = _recovered(pgc_cluster, _alive(pgc_cluster)) + got["delete list"] = _sqlstate(dsn, dlist) + got["alive delete"] = _recovered(pgc_cluster, _alive(pgc_cluster)) + got["entry ctl"] = _one(dsn, lst.format("count(*)"), str(entries["ctl"])) + got["entry kept"] = _one(dsn, lst.format("count(*)"), str(entries["kept"])) + for k in ("pos", "eq", "gone"): + got[f"entry {k}"] = _sqlstate(dsn, entries[k]) + got["alive entries"] = _recovered(pgc_cluster, _alive(pgc_cluster)) + got["no snapshot"] = _one(dsn, lst.format("count(*)"), str(nosnap)) + got["unprivileged"] = _run(dsn, *MAKE_UNPRIV) or _sqlstate(dsn, md, role="icedf_unpriv") + try: yield got finally: - try: - c.execute("DROP ROLE IF EXISTS icedf_unpriv") - except psycopg.Error: - pass - c.close() + _run(dsn, *DROP_UNPRIV) def _need(ice, expect): @@ -301,6 +341,7 @@ def test_python_json_and_zlib_are_available(expect): """The shell's other skip name. The crafted Avro here needs json; zlib is what the shell's helper imports.""" try: + import json as _json # noqa: F401 import zlib # noqa: F401 except ImportError: expect.cannot_run("MISSING_DEPENDENCY", "zlib is not importable",