diff --git a/test/check_ledger.tsv b/test/check_ledger.tsv index 218e6a82f..66ae61070 100644 --- a/test/check_ledger.tsv +++ b/test/check_ledger.tsv @@ -2549,6 +2549,32 @@ iceberg_catalog iceberg_catalog malformed JSON is rejected (22P02), backend surv iceberg_catalog iceberg_catalog resolves the named current snapshot, not snapshots[0] 15;16;17;18;19 2026-10-21 the resolver returns the first snapshot rather than the one named current iceberg_catalog iceberg_catalog the decoy snapshot is not returned 15;16;17;18;19 2026-10-21 the resolver returns the first snapshot rather than the one named current iceberg_catalog iceberg_catalog the first snapshot has no parent (NULL) 15;16;17;18;19 never - +iceberg_dv iceberg_dv ...and the refusal names the compressed footer 15;16;17;18;19 2026-10-21 columnar_puffin.c: a compressed Puffin footer (flags bit 0) is not refused +iceberg_dv iceberg_dv ...and the refusal names the format version 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a deletion vector in a format-version 2 table is not refused +iceberg_dv iceberg_dv a DV blob declaring a compression codec is refused (XX001) 15;16;17;18;19 2026-10-21 columnar_puffin.c: a deletion-vector blob may declare a compression codec +iceberg_dv iceberg_dv a DV entry without referenced_data_file is refused (XX001) 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a DV without referenced_data_file is not refused (a NULL path reaches strcmp and the backend dies) +iceberg_dv iceberg_dv a DV whose offset+size overflows int64 is refused, not read (XX001) 15;16;17;18;19 2026-10-21 columnar_puffin.c: a DV's offset and size are not bounded by the blob region (a wild read kills the backend) +iceberg_dv iceberg_dv a DV with run containers and a high-bucket position decodes 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a DV applies only when strictly newer than the data (> where the spec says >=) +iceberg_dv iceberg_dv a Puffin path outside the table location is refused (22023) 15;16;17;18;19 2026-10-21 columnar_iceberg.c: ice_open_path does not hold a path under the table location +iceberg_dv iceberg_dv a bitset-container DV decodes 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a DV applies only when strictly newer than the data (> where the spec says >=) +iceberg_dv iceberg_dv a compressed Puffin footer is refused (0A000) 15;16;17;18;19 2026-10-21 columnar_puffin.c: a compressed Puffin footer (flags bit 0) is not refused +iceberg_dv iceberg_dv a corrupted DV CRC is refused (XX001) 15;16;17;18;19 2026-10-21 columnar_puffin.c: a deletion vector's CRC-32 is not checked +iceberg_dv iceberg_dv a deletion vector at an equal sequence number applies (survivors == oracle) 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a DV applies only when strictly newer than the data (> where the spec says >=) +iceberg_dv iceberg_dv a deletion vector in a format-version 2 table is refused (0A000) 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a deletion vector in a format-version 2 table is not refused +iceberg_dv iceberg_dv a deletion vector older than the data does not apply 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a DV applies whatever its sequence number (dv->seq >= d->seq always true) +iceberg_dv iceberg_dv a manifest/footer offset mismatch is refused (XX001) 15;16;17;18;19 2026-10-21 columnar_puffin.c: the Puffin footer's blob need not match the manifest's offset and length (the !found refusal skipped) +iceberg_dv iceberg_dv a record_count that disagrees with the DV cardinality is refused (XX001) 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a manifest record_count that disagrees with the DV cardinality is not refused +iceberg_dv iceberg_dv a run container overflowing its 16-bit range is refused (XX001) 15;16;17;18;19 2026-10-21 columnar_puffin.c: a run container whose start+length passes 0x10000 is not refused +iceberg_dv iceberg_dv a run-optimized DV (cookie 12347) decodes 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a DV applies only when strictly newer than the data (> where the spec says >=) +iceberg_dv iceberg_dv a wrong DV blob magic is refused (XX001) 15;16;17;18;19 2026-10-21 columnar_puffin.c: a deletion vector's magic sequence is not checked +iceberg_dv iceberg_dv an aliased duplicate DV (scheme vs no scheme) is refused (XX001) 15;16;17;18;19 2026-10-21 columnar_iceberg.c: the one-DV-per-data-file check compares paths without stripping the file:// scheme +iceberg_dv iceberg_dv an applicable DV supersedes position-delete files for its data file 15;16;17;18;19 2026-10-21 columnar_iceberg.c: position-delete files apply even when a DV applies (no supersede) +iceberg_dv iceberg_dv backend still up after the DV applications 15;16;17;18;19 never - +iceberg_dv iceberg_dv backend still up after the DV refusals 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a DV without referenced_data_file is not refused (a NULL path reaches strcmp and the backend dies) +iceberg_dv iceberg_dv backend still up after the audit arms 15;16;17;18;19 2026-10-21 columnar_puffin.c: a DV's offset and size are not bounded by the blob region (a wild read kills the backend) +iceberg_dv iceberg_dv supersede is per data file, not global 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a DV applies to every data file, not only the one it references +iceberg_dv iceberg_dv two DV blobs in one Puffin file each apply to their own data file 15;16;17;18;19 2026-10-21 columnar_iceberg.c: a DV applies to every data file, not only the one it references +iceberg_dv iceberg_dv two DV entries for one data file are refused (XX001) 15;16;17;18;19 2026-10-21 columnar_iceberg.c: two DVs for one data file are not refused iceberg_fdw_estimate iceberg_fdw_estimate a filtered scan estimates fewer rows than the whole table (selectivity applied) 15;16;17;18;19 2026-10-21 the Iceberg FDW sets rows directly, so the scan's selectivity never applies iceberg_fdw_estimate iceberg_fdw_estimate backend alive 15;16;17;18;19 never - iceberg_fdw_estimate iceberg_fdw_estimate the FDW estimates a positive row count from the manifests 15;16;17;18;19 never - @@ -2876,6 +2902,18 @@ local_open_race_free local_open_race_free the partition coordinator never stats local_open_race_free local_open_race_free the partition coordinator opens O_NONBLOCK (a FIFO cannot block its open) 15;16;17;18;19 2026-10-21 the partition coordinator opens without O_NONBLOCK local_open_race_free local_open_race_free the race-free opener exists 15;16;17;18;19 never - local_open_race_free local_open_race_free the racy stat-before-open helper is gone 15;16;17;18;19 2026-10-21 the racy PgColumnarRejectNonRegularFile helper comes back +logical_decoding_source logical_decoding_source 100x the rows does not give 100x the internal records 15;16;17;18;19 never - +logical_decoding_source logical_decoding_source CONTROL: the heap table's 50 inserts ARE decoded 15;16;17;18;19 never - +logical_decoding_source logical_decoding_source CONTROL: the second slot decodes the heap table's 5000 inserts 15;16;17;18;19 never - +logical_decoding_source logical_decoding_source a columnar table emits no decodable change 15;16;17;18;19 never - +logical_decoding_source logical_decoding_source although the rows are there 15;16;17;18;19 never - +logical_decoding_source logical_decoding_source and it too emits no decodable change of its own 15;16;17;18;19 never - +logical_decoding_source logical_decoding_source premise: both tables hold the same 50 rows 15;16;17;18;19 never - +logical_decoding_source logical_decoding_source premise: the 100x table really holds 100x the rows 15;16;17;18;19 never - +logical_decoding_source logical_decoding_source premise: the server is running at wal_level=logical 15;16;17;18;19 never - +logical_decoding_source logical_decoding_source premise: the slot decoded something at all 15;16;17;18;19 never - +logical_decoding_source logical_decoding_source premise: the slot exists and is logical 15;16;17;18;19 never - +logical_decoding_source logical_decoding_source the slot carries pgcolumnar's internal catalog instead 15;16;17;18;19 never - maintenance_crash maintenance_crash after a crash with compact committed, the index reads the rows the scan reads 15;16;17;18;19 2026-10-16 pgcolumnar_index_fetch_tuple finds no row (columnar_tableam.c) maintenance_crash maintenance_crash after a crash with compact committed, the table is as compact left it, with the same rows 15;16;17;18;19 never - maintenance_crash maintenance_crash after a crash with compact uncommitted, the index reads the rows the scan reads 15;16;17;18;19 2026-10-16 pgcolumnar_index_fetch_tuple finds no row (columnar_tableam.c) diff --git a/test/check_ledger_budget.txt b/test/check_ledger_budget.txt index 2970fed4d..427c1a912 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 102 +suites_not_covered 100 # # checks_never_observed_red -- A CENSUS. NOT a ceiling, and it must not become # one. @@ -5439,4 +5439,42 @@ suites_not_covered 102 # never 3883 + 1 = 3884 # dated 2282 + 4 = 2286 # 3884 + 2286 == 6170 -checks_never_observed_red 3884 +# #1341, iceberg_dv seeded, with its twin (OffgridwithJD): 3884 -> 3885, and +# suites_not_covered 102 -> 101. The suite had no rows. It enters with 26 checks from +# five-major runs (all green on each of 15-19, shell and twin alike), after two fixes. +# "a manifest/footer offset mismatch is refused (XX001)" read the SQLSTATE alone, and +# several guards here raise XX001: with the offset match removed a later XX001 refused +# the same file, and main's shell passed 26/26. Every refusal now reads its own guard's +# message. And the three "backend still up" checks read SELECT 1, which answers again +# after crash recovery: with the referenced_data_file check removed a backend died on +# the arm, and main passed the survival check. They now also read the server log's +# count of killed backends, and each refusal waits for the server first, so a crash +# reddens its own arm and its section's survival check and nothing else. 25 are dated by +# seventeen mutations, each reddening the same names in both twins. 1 arrives never: +# the survival check after the DV applications, which no mutation crashed. +# +# ledger rows 6170 + 26 = 6196 +# never 3884 + 1 = 3885 +# dated 2286 + 25 = 2311 +# 3885 + 2311 == 6196 +# +# #1341, logical_decoding_source seeded, with its twin (OffgridwithJD): 3885 -> 3897, +# and suites_not_covered 101 -> 100. The suite had no rows. It enters with its 12 +# checks, all green on each of 15-19 in shell and twin alike. The missing-plugin name +# pgc_skip records is now a literal both twins share, but it is a skip, not a check, so +# it has no row: CI has test_decoding and never emits it. After one fix. +# The second slot had no heap control: created after the 5000-row insert, it decoded +# nothing, and "it too emits no decodable change" and "100x the rows does not give 100x +# the internal records" (0 < anything) both passed, 11/11 on main. A heap table now writes +# through that slot in the same window and must decode its 5000 inserts, and the ratio +# check refuses a slot that carried no internal records. Under that harness mutation both +# are red in both twins. No source mutation reddens a check: the claim is that pgcolumnar +# writes nothing logical decoding can read, and a default chunk group of 100 rows still +# keeps the internal records following groups (114 for 50 groups, under the 10x bound). +# 0 are dated. 12 arrive never. +# +# ledger rows 6196 + 12 = 6208 +# never 3885 + 12 = 3897 +# dated 2311 + 0 = 2311 +# 3897 + 2311 == 6208 +checks_never_observed_red 3897 diff --git a/test/iceberg_dv.sh b/test/iceberg_dv.sh index d9df48ca8..5f8d80677 100755 --- a/test/iceberg_dv.sh +++ b/test/iceberg_dv.sh @@ -16,22 +16,71 @@ set -uo pipefail . "$(dirname "${BASH_SOURCE[0]}")/lib.sh" pgc_setup "${1:-/usr/local/pg17/bin/pg_config}" -sqlstate_of() { - env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \ - -d "$PGC_DB" -qtA 2>&1 < the SQLSTATE when SQL is refused with PHRASE in its message; +# "SQLSTATE (another refusal: MESSAGE)" when a different guard refused it; "no +# error" when it ran; nothing when the server could not be asked. An arm used to +# read the SQLSTATE alone, and several guards here raise XX001: with the +# manifest/footer offset match removed, a later XX001 refused the same file and +# that arm still passed. +refusal() { + local out st msg + await_up + out="$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres \ + -d "$PGC_DB" -qtA 2>&1 <&1 < wait, at most 60 s, until the server answers. A refusal arm that kills a +# backend sends the server through crash recovery; without the wait, the next arms' +# readings fail on recovery and red as collateral, more or fewer of them by timing. +# The crash itself is still caught: survived() counts it. +await_up() { + local i + for i in $(seq 1 120); do + [ "$(q 'SELECT 1')" = 1 ] && return 0 + sleep 0.5 + done +} + +# 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 with a backend killed by a refusal arm (no +# referenced_data_file check: a NULL path reached strcmp); the survival checks +# below also require that this count has not moved. +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 + local _up _now + _up="$(q 'SELECT 1')"; _now="$(crashes)" + if ! [[ "$1" =~ ^[0-9]+$ && "$_now" =~ ^[0-9]+$ ]]; then echo "no (crash count unreadable)" + elif [ "$_up" = 1 ] && [ "$_now" = "$1" ]; then echo 1 + else echo "no ([$_up], $(( _now - $1 )) backend crashes)"; fi +} + FX="$(dirname "${BASH_SOURCE[0]}")/fixtures/iceberg" WHD="$FX/warehouse_del" [ -f "$WHD/db/t/metadata/dvapply.metadata.json" ] || pgc_skip fixture "deletion-vector fixtures are missing" @@ -60,100 +109,103 @@ dv_ids_of() { # ---- deletion vectors are applied ------------------------------------------- # THE <= boundary: a DV at the data file's own sequence number applies (same # rule as 4a position-delete files; a strict > would wrongly exclude it) -check "a deletion vector at an equal sequence number applies (survivors == oracle)" \ +CRASHES_BEFORE="$(crashes)" +check_text "a deletion vector at an equal sequence number applies (survivors == oracle)" \ "$(dv_ids_of dvapply)" "$(dv_oracle dvapply)" # the other side: a DV strictly older than the data applies to nothing -check "a deletion vector older than the data does not apply" \ +check_text "a deletion vector older than the data does not apply" \ "$(dv_ids_of dvnoapply)" "$(dv_oracle dvnoapply)" # the spec's supersede rule: a position-delete FILE applies only when no DV # must be applied to the data file. The fixture's DV drops ordinal 1 and its # (also applicable) Parquet posdel names ordinals 1 and 3; a UNION would drop # both, supersede drops only ordinal 1. -check "an applicable DV supersedes position-delete files for its data file" \ +check_text "an applicable DV supersedes position-delete files for its data file" \ "$(dv_ids_of dvsupersede)" "$(dv_oracle dvsupersede)" # supersede is per data file: a posdel targeting a DIFFERENT data file still # applies alongside the DV -check "supersede is per data file, not global" \ +check_text "supersede is per data file, not global" \ "$(dv_ids_of dvother)" "$(dv_oracle dvother)" # roaring coverage: run-form containers plus a second 64-bit bucket -check "a DV with run containers and a high-bucket position decodes" \ +check_text "a DV with run containers and a high-bucket position decodes" \ "$(dv_ids_of dvwide)" "$(dv_oracle dvwide)" # cookie 12347 (run containers marked in the run bitset) -check "a run-optimized DV (cookie 12347) decodes" \ +check_text "a run-optimized DV (cookie 12347) decodes" \ "$(dv_ids_of dvrun)" "$(dv_oracle dvrun)" # a bitset container (cardinality above 4096) -check "a bitset-container DV decodes" \ +check_text "a bitset-container DV decodes" \ "$(dv_ids_of dvbitset)" "$(dv_oracle dvbitset)" # one Puffin file carrying two DV blobs for two data files -check "two DV blobs in one Puffin file each apply to their own data file" \ +check_text "two DV blobs in one Puffin file each apply to their own data file" \ "$(dv_ids_of dvtwo)" "$(dv_oracle dvtwo)" -check "backend still up after the DV applications" "$(q 'SELECT 1')" "1" +check "backend still up after the DV applications" "$(survived "$CRASHES_BEFORE")" "1" # ---- refusal and corruption arms -------------------------------------------- +CRASHES_BEFORE="$(crashes)" # two DV entries referencing one data file: the spec says scan results are # undefined and readers may raise; we raise rather than guess check "two DV entries for one data file are refused (XX001)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvdup.metadata.json') - AS t(id bigint, region text, amount int)")" "XX001" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvdup.metadata.json') + AS t(id bigint, region text, amount int)" "two deletion vectors reference data file")" "XX001" check "a corrupted DV CRC is refused (XX001)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvbadcrc.metadata.json') - AS t(id bigint, region text, amount int)")" "XX001" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvbadcrc.metadata.json') + AS t(id bigint, region text, amount int)" "fails its CRC-32 check")" "XX001" check "a wrong DV blob magic is refused (XX001)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvbadmagic.metadata.json') - AS t(id bigint, region text, amount int)")" "XX001" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvbadmagic.metadata.json') + AS t(id bigint, region text, amount int)" "has a bad magic sequence")" "XX001" # the manifest's content_offset must exactly match the footer's blob offset check "a manifest/footer offset mismatch is refused (XX001)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvoffmismatch.metadata.json') - AS t(id bigint, region text, amount int)")" "XX001" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvoffmismatch.metadata.json') + AS t(id bigint, region text, amount int)" "has no blob matching the manifest's offset")" "XX001" # referenced_data_file is required for DVs check "a DV entry without referenced_data_file is refused (XX001)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvnoref.metadata.json') - AS t(id bigint, region text, amount int)")" "XX001" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvnoref.metadata.json') + AS t(id bigint, region text, amount int)" "has no referenced_data_file")" "XX001" # record_count is defined as the DV's cardinality check "a record_count that disagrees with the DV cardinality is refused (XX001)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvbadcount.metadata.json') - AS t(id bigint, region text, amount int)")" "XX001" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvbadcount.metadata.json') + AS t(id bigint, region text, amount int)" "positions but its manifest entry records")" "XX001" # deletion-vector-v1 blobs must not declare a compression codec check "a DV blob declaring a compression codec is refused (XX001)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvcompressed.metadata.json') - AS t(id bigint, region text, amount int)")" "XX001" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvcompressed.metadata.json') + AS t(id bigint, region text, amount int)" "declares a compression codec")" "XX001" # a compressed Puffin footer is refused loudly, never misparsed check "a compressed Puffin footer is refused (0A000)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvflags.metadata.json') - AS t(id bigint, region text, amount int)")" "0A000" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvflags.metadata.json') + AS t(id bigint, region text, amount int)" "has a compressed Puffin footer")" "0A000" check "...and the refusal names the compressed footer" \ "$(errmsg_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvflags.metadata.json') AS t(id bigint, region text, amount int)" | grep -c "compressed")" "1" # deletion vectors are v3-only check "a deletion vector in a format-version 2 table is refused (0A000)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvv2.metadata.json') - AS t(id bigint, region text, amount int)")" "0A000" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvv2.metadata.json') + AS t(id bigint, region text, amount int)" "is a deletion vector, which format-version")" "0A000" check "...and the refusal names the format version" \ "$(errmsg_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvv2.metadata.json') AS t(id bigint, region text, amount int)" | grep -c "format-version")" "1" # a Puffin path outside the table root is stopped by the path boundary check "a Puffin path outside the table location is refused (22023)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvescape.metadata.json') - AS t(id bigint, region text, amount int)")" "22023" -check "backend still up after the DV refusals" "$(q 'SELECT 1')" "1" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvescape.metadata.json') + AS t(id bigint, region text, amount int)" "is not under the table location")" "22023" +check "backend still up after the DV refusals" "$(survived "$CRASHES_BEFORE")" "1" # ---- audit arms (multi-agent adversarial audit findings) -------------------- +CRASHES_BEFORE="$(crashes)" # content_offset + content_size chosen to overflow int64: a bounds check that # adds the operands would wrap negative and read a wild pointer; testing each # operand against the file size refuses it cleanly check "a DV whose offset+size overflows int64 is refused, not read (XX001)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvbigoff.metadata.json') - AS t(id bigint, region text, amount int)")" "XX001" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvbigoff.metadata.json') + AS t(id bigint, region text, amount int)" "lies outside the file's blob region")" "XX001" # a run container whose start+length exceeds the 16-bit container range would # fabricate positions in a neighboring container; refuse rather than corrupt check "a run container overflowing its 16-bit range is refused (XX001)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvrunoverflow.metadata.json') - AS t(id bigint, region text, amount int)")" "XX001" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvrunoverflow.metadata.json') + AS t(id bigint, region text, amount int)" "overflows its 16-bit range")" "XX001" # two DV entries for one data file recorded with and without the file:// scheme: # the one-DV-per-file check strips the scheme, so the aliased pair is caught check "an aliased duplicate DV (scheme vs no scheme) is refused (XX001)" \ - "$(sqlstate_of "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvdupscheme.metadata.json') - AS t(id bigint, region text, amount int)")" "XX001" -check "backend still up after the audit arms" "$(q 'SELECT 1')" "1" + "$(refusal "SELECT * FROM pgcolumnar.iceberg_scan('$MDIR/dvdupscheme.metadata.json') + AS t(id bigint, region text, amount int)" "two deletion vectors reference data file")" "XX001" +check "backend still up after the audit arms" "$(survived "$CRASHES_BEFORE")" "1" pgc_summary diff --git a/test/logical_decoding_source.sh b/test/logical_decoding_source.sh index c17b4b1df..8e406bf7d 100755 --- a/test/logical_decoding_source.sh +++ b/test/logical_decoding_source.sh @@ -35,7 +35,8 @@ pgc_setup "${1:-/usr/local/pg17/bin/pg_config}" # test_decoding is stock contrib, but it is not built by default in a source # install. A missing dependency is an environment defect rather than a pass. if [ ! -f "$("$PGC_PG_CONFIG" --pkglibdir)/test_decoding.so" ]; then - pgc_skip "test_decoding" "test_decoding.so is not in $("$PGC_PG_CONFIG" --pkglibdir); build contrib/test_decoding" + # The message is the check's name, so it is a literal: the twin records the same one. + pgc_skip "test_decoding" "test_decoding.so is not installed; build contrib/test_decoding" fi check "premise: the server is running at wal_level=logical" \ @@ -100,20 +101,30 @@ check "the slot carries pgcolumnar's internal catalog instead" \ # it useless as a change stream: it does not carry the data and it does not even # carry a row count. Asserted by writing 100x the rows into a second columnar # table and showing the record count does not follow. +# +# THE SECOND SLOT GETS ITS OWN HEAP CONTROL. Without one, a slot that decoded +# nothing passed both checks below: "no decodable change" reads 0 and "not 100x +# the records" reads 0 < anything. Measured: with pgc_754b created after the +# insert, the slot carried 0 records and every check passed. psql_run "CREATE TABLE col_big (id int PRIMARY KEY, v text) USING pgcolumnar;" +psql_run "CREATE TABLE heap_big (id int PRIMARY KEY, v text) USING heap;" psql_run "SELECT pg_create_logical_replication_slot('pgc_754b', 'test_decoding');" psql_run "INSERT INTO col_big SELECT g, 'c' || g FROM generate_series(1, 5000) g;" +psql_run "INSERT INTO heap_big SELECT g, 'h' || g FROM generate_series(1, 5000) g;" psql_run "CREATE TABLE decoded_big AS SELECT data FROM pg_logical_slot_get_changes('pgc_754b', NULL, NULL);" SMALL="$(d 'table pgcolumnar.')" BIG="$(q "SELECT count(*) FROM decoded_big WHERE data LIKE '%table pgcolumnar.%';")" check "premise: the 100x table really holds 100x the rows" \ "$(q 'SELECT count(*) FROM col_big;')" "5000" +check_num "CONTROL: the second slot decodes the heap table's 5000 inserts" \ + "$(q "SELECT count(*) FROM decoded_big WHERE data LIKE '%table public.heap_big: INSERT%';")" "5000" check_num "and it too emits no decodable change of its own" \ "$(q "SELECT count(*) FROM decoded_big WHERE data LIKE '%table public.col_big:%';")" "0" echo "-- internal pgcolumnar records: $SMALL for 50 rows, $BIG for 5000" check "100x the rows does not give 100x the internal records" \ - "$([ "$BIG" -lt $(( SMALL * 10 )) ] && echo yes || echo "no ($SMALL then $BIG)")" "yes" + "$([ "${BIG:-0}" -gt 0 ] 2>/dev/null && [ "$BIG" -lt $(( SMALL * 10 )) ] 2>/dev/null && echo yes || + echo "no ($SMALL then $BIG; the second slot must carry some)")" "yes" psql_run "SELECT pg_drop_replication_slot('pgc_754');" psql_run "SELECT pg_drop_replication_slot('pgc_754b');" diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md index 77e562c8b..74bd6ff82 100644 --- a/test/pytest/TESTS.md +++ b/test/pytest/TESTS.md @@ -313,6 +313,8 @@ behaviour, the source of that number is named. - [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) - [266. test_iceberg_name_mapping.py: an Iceberg data file with no field ids binds by name mapping](#266-test_iceberg_name_mappingpy-an-iceberg-data-file-with-no-field-ids-binds-by-name-mapping) - [267. test_iceberg_objstore.py: Iceberg tables read over object storage](#267-test_iceberg_objstorepy-iceberg-tables-read-over-object-storage) +- [268. test_iceberg_dv.py: Iceberg v3 deletion vectors apply, supersede, and refuse what they must](#268-test_iceberg_dvpy-iceberg-v3-deletion-vectors-apply-supersede-and-refuse-what-they-must) +- [269. test_logical_decoding_source.py: a columnar table is not a logical decoding source](#269-test_logical_decoding_sourcepy-a-columnar-table-is-not-a-logical-decoding-source) ## 1. How to read a test in here @@ -10396,3 +10398,64 @@ One check per test, each by the shell's name, every reading a value from a modul | `test_the_backend_is_still_up_after_the_s3_refusal` | backend still up after the s3 refusal | | `test_an_endpoint_not_on_the_allow_list_is_refused` | an endpoint not on objstore_allowed_endpoints is refused | | `test_the_backend_is_still_up_after_the_endpoint_refusal` | backend still up after the endpoint refusal | + +## 268. test_iceberg_dv.py: Iceberg v3 deletion vectors apply, supersede, and refuse what they must + +The twin of `test/iceberg_dv.sh`. Eight hand-crafted snapshots read the oracle's surviving ids, thirteen corrupt or unsupported ones are refused, each by its own guard, and no backend dies on the way. The warehouse is relocated into `server_dir` so its recorded root differs from where it lies. + +### Every test + +Every test makes one check under the shell's name. A refusal reads as its SQLSTATE only when the message is that guard's own (several guards raise XX001), and a failed reading is a `query_error()`. Each survival check also requires that the server log's count of backends killed by a signal has not moved since its section began, and each refusal first waits for the server to answer, so a crash reddens only its own arm and its section's survival check. + +| test | what it holds | +| --- | --- | +| `test_a_deletion_vector_at_an_equal_sequence_number_applies` | a deletion vector at an equal sequence number applies (survivors == oracle) | +| `test_a_deletion_vector_older_than_the_data_does_not_apply` | a deletion vector older than the data does not apply | +| `test_an_applicable_dv_supersedes_position_delete_files` | an applicable DV supersedes position-delete files for its data file | +| `test_dv_supersede_is_per_data_file_not_global` | supersede is per data file, not global | +| `test_a_dv_with_run_containers_and_a_high_bucket_decodes` | a DV with run containers and a high-bucket position decodes | +| `test_a_run_optimized_dv_decodes` | a run-optimized DV (cookie 12347) decodes | +| `test_a_bitset_container_dv_decodes` | a bitset-container DV decodes | +| `test_two_dv_blobs_in_one_puffin_file_each_apply_to_their_own_file` | two DV blobs in one Puffin file each apply to their own data file | +| `test_the_backend_survives_the_dv_applications` | backend still up after the DV applications | +| `test_two_dv_entries_for_one_data_file_are_refused` | two DV entries for one data file are refused (XX001) | +| `test_a_corrupted_dv_crc_is_refused` | a corrupted DV CRC is refused (XX001) | +| `test_a_wrong_dv_blob_magic_is_refused` | a wrong DV blob magic is refused (XX001) | +| `test_a_manifest_footer_offset_mismatch_is_refused` | a manifest/footer offset mismatch is refused (XX001) | +| `test_a_dv_entry_without_referenced_data_file_is_refused` | a DV entry without referenced_data_file is refused (XX001) | +| `test_a_record_count_disagreeing_with_the_dv_is_refused` | a record_count that disagrees with the DV cardinality is refused (XX001) | +| `test_a_dv_blob_declaring_a_codec_is_refused` | a DV blob declaring a compression codec is refused (XX001) | +| `test_a_compressed_puffin_footer_is_refused` | a compressed Puffin footer is refused (0A000) | +| `test_the_footer_refusal_names_the_compressed_footer` | ...and the refusal names the compressed footer | +| `test_a_dv_in_a_format_version_2_table_is_refused` | a deletion vector in a format-version 2 table is refused (0A000) | +| `test_the_v2_refusal_names_the_format_version` | ...and the refusal names the format version | +| `test_a_puffin_path_outside_the_table_is_refused` | a Puffin path outside the table location is refused (22023) | +| `test_the_backend_survives_the_dv_refusals` | backend still up after the DV refusals | +| `test_a_dv_whose_offset_plus_size_overflows_is_refused` | a DV whose offset+size overflows int64 is refused, not read (XX001) | +| `test_a_run_container_overflowing_its_range_is_refused` | a run container overflowing its 16-bit range is refused (XX001) | +| `test_an_aliased_duplicate_dv_is_refused` | an aliased duplicate DV (scheme vs no scheme) is refused (XX001) | +| `test_the_backend_survives_the_audit_arms` | backend still up after the audit arms | + +## 269. test_logical_decoding_source.py: a columnar table is not a logical decoding source + +The twin of `test/logical_decoding_source.sh` (#754). Through test_decoding slots on a private cluster at `wal_level=logical`: a heap table's inserts decode and a columnar table's do not, although its rows are there; the slot carries pgcolumnar's own catalog writes instead; and those follow groups, not rows (5000 rows give fewer than 10x the records of 50). Each slot has a heap control written in the same window, so a dead slot cannot pass as an exclusion. + +### Every test + +One check per test, each by the shell's name, every reading a value (a failed query is a `query_error()` that fails its own check) from a module fixture on a private cluster. Without the test_decoding library both harnesses fail one check, the shell's `pgc_skip` name, and the twin's other tests decline MISSING_DEPENDENCY. + +| test | what it holds | +| --- | --- | +| `test_lds_test_decoding_is_installed` | test_decoding.so is not installed; build contrib/test_decoding | +| `test_lds_server_runs_at_wal_level_logical` | premise: the server is running at wal_level=logical | +| `test_lds_slot_exists_and_is_logical` | premise: the slot exists and is logical | +| `test_lds_both_tables_hold_the_same_rows` | premise: both tables hold the same 50 rows | +| `test_lds_slot_decoded_something` | premise: the slot decoded something at all | +| `test_lds_heap_control_is_decoded` | CONTROL: the heap table's 50 inserts ARE decoded | +| `test_lds_columnar_emits_no_change` | a columnar table emits no decodable change | +| `test_lds_columnar_rows_are_there` | although the rows are there | +| `test_lds_slot_carries_internal_catalog` | the slot carries pgcolumnar's internal catalog instead | +| `test_lds_big_table_holds_its_rows` | premise: the 100x table really holds 100x the rows | +| `test_lds_second_slot_heap_control_is_decoded` | CONTROL: the second slot decodes the heap table's 5000 inserts | +| `test_lds_big_columnar_emits_no_change` | and it too emits no decodable change of its own | +| `test_lds_internal_records_do_not_follow_rows` | 100x the rows does not give 100x the internal records | diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt index aa84b2450..630bd3f16 100644 --- a/test/pytest/expected_tests.txt +++ b/test/pytest/expected_tests.txt @@ -906,6 +906,10 @@ guard_tests 476 # 22 checks in 22 tests, by collection over the cluster files. # cluster_tests 2732 -> 2748 for #1341's test_iceberg_objstore.py, the twin of iceberg_objstore.sh: # 16 checks in 16 tests, by collection over the cluster files. +# cluster_tests 2753 -> 2779 for #1341's test_iceberg_dv.py, the twin of iceberg_dv.sh: +# 26 checks in 26 tests, by collection over the cluster files. +# cluster_tests 2779 -> 2792 for #1341's test_logical_decoding_source.py, the twin of logical_decoding_source.sh: +# 13 checks in 13 tests, by collection over the cluster files. # cluster_tests 1264 -> 1287 for #1341's test_vector_agg_rescan_memory.py, the twin of vector_agg_rescan_memory.sh: # 23 tests, one per check, by collection over the cluster files. # cluster_tests 1287 -> 1329 for #1341's test_analyze_stats.py, the twin of analyze_stats.sh: @@ -1037,4 +1041,4 @@ guard_tests 476 # 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 2748 -> 2753 for #1706's five timestamp-flag tests in test_native_parquet_schema.py, by collection. -cluster_tests 2753 +cluster_tests 2792 diff --git a/test/pytest/test_compare_to_bash.py b/test/pytest/test_compare_to_bash.py index 5a7aa689a..c9fcd7ae8 100644 --- a/test/pytest/test_compare_to_bash.py +++ b/test/pytest/test_compare_to_bash.py @@ -127,6 +127,7 @@ "iceberg_scan", "iceberg_malformed", "iceberg_objstore", + "iceberg_dv", "concurrency", "isolation", "index_only", @@ -139,6 +140,7 @@ "doc_parallel_premise", "encode_effort", "local_open_race_free", + "logical_decoding_source", "sorted_mark_rename", "tablesample", "ungrouped_vector_agg", diff --git a/test/pytest/test_iceberg_dv.py b/test/pytest/test_iceberg_dv.py new file mode 100644 index 000000000..795608f0c --- /dev/null +++ b/test/pytest/test_iceberg_dv.py @@ -0,0 +1,344 @@ +"""Iceberg v3 deletion vectors (#388 phase 4c): iceberg_scan reads a table whose current +snapshot carries Puffin deletion vectors and drops the rows they name. The pytest twin +of `test/iceberg_dv.sh`, every check by the shell's name (#1341). + +The fixtures are hand-crafted (test/fixtures/iceberg/warehouse_del): pyroaring bitmaps +in hand-built Puffin files, hand-encoded manifests, exact sequence numbers, and an +oracle cross-checked against pyiceberg and DuckDB. The warehouse is relocated so its +recorded root differs from where it lies. + +Asserted: eight snapshots read the oracle's surviving ids (the <= sequence boundary, +supersede per data file, run, bitset and high-bucket containers, two blobs in one +file); thirteen corrupt or unsupported snapshots are refused, each by its own guard; +and no backend dies on the way. + +HOW A REFUSAL IS READ. Several guards raise XX001, so a refusal counts only when its +message is that guard's own: with the manifest/footer offset match removed, a later +XX001 refused the same file and the arm, reading the SQLSTATE alone, still passed. A +refusal reads as its SQLSTATE, " (another refusal: ...)", "no error", or a +query_error() when the server could not be asked. + +HOW SURVIVAL IS READ. A fresh connection answers again after crash recovery, so +"SELECT 1" alone passed with a backend killed by a refusal arm. Each survival check +also requires that the server log's count of backends terminated by a signal has not +moved since its section began. + +ON A PRIVATE DATABASE, with the warehouse copied into `server_dir`, because the backend +reads it. Where the fixture is missing the first test fails under the shell's pgc_skip +message, and the rest decline. +""" +import json +import pathlib +import shutil +import time + +import psycopg +import pytest + +from pgc_vacuity import query_error + +DB = "pgc_iceberg_dv" +FX = pathlib.Path(__file__).resolve().parents[1] / "fixtures" / "iceberg" +WHD = FX / "warehouse_del" +COLS = "AS t(id bigint, region text, amount int)" +APPLIED = ("dvapply", "dvnoapply", "dvsupersede", "dvother", "dvwide", "dvrun", "dvbitset", "dvtwo") +REFUSED = { + "dvdup": "two deletion vectors reference data file", + "dvbadcrc": "fails its CRC-32 check", + "dvbadmagic": "has a bad magic sequence", + "dvoffmismatch": "has no blob matching the manifest's offset", + "dvnoref": "has no referenced_data_file", + "dvbadcount": "positions but its manifest entry records", + "dvcompressed": "declares a compression codec", + "dvflags": "has a compressed Puffin footer", + "dvv2": "is a deletion vector, which format-version", + "dvescape": "is not under the table location", +} +AUDIT = { + "dvbigoff": "lies outside the file's blob region", + "dvrunoverflow": "overflows its 16-bit range", + "dvdupscheme": "two deletion vectors reference data file", +} + + +def _message(exc): + return str(exc).splitlines()[0] if str(exc) else type(exc).__name__ + + +def _value(dsn, sql): + try: + with psycopg.connect(dsn, autocommit=True, connect_timeout=10) as conn: + row = conn.execute(sql).fetchone() + except psycopg.Error as exc: + return query_error(_message(exc)) + return "" if row is None or row[0] is None else row[0] + + +def _await_up(dsn): + """Wait, at most 60 s, until the server answers, as the shell's await_up: a refusal + that kills a backend sends the server through crash recovery, and the next readings + would otherwise red as collateral, by timing. survived() still counts the crash.""" + for _ in range(120): + if _value(dsn, "SELECT 1") == 1: + return + time.sleep(0.5) + + +def _refusal(dsn, sql, phrase): + """-> the SQLSTATE when sql is refused with phrase in its primary message, the + SQLSTATE with the other guard's message, "no error", or query_error().""" + _await_up(dsn) + try: + with psycopg.connect(dsn, autocommit=True, connect_timeout=10) as conn: + conn.execute(sql).fetchall() + except psycopg.Error as exc: + if not exc.sqlstate: + return query_error(_message(exc)) + msg = exc.diag.message_primary or "" + return exc.sqlstate if phrase in msg else f"{exc.sqlstate} (another refusal: {msg})" + return "no error" + + +def _names(dsn, sql, phrase): + """-> "1" when sql's refusal message carries phrase, "0" when it does not, as the + shell's grep -c over errmsg_of; query_error() when the server could not be asked.""" + _await_up(dsn) + try: + with psycopg.connect(dsn, autocommit=True, connect_timeout=10) as conn: + conn.execute(sql).fetchall() + except psycopg.Error as exc: + if not exc.sqlstate: + return query_error(_message(exc)) + return "1" if phrase in (exc.diag.message_primary or "") else "0" + return "0" + + +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): + 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}") + return "1" if up == 1 and now == before else f"no ([{up}], {now - before} backend crashes)" + + +@pytest.fixture(scope="module") +def dv(pgc_cluster, server_dir): + if not (WHD / "db" / "t" / "metadata" / "dvapply.metadata.json").is_file(): + yield None + return + dest = server_dir / "dv" + shutil.copytree(WHD / "db", dest / "db") + for p in [dest, *dest.rglob("*")]: + p.chmod(0o777 if p.is_dir() else 0o666) + mdir = dest / "db" / "t" / "metadata" + oracle = json.loads((WHD / "expected_deletes.json").read_text())["dv_surviving"] + + def scan(name): + return f"SELECT * FROM pgcolumnar.iceberg_scan('{mdir / (name + '.metadata.json')}') {COLS}" + + admin = pgc_cluster.dsn() + with psycopg.connect(admin, 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") + r = {"oracle": {k: ",".join(str(i) for i in v) for k, v in oracle.items()}} + # the shell's order, section by section, each with its own crash baseline + before = _crashes(pgc_cluster) + for name in APPLIED: + r[name] = _value(dsn, f"SELECT string_agg(id::text, ',' ORDER BY id) FROM " + f"pgcolumnar.iceberg_scan('{mdir / (name + '.metadata.json')}') {COLS}") + r["up after applications"] = _survived(dsn, pgc_cluster, before) + before = _crashes(pgc_cluster) + for name, phrase in REFUSED.items(): + r[name] = _refusal(dsn, scan(name), phrase) + if name == "dvflags": + r["dvflags names"] = _names(dsn, scan(name), "compressed") + if name == "dvv2": + r["dvv2 names"] = _names(dsn, scan(name), "format-version") + r["up after refusals"] = _survived(dsn, pgc_cluster, before) + before = _crashes(pgc_cluster) + for name, phrase in AUDIT.items(): + r[name] = _refusal(dsn, scan(name), phrase) + r["up after audit"] = _survived(dsn, pgc_cluster, before) + yield r + finally: + with psycopg.connect(admin, autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + + +def _declined(dv, expect): + if dv is not None: + return False + expect.cannot_run("ABSENT_FIXTURE", "the deletion-vector fixtures are missing") + return True + + +def test_a_deletion_vector_at_an_equal_sequence_number_applies(dv, expect): + if dv is None: + expect.text("missing", "present", "deletion-vector fixtures are missing") + return + expect.require_module("json", name="python3 is needed") + expect.text(str(dv["dvapply"]), dv["oracle"]["dvapply"], + "a deletion vector at an equal sequence number applies (survivors == oracle)") + + +def test_a_deletion_vector_older_than_the_data_does_not_apply(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvnoapply"]), dv["oracle"]["dvnoapply"], "a deletion vector older than the data does not apply") + + +def test_an_applicable_dv_supersedes_position_delete_files(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvsupersede"]), dv["oracle"]["dvsupersede"], + "an applicable DV supersedes position-delete files for its data file") + + +def test_dv_supersede_is_per_data_file_not_global(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvother"]), dv["oracle"]["dvother"], "supersede is per data file, not global") + + +def test_a_dv_with_run_containers_and_a_high_bucket_decodes(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvwide"]), dv["oracle"]["dvwide"], "a DV with run containers and a high-bucket position decodes") + + +def test_a_run_optimized_dv_decodes(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvrun"]), dv["oracle"]["dvrun"], "a run-optimized DV (cookie 12347) decodes") + + +def test_a_bitset_container_dv_decodes(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvbitset"]), dv["oracle"]["dvbitset"], "a bitset-container DV decodes") + + +def test_two_dv_blobs_in_one_puffin_file_each_apply_to_their_own_file(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvtwo"]), dv["oracle"]["dvtwo"], + "two DV blobs in one Puffin file each apply to their own data file") + + +def test_the_backend_survives_the_dv_applications(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["up after applications"]), "1", "backend still up after the DV applications") + + +def test_two_dv_entries_for_one_data_file_are_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvdup"]), "XX001", "two DV entries for one data file are refused (XX001)") + + +def test_a_corrupted_dv_crc_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvbadcrc"]), "XX001", "a corrupted DV CRC is refused (XX001)") + + +def test_a_wrong_dv_blob_magic_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvbadmagic"]), "XX001", "a wrong DV blob magic is refused (XX001)") + + +def test_a_manifest_footer_offset_mismatch_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvoffmismatch"]), "XX001", "a manifest/footer offset mismatch is refused (XX001)") + + +def test_a_dv_entry_without_referenced_data_file_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvnoref"]), "XX001", "a DV entry without referenced_data_file is refused (XX001)") + + +def test_a_record_count_disagreeing_with_the_dv_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvbadcount"]), "XX001", + "a record_count that disagrees with the DV cardinality is refused (XX001)") + + +def test_a_dv_blob_declaring_a_codec_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvcompressed"]), "XX001", "a DV blob declaring a compression codec is refused (XX001)") + + +def test_a_compressed_puffin_footer_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvflags"]), "0A000", "a compressed Puffin footer is refused (0A000)") + + +def test_the_footer_refusal_names_the_compressed_footer(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvflags names"]), "1", "...and the refusal names the compressed footer") + + +def test_a_dv_in_a_format_version_2_table_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvv2"]), "0A000", "a deletion vector in a format-version 2 table is refused (0A000)") + + +def test_the_v2_refusal_names_the_format_version(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvv2 names"]), "1", "...and the refusal names the format version") + + +def test_a_puffin_path_outside_the_table_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvescape"]), "22023", "a Puffin path outside the table location is refused (22023)") + + +def test_the_backend_survives_the_dv_refusals(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["up after refusals"]), "1", "backend still up after the DV refusals") + + +def test_a_dv_whose_offset_plus_size_overflows_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvbigoff"]), "XX001", "a DV whose offset+size overflows int64 is refused, not read (XX001)") + + +def test_a_run_container_overflowing_its_range_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvrunoverflow"]), "XX001", "a run container overflowing its 16-bit range is refused (XX001)") + + +def test_an_aliased_duplicate_dv_is_refused(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["dvdupscheme"]), "XX001", "an aliased duplicate DV (scheme vs no scheme) is refused (XX001)") + + +def test_the_backend_survives_the_audit_arms(dv, expect): + if _declined(dv, expect): + return + expect.text(str(dv["up after audit"]), "1", "backend still up after the audit arms") diff --git a/test/pytest/test_logical_decoding_source.py b/test/pytest/test_logical_decoding_source.py new file mode 100644 index 000000000..d8e6a09db --- /dev/null +++ b/test/pytest/test_logical_decoding_source.py @@ -0,0 +1,189 @@ +"""pgColumnar is not a logical decoding source (#754). The pytest twin of +`test/logical_decoding_source.sh`, every check by the shell's name. + +Logical decoding reads the WAL records of heap tuples. Columnar data reaches WAL as +full-page images, which carry no tuple structure, so no change to a columnar table is +emitted. A slot on a database holding columnar tables is not silent either: it carries +pgcolumnar's own catalog writes, which follow groups and not rows. + +"Zero changes decoded" is also what a dead slot looks like, so each slot gets a heap +CONTROL written in the same window and asserted to decode. The second slot had none, +and with it created after the insert it carried nothing and every check passed. + +A PRIVATE CLUSTER at wal_level=logical: the level cannot be raised without a restart, +and the shared cluster runs at the default. +""" +import pathlib +import shutil +import tempfile + +import psycopg +import pytest + +from pgc_vacuity import query_error + +SLOT, SLOT_B = "pgc_754", "pgc_754b" + + +def _q(dsn, sql): + """-> the first column of the first row as text, or query_error() carrying the failure.""" + try: + with psycopg.connect(dsn, autocommit=True) as c: + row = c.execute(sql).fetchone() + return "" if row is None or row[0] is None else str(row[0]) + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + + +def _run(dsn, sql): + """-> None, or query_error() when the statement failed (the next reading then fails by name).""" + try: + with psycopg.connect(dsn, autocommit=True) as c: + c.execute(sql) + return None + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + + +def _int(v): + return int(v) if isinstance(v, str) and v.isdigit() else v + + +@pytest.fixture(scope="module") +def lds(pgc_cluster): + from pgc_cluster import Cluster, pick_port + + if not (pathlib.Path(pgc_cluster.libdir) / "test_decoding.so").is_file(): + yield {"missing": f"test_decoding.so is not in {pgc_cluster.libdir}"} + return + work = pathlib.Path(tempfile.mkdtemp(prefix="pgc-lds-")) + work.chmod(0o777) + datadir = work / "data" + datadir.mkdir() + cluster = Cluster(pgc_cluster.pg_config, "lds", datadir, pick_port(26)) + try: + cluster.initdb() + with open(datadir / "postgresql.conf", "a") as fh: + fh.write("\nwal_level=logical\nmax_replication_slots=8\nmax_wal_senders=8\n") + cluster.start() + dsn = cluster.dsn() + r = {} + _run(dsn, "CREATE EXTENSION IF NOT EXISTS pgcolumnar") + r["wal_level"] = _q(dsn, "SHOW wal_level") + _run(dsn, "CREATE TABLE heap_t (id int PRIMARY KEY, v text) USING heap") + _run(dsn, "CREATE TABLE col_t (id int PRIMARY KEY, v text) USING pgcolumnar") + # The slot is created before any row is written: a slot only carries later changes. + _run(dsn, f"SELECT pg_create_logical_replication_slot('{SLOT}', 'test_decoding')") + r["slot type"] = _q(dsn, f"SELECT slot_type FROM pg_replication_slots WHERE slot_name = '{SLOT}'") + _run(dsn, "INSERT INTO heap_t SELECT g, 'h' || g FROM generate_series(1, 50) g") + _run(dsn, "INSERT INTO col_t SELECT g, 'c' || g FROM generate_series(1, 50) g") + r["heap rows"] = _int(_q(dsn, "SELECT count(*) FROM heap_t")) + r["col rows"] = _int(_q(dsn, "SELECT count(*) FROM col_t")) + # Drained once, so every count reads the same snapshot of the slot. + _run(dsn, f"CREATE TABLE decoded AS SELECT data FROM pg_logical_slot_get_changes('{SLOT}', NULL, NULL)") + + def d(table, pattern): + return _int(_q(dsn, f"SELECT count(*) FROM {table} WHERE data LIKE '%{pattern}%'")) + + r["decoded"] = _int(_q(dsn, "SELECT count(*) FROM decoded")) + r["heap decoded"] = d("decoded", "table public.heap_t: INSERT") + r["col decoded"] = d("decoded", "table public.col_t:") + r["internal small"] = d("decoded", "table pgcolumnar.") + # The second slot, with its own heap control in the same window. + _run(dsn, "CREATE TABLE col_big (id int PRIMARY KEY, v text) USING pgcolumnar") + _run(dsn, "CREATE TABLE heap_big (id int PRIMARY KEY, v text) USING heap") + _run(dsn, f"SELECT pg_create_logical_replication_slot('{SLOT_B}', 'test_decoding')") + _run(dsn, "INSERT INTO col_big SELECT g, 'c' || g FROM generate_series(1, 5000) g") + _run(dsn, "INSERT INTO heap_big SELECT g, 'h' || g FROM generate_series(1, 5000) g") + _run(dsn, f"CREATE TABLE decoded_big AS SELECT data FROM pg_logical_slot_get_changes('{SLOT_B}', NULL, NULL)") + r["big rows"] = _q(dsn, "SELECT count(*) FROM col_big") + r["heap big decoded"] = d("decoded_big", "table public.heap_big: INSERT") + r["col big decoded"] = d("decoded_big", "table public.col_big:") + r["internal big"] = d("decoded_big", "table pgcolumnar.") + _run(dsn, f"SELECT pg_drop_replication_slot('{SLOT}')") + _run(dsn, f"SELECT pg_drop_replication_slot('{SLOT_B}')") + yield r + finally: + cluster.stop() + shutil.rmtree(work, ignore_errors=True) + + +def test_lds_test_decoding_is_installed(lds, expect): + # The shell's pgc_skip records a FAIL under this name when the plugin is missing, and + # nothing when it is there. Here the same name passes when it is there. + expect.text("missing" if "missing" in lds else "installed", "installed", + "test_decoding.so is not installed; build contrib/test_decoding") + + +def _declined(lds, expect): + if "missing" in lds: + expect.cannot_run("MISSING_DEPENDENCY", lds["missing"]) + return True + return False + + +def test_lds_server_runs_at_wal_level_logical(lds, expect): + if not _declined(lds, expect): + expect.text(lds["wal_level"], "logical", "premise: the server is running at wal_level=logical") + + +def test_lds_slot_exists_and_is_logical(lds, expect): + if not _declined(lds, expect): + expect.text(lds["slot type"], "logical", "premise: the slot exists and is logical") + + +def test_lds_both_tables_hold_the_same_rows(lds, expect): + if not _declined(lds, expect): + expect.num(lds["heap rows"], lds["col rows"], "premise: both tables hold the same 50 rows") + + +def test_lds_slot_decoded_something(lds, expect): + if not _declined(lds, expect): + n = lds["decoded"] + expect.text("yes" if isinstance(n, int) and n > 0 else f"no ({n})", "yes", + "premise: the slot decoded something at all") + + +def test_lds_heap_control_is_decoded(lds, expect): + if not _declined(lds, expect): + expect.num(lds["heap decoded"], 50, "CONTROL: the heap table's 50 inserts ARE decoded") + + +def test_lds_columnar_emits_no_change(lds, expect): + if not _declined(lds, expect): + expect.num(lds["col decoded"], 0, "a columnar table emits no decodable change") + + +def test_lds_columnar_rows_are_there(lds, expect): + if not _declined(lds, expect): + expect.num(lds["col rows"], 50, "although the rows are there") + + +def test_lds_slot_carries_internal_catalog(lds, expect): + if not _declined(lds, expect): + n = lds["internal small"] + expect.text("yes" if isinstance(n, int) and n > 0 else f"no ({n})", "yes", + "the slot carries pgcolumnar's internal catalog instead") + + +def test_lds_big_table_holds_its_rows(lds, expect): + if not _declined(lds, expect): + expect.text(lds["big rows"], "5000", "premise: the 100x table really holds 100x the rows") + + +def test_lds_second_slot_heap_control_is_decoded(lds, expect): + if not _declined(lds, expect): + expect.num(lds["heap big decoded"], 5000, "CONTROL: the second slot decodes the heap table's 5000 inserts") + + +def test_lds_big_columnar_emits_no_change(lds, expect): + if not _declined(lds, expect): + expect.num(lds["col big decoded"], 0, "and it too emits no decodable change of its own") + + +def test_lds_internal_records_do_not_follow_rows(lds, expect): + if not _declined(lds, expect): + small, big = lds["internal small"], lds["internal big"] + ok = isinstance(small, int) and isinstance(big, int) and 0 < big < small * 10 + expect.text("yes" if ok else f"no ({small} then {big}; the second slot must carry some)", "yes", + "100x the rows does not give 100x the internal records") diff --git a/test/pytest/unported_suites.txt b/test/pytest/unported_suites.txt index 9a22f4629..649c15d89 100644 --- a/test/pytest/unported_suites.txt +++ b/test/pytest/unported_suites.txt @@ -20,13 +20,11 @@ fuzz_parquet harness_selftest hilbert_curve iceberg_deletes -iceberg_dv iceberg_rest iceberg_rest_scan iceberg_rest_server iceberg_rest_vended logical_decoding_cdc_recipe -logical_decoding_source logical_subscriber native_agg_deletes native_cancel