diff --git a/test/check_ledger.tsv b/test/check_ledger.tsv index c5571c55d..7e722cfe6 100644 --- a/test/check_ledger.tsv +++ b/test/check_ledger.tsv @@ -3902,6 +3902,43 @@ parallel_scan_cost parallel_scan_cost premise: the serial plan has no Gather 15; parallel_scan_cost parallel_scan_cost premise: the serial plan is a columnar scan 15;16;17;18;19 never - parallel_scan_cost parallel_scan_cost premise: the table holds every inserted row 15;16;17;18;19 never - parallel_scan_cost parallel_scan_cost the leader-participation branch changes the divisor 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds a huge num_rows is rejected rather than walking off the allocation 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds a valid three-record list imports three rows (not over-rejected) 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds an honest file imports its rows 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds an honest one-record list imports one row 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds and it did not silently return more rows than the file holds 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds and it did not silently return more rows than the file's records 15;16;17;18;19 2026-10-21 columnar_parquet_reader.c: pq_require_entry never refuses a read past the decoded entries +parquet_count_bounds parquet_count_bounds and the backend is still up (it did not wrap-allocate and crash) 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds and the backend survived it 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds and the backend survived the list over-read 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds num_rows past a list's record count is rejected, not emitted as phantom rows 15;16;17;18;19 2026-10-21 columnar_parquet_reader.c: pq_require_entry never refuses a read past the decoded entries +parquet_count_bounds parquet_count_bounds num_rows past the chunk's values is rejected, not emitted as phantom rows 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds num_values 2^62 is rejected, not imported 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds premise: the crafter produced the seven files 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds premise: the server is up before any malformed file is read 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds the num_rows rejection names num_rows in its detail 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds the num_values rejection names num_values in its detail 15;16;17;18;19 never - +parquet_count_bounds parquet_count_bounds the server is still up after the honest import 15;16;17;18;19 never - +parquet_level_width parquet_level_width a non-positive maximum returns before shifting at all 15;16;17;18;19 2026-10-21 columnar_parquet_format.h: pq_bits_for's non-positive guard is commented out +parquet_level_width parquet_level_width and both callers reach the shared one 15;16;17;18;19 never - +parquet_level_width parquet_level_width and no signed 1 << remains in it 15;16;17;18;19 2026-10-21 columnar_parquet_format.h: pq_bits_for shifts a signed 1 +parquet_level_width parquet_level_width no local copy of the helper survives in the writer or the reader 15;16;17;18;19 2026-10-21 columnar_parquet.c: a local bits_for copy reappears in the writer +parquet_level_width parquet_level_width the loop is bounded below 32 shifts 15;16;17;18;19 2026-10-21 columnar_parquet_format.h: pq_bits_for's loop loses its b < 31 bound +parquet_level_width parquet_level_width the shared level-width helper exists 15;16;17;18;19 never - +parquet_level_width parquet_level_width the shift is unsigned 15;16;17;18;19 2026-10-21 columnar_parquet_format.h: pq_bits_for shifts a signed 1 +parquet_nested parquet_nested nested Parquet export rows written 15;16;17;18;19 never - +parquet_nested parquet_nested nested Parquet field types 15;16;17;18;19 never - +parquet_nested parquet_nested nested Parquet values match expectation 15;16;17;18;19 2026-10-21 columnar_parquet.c: an empty array is written with definition level 0, so it reads back NULL +parquet_nested parquet_nested np2 export rows 15;16;17;18;19 never - +parquet_nested parquet_nested np2 values match 15;16;17;18;19 never - +parquet_nested parquet_nested reject multi-dim array 15;16;17;18;19 2026-10-21 columnar_parquet.c: a multi-dimensional array is no longer refused +parquet_nested_import parquet_nested_import export rows 15;16;17;18;19 never - +parquet_nested_import parquet_nested_import import count matches 15;16;17;18;19 never - +parquet_nested_import parquet_nested_import import rows 15;16;17;18;19 never - +parquet_nested_import parquet_nested_import nested import round-trips (arrays + composite) 15;16;17;18;19 2026-10-21 columnar_parquet_reader.c: an empty Parquet list is read back as a NULL array +parquet_nested_import parquet_nested_import pyarrow import rows 15;16;17;18;19 never - +parquet_nested_import parquet_nested_import pyarrow nested file matches oracle 15;16;17;18;19 2026-10-21 columnar_parquet_reader.c: an empty Parquet list is read back as a NULL array +parquet_nested_import parquet_nested_import round-trips after delete 15;16;17;18;19 2026-10-21 columnar_parquet_reader.c: an empty Parquet list is read back as a NULL array partition_lifecycle partition_lifecycle a NULL cutoff is refused, in both overloads 15;16;17;18;19 2026-10-13 partition_drop_plan without its NULL-cutoff check, in both overloads partition_lifecycle partition_lifecycle a call that would create more than 1000 partitions is refused before any DDL 15;16;17;18;19 2026-10-12 partition_create_ahead without the 1000-partition bound partition_lifecycle partition_lifecycle a cutoff before every partition plans nothing 15;16;17;18;19 2026-10-13 the cutoff filter removed: every partition planned @@ -4497,6 +4534,19 @@ range_pruning range_pruning premise: the range type collates differently from it range_pruning range_pruning so does a containment probe 15;16;17;18;19 never - range_pruning range_pruning the overlap qual is pushed down, where it used to be dropped 15;16;17;18;19 never - range_pruning range_pruning the same for a containment probe far to its right 15;16;17;18;19 2026-10-17 columnar_write_state.c: the writer ignores an unbounded range, so its unit records only the largest finite upper bound +read_stream read_stream [off] a fresh session reads enable_read_stream as off 15;16;17;18;19 2026-10-21 columnar_tableam.c: pgcolumnar.enable_read_stream registered as pgcolumnar.enable_read_streamz, so the product's setting is gone +read_stream read_stream [off] aggregate 15;16;17;18;19 2026-10-20 the synchronous block-read path copies each page's bytes one byte late +read_stream read_stream [off] filtered 15;16;17;18;19 2026-10-20 the synchronous block-read path copies each page's bytes one byte late +read_stream read_stream [off] full scan 15;16;17;18;19 2026-10-20 the synchronous block-read path copies each page's bytes one byte late +read_stream read_stream [off] range 15;16;17;18;19 2026-10-20 the synchronous block-read path copies each page's bytes one byte late +read_stream read_stream [off] wide sample 15;16;17;18;19 2026-10-20 the synchronous block-read path copies each page's bytes one byte late +read_stream read_stream [on] a fresh session reads enable_read_stream as on 15;16;17;18;19 2026-10-21 columnar_tableam.c: pgcolumnar.enable_read_stream registered as pgcolumnar.enable_read_streamz, so the product's setting is gone +read_stream read_stream [on] aggregate 15;16;17;18;19 2026-10-20 the read-stream path copies each page's bytes one byte late (PG17+) +read_stream read_stream [on] filtered 15;16;17;18;19 2026-10-20 the read-stream path copies each page's bytes one byte late (PG17+) +read_stream read_stream [on] full scan 15;16;17;18;19 2026-10-20 the read-stream path copies each page's bytes one byte late (PG17+) +read_stream read_stream [on] range 15;16;17;18;19 2026-10-20 the read-stream path copies each page's bytes one byte late (PG17+) +read_stream read_stream [on] wide sample 15;16;17;18;19 2026-10-20 the read-stream path copies each page's bytes one byte late (PG17+) +read_stream read_stream multiple stripes present 15;16;17;18;19 never - reader_buffer_reuse reader_buffer_reuse a PROJECTED scan (one column of two) matches too 15;16;17;18;19 never - reader_buffer_reuse reader_buffer_reuse a column ADDED after the groups exist reads back NULL, as on heap 15;16;17;18;19 2026-10-21 columnar_reader.c: when the group buffer is reused rather than grown, a projected read does not refresh it, so it decodes the previous group's bytes reader_buffer_reuse reader_buffer_reuse a rescan (nested loop) returns the same rows 15;16;17;18;19 never - diff --git a/test/check_ledger_budget.txt b/test/check_ledger_budget.txt index 1f0344dd7..eefa96066 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 113 +suites_not_covered 108 # # checks_never_observed_red -- A CENSUS. NOT a ceiling, and it must not become # one. @@ -5211,4 +5211,88 @@ suites_not_covered 113 # never 3790 + 3 = 3793 # dated 2063 + 10 = 2073 # 3793 + 2073 == 5866 -checks_never_observed_red 3793 +# +# #1341, read_stream seeded, with its twin (jdatcmd): 3503 -> 3504, and +# suites_not_covered 149 -> 148. The suite had no rows. It enters with 13 checks from +# five-major runs (13/13 on each of 15-19, shell and twin alike), two of them new. +# The setting arrives by ALTER DATABASE SET, whose error the shell's q drops, so a +# setting that never took ran both modes down one path: with it misspelled and the +# synchronous path broken, the shell passed 11 of 11. Each mode now reads the setting +# back from a fresh session first. Those two reads name a pgcolumnar setting, and the +# setting registered under another name reddens them, so they are not premises. The +# twin's fixture returns the failed setting as their value, so they red by name there +# too. The ten comparisons' names are literal, and the stripe premise carries its +# count and leaves lossy_arms.tsv. 12 are dated on PG17, each in both twins with the +# same names: each mode's five comparisons by its path copying a page one byte late, +# and the two read-backs by the renamed setting. 1 arrives never. Nothing observes +# which path a read took: with the stream path taken whatever the setting says and the +# synchronous path broken, all 13 pass, so the [off] rows hold only while the setting +# routes reads. On 15 and 16 the stream path is compiled out and [on] repeats [off]. +# +# ledger rows 5866 + 13 = 5879 +# never 3793 + 1 = 3794 +# dated 2073 + 12 = 2085 +# 3794 + 2085 == 5879 +# +# #1341, parquet_level_width seeded, with its twin (jdatcmd): 3543 -> 3545, and +# suites_not_covered 144 -> 143. The suite had no rows. It enters with 7 checks from +# five-major runs (7/7 on each of 15-19; its twin needs no cluster). Three fixes: "no local +# copy" summed grep -c over two files, so with the writer's path wrong it read 0 and +# passed; it now needs both files. "no signed 1 << remains" counted over the extracted +# body, so with the header unreadable it read 0 and passed; it now needs a body. And the +# body checks read the comments inside the body, so a guard or bound commented out of the +# code passed; the body now has its comments blanked. 5 are dated by four source +# mutations, each reddening the same names in both twins: a signed shift, the b < 31 +# bound removed, a local copy back in the writer, and the non-positive guard commented +# out. 2 arrive never. +# +# ledger rows 5879 + 7 = 5886 +# never 3794 + 2 = 3796 +# dated 2085 + 5 = 2090 +# 3796 + 2090 == 5886 +# +# #1341, parquet_nested seeded, with its twin (jdatcmd): 3543 -> 3547, and +# suites_not_covered 144 -> 143. The suite had no rows. It enters with 6 checks from +# five-major runs (6/6 on each of 15-19, shell and twin alike), after two fixes. The +# value comparison visited only the rows the file held, so a zero-row file of the same +# schema printed MATCH; it now requires every id 1..5000. And any error counted as the +# multi-dimensional refusal, a missing table's included; it now reads that refusal. 2 are +# dated by two mutations, each reddening the same name in both twins: an empty array +# written as NULL, and the multi-dimensional refusal removed. 4 arrive never. +# +# ledger rows 5886 + 6 = 5892 +# never 3796 + 4 = 3800 +# dated 2090 + 2 = 2092 +# 3800 + 2092 == 5892 +# +# #1341, parquet_nested_import seeded, with its twin (jdatcmd): 3600 -> 3604, and +# suites_not_covered 139 -> 138. The suite had no rows. It enters with 7 checks from +# five-major runs (7/7 on each of 15-19, shell and twin alike), after one fix: "import +# count matches" compared two counts with check, so with both tables missing two empty +# readings agreed; it is check_num. 3 are dated by one mutation, reddening the same +# names in both twins: the reader taking an empty Parquet list for a NULL array, which +# every round-trip and the pyarrow oracle catch. 4 arrive never. +# +# ledger rows 5892 + 7 = 5899 +# never 3800 + 4 = 3804 +# dated 2092 + 3 = 2095 +# 3804 + 2095 == 5899 +# +# #1341, parquet_count_bounds seeded, with its twin (jdatcmd): 3650 -> 3665, and +# suites_not_covered 125 -> 124. The suite had no rows. It enters with 17 checks from +# five-major runs (17/17 on each of 15-19, shell and twin alike), after three fixes. The +# survival checks read SELECT 1, which answers after crash recovery: with the num_values +# bound removed the 2^62 file crashed the backend and "the backend survived it" passed; +# they now also need the log's count of killed backends unmoved. The phantom-row checks +# caught only the crafter's exact count; they now catch any count above the file's. And +# the crafter premise covers all seven files, not four. 2 are dated, by pq_require_entry +# refusing nothing, which reddens the two list checks in both twins. The num_values and +# num_rows mutations crash the backend, and a crash reddens checks it does not target +# (every read during recovery fails), so the ledger cannot attribute them and those rows +# stay never; the PR records what they reddened. 15 arrive never. +# +# ledger rows 5899 + 17 = 5916 +# never 3804 + 15 = 3819 +# dated 2095 + 2 = 2097 +# 3819 + 2097 == 5916 +checks_never_observed_red 3819 diff --git a/test/lossy_arms.tsv b/test/lossy_arms.tsv index 5a9faf39d..db96d0d54 100644 --- a/test/lossy_arms.tsv +++ b/test/lossy_arms.tsv @@ -147,7 +147,6 @@ projection_scan_cost a tight covering projection is cheaper relative to the base projection_scan_cost premise: the prunable range is priced above the one-stripe floor, so the arms differ projection_scan_cost while a plain range on the sort key still earns its discount projections fp2 spans multiple projection row groups -read_stream multiple stripes present scan_decode_cost a wide projection costs materially more than a narrow one (decode is priced) scan_decode_cost four wide text columns cost more than eight narrow int ones (#768) truncate_cleanup PREMISE the projection has its own storage, distinct from the base diff --git a/test/parquet_count_bounds.sh b/test/parquet_count_bounds.sh index bd8b99761..0cda215e1 100755 --- a/test/parquet_count_bounds.sh +++ b/test/parquet_count_bounds.sh @@ -69,16 +69,32 @@ detail_of() { -c "SELECT pgcolumnar.import_parquet('pcb','$W/$1');" 2>&1 } -# The server must still be up between calls. If a malformed file crashed the -# backend, this probe fails and every arm below is meaningless, so it gates them. -alive() { - env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -At \ - -c 'SELECT 1;' 2>&1 | head -1 +# The server must still be up between calls, AND no backend may have died since setup. +# A fresh connection answers again after crash recovery, so "SELECT 1" alone passed with +# the num_values bound removed: the 2^62 file crashed the backend, recovery finished, +# and "the backend survived it" read 1 (#1341, as avro_manifest found in #1576). The +# count of killed backends in the server log must not have moved either. +crashes() { + local _n + _n="$(pgc_pg "grep -c 'terminated by signal' '$PGC_LOGFILE'" 2>/dev/null)" + case "$_n" in '' | *[!0-9]*) ;; *) echo "$_n" ;; esac +} +alive() { # -> 1 when the server answers and no backend died since setup + local _up _now + _up="$(env PATH="$PGC_BINDIR:$PATH" psql -h 127.0.0.1 -p "$PGC_PORT" -U postgres -d "$PGC_DB" -At \ + -c 'SELECT 1;' 2>&1 | head -1)" + _now="$(crashes)" + if ! [[ "$CRASHES_BEFORE" =~ ^[0-9]+$ && "$_now" =~ ^[0-9]+$ ]]; then echo "no (crash count unreadable)" + elif [ "$_up" = 1 ] && [ "$_now" = "$CRASHES_BEFORE" ]; then echo 1 + else echo "no ([$_up], $(( _now - CRASHES_BEFORE )) backend crashes)"; fi } +CRASHES_BEFORE="$(crashes)" # ---- premises --------------------------------------------------------------- -check "premise: the crafter produced the four files" \ - "$([ -s "$W/honest.parquet" ] && [ -s "$W/bignv.parquet" ] && [ -s "$W/bignr.parquet" ] && [ -s "$W/hugenr.parquet" ] && echo yes || echo no)" "yes" +# All seven: the three list files were never checked, so a crafter that skipped them +# left the list arms reading "could not open file" (#1341). +check "premise: the crafter produced the seven files" \ + "$(m=""; for f in honest bignv bignr hugenr list_honest list_valid3 list_bad; do [ -s "$W/$f.parquet" ] || m="$m $f"; done; echo "${m:-yes}" | sed 's/^ /missing:/')" "yes" check "premise: the server is up before any malformed file is read" "$(alive)" "1" # ---- the honest file is unaffected ----------------------------------------- @@ -96,8 +112,10 @@ r_bignr="$(import_of bignr.parquet)" check "num_rows past the chunk's values is rejected, not emitted as phantom rows" \ "$(grep -qi 'malformed row group' <<<"$r_bignr" && echo rejected || echo "$r_bignr")" "rejected" # The phantom-row harm specifically: it must NOT return 40 rows for an 8-value file. +# ANY count above the file's 8 rows is a leak, not only the 40 the crafter declares: +# with the bound removed, a different phantom count read "ok" (#1341). check "and it did not silently return more rows than the file holds" \ - "$([ "$r_bignr" = "40" ] && echo leaked || echo ok)" "ok" + "$(if [[ "$r_bignr" =~ ^[0-9]+$ ]] && [ "$r_bignr" -gt 8 ]; then echo "leaked ($r_bignr rows)"; else echo ok; fi)" "ok" r_hugenr="$(import_of hugenr.parquet)" check "a huge num_rows is rejected rather than walking off the allocation" \ @@ -134,7 +152,7 @@ r_listbad="$(import_list_of list_bad.parquet)" check "num_rows past a list's record count is rejected, not emitted as phantom rows" \ "$(grep -qi 'malformed row group' <<<"$r_listbad" && echo rejected || echo "$r_listbad")" "rejected" check "and it did not silently return more rows than the file's records" \ - "$([ "$r_listbad" = "8" ] && echo leaked || echo ok)" "ok" + "$(if [[ "$r_listbad" =~ ^[0-9]+$ ]] && [ "$r_listbad" -gt 1 ]; then echo "leaked ($r_listbad rows)"; else echo ok; fi)" "ok" check "and the backend survived the list over-read" "$(alive)" "1" pgc_summary diff --git a/test/parquet_level_width.sh b/test/parquet_level_width.sh index eca876a85..b43941cfa 100755 --- a/test/parquet_level_width.sh +++ b/test/parquet_level_width.sh @@ -40,17 +40,23 @@ SRC="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)/src" HDR="$SRC/columnar_parquet_format.h" # The body, so every assertion below reads the shared definition and not a -# comment that happens to mention it. -body="$(awk '/^pq_bits_for\(int maxval\)/{f=1} f{print} f&&/^}/{exit}' "$HDR")" +# comment that happens to mention it. Comments inside the body are blanked too, +# multi-line ones included: a guard commented out of the code is not a guard. +body="$(awk '/^pq_bits_for\(int maxval\)/{f=1} f{print} f&&/^}/{exit}' "$HDR" | + perl -0777 -pe 's{/\*.*?\*/}{}gs; s{//[^\n]*}{}g')" check "the shared level-width helper exists" \ "$([ -n "$body" ] && echo yes || echo no)" "yes" # SINGLE-SOURCED. A local copy reappearing in either .c file is the drift this # is here to catch, and it would not fail any other check in the tree. +# Both files must be READ: grep -c over a missing file prints nothing for it, and the +# sum then reads 0, the passing answer -- with the writer's path wrong this passed (#1341). check "no local copy of the helper survives in the writer or the reader" \ - "$(grep -cE '^[a-z_]*bits_for\(int' "$SRC/columnar_parquet.c" "$SRC/columnar_parquet_reader.c" \ - | awk -F: '{n += $2} END {print n+0}')" "0" + "$(if [ -f "$SRC/columnar_parquet.c" ] && [ -f "$SRC/columnar_parquet_reader.c" ]; then + grep -cE '^[a-z_]*bits_for\(int' "$SRC/columnar_parquet.c" "$SRC/columnar_parquet_reader.c" \ + | awk -F: '{n += $2} END {print n+0}' + else echo "no reading (a source file is missing)"; fi)" "0" check "and both callers reach the shared one" \ "$([ "$(grep -c 'pq_bits_for(' "$SRC/columnar_parquet.c")" -ge 1 ] && [ "$(grep -c 'pq_bits_for(' "$SRC/columnar_parquet_reader.c")" -ge 1 ] && @@ -59,8 +65,10 @@ check "and both callers reach the shared one" \ # UNSIGNED. This is the #710 defect itself. check "the shift is unsigned" \ "$(grep -c '1u <<' <<<"$body")" "1" +# Over the body only when there IS one: a count over an empty extraction is 0, the +# passing answer, so with the header unreadable this passed (#1341). check "and no signed 1 << remains in it" \ - "$(grep -cE '\(1 <<|[^u] 1 <<' <<<"$body")" "0" + "$(if [ -n "$body" ]; then grep -cE '\(1 <<|[^u] 1 <<' <<<"$body"; else echo "no reading (no body)"; fi)" "0" # BOUNDED, which `1u <<` alone does NOT give. Making the shift unsigned also # makes the comparison unsigned, so a negative maxval converts to a huge value diff --git a/test/parquet_nested.sh b/test/parquet_nested.sh index aa40f5eef..5712debb6 100755 --- a/test/parquet_nested.sh +++ b/test/parquet_nested.sh @@ -23,13 +23,19 @@ if ! python3 -c 'import pyarrow' 2>/dev/null; then pgc_skip pyarrow "pyarrow not available; skipping Parquet nested export verification" fi -expect_error() { - local label="$1" sql="$2" - if psql_run "$sql" >/dev/null 2>&1; then - check "$label (expected error)" "succeeded" "error" - else - check "$label" "error" "error" +# refused SQL -> "refused: multi-dimensional", "succeeded", or the other error. Any +# error used to pass, so with the export pointed at a missing table it read as the +# refusal (#1341), as arrow_nested found before it. +md_refusal() { + local out + if out="$(psql_run "$1" 2>&1)"; then + echo "succeeded" + return fi + case "$out" in + *"does not support multi-dimensional arrays"*) echo "refused: multi-dimensional" ;; + *) echo "other error: $(printf '%s' "$out" | head -n 1)" ;; + esac } NF="$PGC_WORKDIR/nt.parquet" @@ -62,6 +68,11 @@ check "nested Parquet field types" "$types" "True int32 True string True 2 x int vals="$(python3 - "$NF" <<'PY' import sys, pyarrow as pa, pyarrow.parquet as pq d = pq.read_table(sys.argv[1]).to_pydict() +# Every id 1..5000, once: the loop below compares only rows the file holds, so a file +# with none printed MATCH (#1341). +if sorted(d['id']) != list(range(1, 5001)): + print("ROWS %d, not ids 1..5000" % len(d['id'])) + sys.exit(0) bad = None for i in range(len(d['id'])): g = d['id'][i] @@ -93,6 +104,8 @@ PY echo "-- multi-dimensional array rejected" psql_run "CREATE TABLE npm (id int, m int[]) USING pgcolumnar;" psql_run "INSERT INTO npm VALUES (1, ARRAY[[1,2],[3,4]]);" -expect_error "reject multi-dim array" "SELECT pgcolumnar.export_parquet('npm', '$PGC_WORKDIR/npm.parquet');" +check "reject multi-dim array" \ + "$(md_refusal "SELECT pgcolumnar.export_parquet('npm', '$PGC_WORKDIR/npm.parquet');")" \ + "refused: multi-dimensional" pgc_summary diff --git a/test/parquet_nested_import.sh b/test/parquet_nested_import.sh index cdae84880..165ac19bc 100755 --- a/test/parquet_nested_import.sh +++ b/test/parquet_nested_import.sh @@ -34,7 +34,9 @@ SEL="SELECT id, ia::text, ta::text, c::text FROM" check "nested import round-trips (arrays + composite)" \ "$(pgc_set_hash "$SEL pna2")" \ "$(pgc_set_hash "$SEL pna")" -check "import count matches" "$(q 'SELECT count(*) FROM pna2;')" "$(q 'SELECT count(*) FROM pna;')" +# check_num: under check two failed counts were both empty and agreed, so with both +# tables missing this passed (#1341). +check_num "import count matches" "$(q 'SELECT count(*) FROM pna2;')" "$(q 'SELECT count(*) FROM pna;')" # after a delete on the source, a fresh export+import still round-trips the live set psql_run "DELETE FROM pna WHERE id BETWEEN 1000 AND 2000;" diff --git a/test/pytest/TESTS.md b/test/pytest/TESTS.md index f63bef3ed..3ec8c6d0b 100644 --- a/test/pytest/TESTS.md +++ b/test/pytest/TESTS.md @@ -301,6 +301,11 @@ behaviour, the source of that number is named. - [253. test_pg19_vacuum_options.py: PostgreSQL 19's parallel-autovacuum parameter is accepted on a columnar table and changes nothing](#253-test_pg19_vacuum_optionspy-postgresql-19s-parallel-autovacuum-parameter-is-accepted-on-a-columnar-table-and-changes-nothing) - [254. test_cost_written_geometry.py: the cost model describes the geometry a table was written with](#254-test_cost_written_geometrypy-the-cost-model-describes-the-geometry-a-table-was-written-with) - [255. test_encode_invariants.py: encoding invariants for the #155 bulk-load optimisations](#255-test_encode_invariantspy-encoding-invariants-for-the-155-bulk-load-optimisations) +- [256. test_read_stream.py: reads through a read stream return what the synchronous path returns](#256-test_read_streampy-reads-through-a-read-stream-return-what-the-synchronous-path-returns) +- [257. test_parquet_level_width.py: the Parquet level-width helper is single-sourced, unsigned and bounded](#257-test_parquet_level_widthpy-the-parquet-level-width-helper-is-single-sourced-unsigned-and-bounded) +- [258. test_parquet_nested.py: Parquet nested export writes lists and groups that read back as the generator's values](#258-test_parquet_nestedpy-parquet-nested-export-writes-lists-and-groups-that-read-back-as-the-generators-values) +- [259. test_parquet_nested_import.py: nested Parquet import round-trips arrays and composites](#259-test_parquet_nested_importpy-nested-parquet-import-round-trips-arrays-and-composites) +- [260. test_parquet_count_bounds.py: file-declared Parquet counts are bounded before use](#260-test_parquet_count_boundspy-file-declared-parquet-counts-are-bounded-before-use) ## 1. How to read a test in here @@ -10040,3 +10045,95 @@ Every test reads one value from a module fixture that builds a private database | `test_no_vector_selects_fsst_at_or_below_the_distinct_cap` | no vector selects FSST at or below the distinct cap, so skipping its build cannot change output | | `test_the_block_compression_default_is_zstd` | the block compression default is zstd, pinned to #890 phase 1 | | `test_its_level_is_3` | and its level is 3 | + +## 256. test_read_stream.py: reads through a read stream return what the synchronous path returns + +The pytest twin of `test/read_stream.sh` (ported for #1341), every check by the shell's name. Five query shapes over a multi-stripe table are compared with a heap twin, once with `pgcolumnar.enable_read_stream` on and once off. On a private database. + +### Every test + +Each mode first reads its setting back from a fresh session, since a setting that never took would run both modes down one path. If the setting cannot be applied, the error is the value those two checks compare, so they red by name and the comparisons still run. A comparison is a set hash of each side. + +**What the `[off]` arms vouch for.** The read-back checks prove that the setting reached the session. They do not prove which path a read took, and no check here observes the path. With the stream path taken whatever the setting says, all 13 checks pass; with that and the synchronous path copying one byte late, all 13 still pass. So the `[off]` arms vouch for the synchronous path only while the setting routes reads to it. + +**On PostgreSQL 15 and 16** the stream path is compiled out, so `[on]` and `[off]` run the same synchronous code. There the `[on]` arms repeat the `[off]` arms. + +| test | what it holds | +| --- | --- | +| `test_multiple_stripes_present` | multiple stripes present | +| `test_a_fresh_session_reads_the_setting_on` | [on] a fresh session reads enable_read_stream as on | +| `test_a_fresh_session_reads_the_setting_off` | [off] a fresh session reads enable_read_stream as off | +| `test_a_shape_matches_its_heap_twin` | one test per row of `NAMES` (10 rows), each by the shell's name | + +## 257. test_parquet_level_width.py: the Parquet level-width helper is single-sourced, unsigned and bounded + +The pytest twin of `test/parquet_level_width.sh` (ported for #1341), every check by the shell's name. A source check, needing no cluster: `pq_bits_for` in `columnar_parquet_format.h` is the one definition of the RLE/bit-pack level width, called from the writer and the reader, with an unsigned shift and a loop bounded below 32 shifts (#710). Read in Python with `re`, independently of the shell's grep and awk. + +### Every test + +| test | what it holds | +| --- | --- | +| `test_the_shared_helper_exists` | the shared level-width helper exists | +| `test_no_local_copy_survives` | no local copy of the helper survives in the writer or the reader -- read only when both files are, since a count over a missing file is 0 | +| `test_both_callers_reach_the_shared_helper` | and both callers reach the shared one | +| `test_the_shift_is_unsigned` | the shift is unsigned | +| `test_no_signed_shift_remains` | and no signed 1 << remains in it -- read only over a body that was found | +| `test_the_loop_is_bounded` | the loop is bounded below 32 shifts | +| `test_a_non_positive_maximum_returns_first` | a non-positive maximum returns before shifting at all | + +## 258. test_parquet_nested.py: Parquet nested export writes lists and groups that read back as the generator's values + +The pytest twin of `test/parquet_nested.sh` (ported for #1341), every check by the shell's name. A columnar table with int[], text[] and composite columns, NULL arrays, empty arrays, NULL elements and NULL composites among them, is exported to Parquet; pyarrow reads it back as list/list/struct with every row equal to the generator's. A 70,000-row table spans several row groups and round-trips whole, and a multi-dimensional array is refused. Every test declines by the shell's skip name without pyarrow. On a private database, with the files in `server_dir`. + +### Every test + +| test | what it holds | +| --- | --- | +| `test_the_nested_parquet_export_writes_every_row` | nested Parquet export rows written | +| `test_the_parquet_fields_are_list_list_struct` | nested Parquet field types | +| `test_every_parquet_value_matches_the_generator` | nested Parquet values match expectation -- every id 1..5000 once, then each row equals the generator's; an empty file used to match | +| `test_the_multi_row_group_export_writes_every_row` | np2 export rows | +| `test_the_multi_row_group_values_match` | np2 values match | +| `test_a_multidimensional_array_is_refused_for_parquet` | reject multi-dim array -- the multi-dimensional refusal itself; any error used to count | + +## 259. test_parquet_nested_import.py: nested Parquet import round-trips arrays and composites + +The pytest twin of `test/parquet_nested_import.sh` (ported for #1341), every check by the shell's name. A table with int[], text[] and composite columns, NULL arrays, empty arrays, NULL elements and NULL composites among them, is exported to Parquet and imported into a fresh table, which must equal the original as a set, before and after a DELETE; a nested file written by pyarrow must import as the generator's rows. Sets are hashed in SQL as the shell's `pgc_set_hash` does, a failed query reading as a unique `query_error()`. On a private database, with the files in `server_dir`. + +### Every test + +| test | what it holds | +| --- | --- | +| `test_the_nested_table_exports_every_row` | export rows | +| `test_the_nested_file_imports_every_row` | import rows | +| `test_the_nested_parquet_import_round_trips` | nested import round-trips (arrays + composite) | +| `test_the_parquet_import_count_matches` | import count matches -- compared as numbers, since two failed counts agreed as text | +| `test_a_fresh_round_trip_after_a_delete_matches` | round-trips after delete | +| `test_a_pyarrow_nested_file_imports_every_row` | pyarrow import rows; declines by this name without pyarrow | +| `test_a_pyarrow_nested_file_matches_the_oracle` | pyarrow nested file matches oracle; declines by this name without pyarrow | + +## 260. test_parquet_count_bounds.py: file-declared Parquet counts are bounded before use + +The pytest twin of `test/parquet_count_bounds.sh` (ported for #1341), every check by the shell's name. With the shell's crafter, which patches the footer counts of an honest pyarrow file, a column chunk's num_values of 2^62 and a row group's num_rows past its values (scalar and LIST) must each be rejected as a malformed row group whose detail names the count, without phantom rows; the honest files import. A survival check passes only when the server answers AND the log's count of killed backends has not moved since setup, since a fresh connection answers again after crash recovery. Every test declines by the shell's skip name without pyarrow. On a private database. + +### Every test + +| test | what it holds | +| --- | --- | +| `test_the_crafter_produced_the_files` | premise: the crafter produced the seven files | +| `test_the_server_is_up_before_any_malformed_file` | premise: the server is up before any malformed file is read | +| `test_an_honest_file_imports_its_rows` | an honest file imports its rows | +| `test_the_server_is_up_after_the_honest_import` | the server is still up after the honest import | +| `test_a_wrapping_num_values_is_rejected` | num_values 2^62 is rejected, not imported | +| `test_the_backend_survives_the_wrapping_num_values` | and the backend is still up (it did not wrap-allocate and crash) | +| `test_num_rows_past_the_values_is_rejected` | num_rows past the chunk's values is rejected, not emitted as phantom rows | +| `test_num_rows_past_the_values_returns_no_phantom_rows` | and it did not silently return more rows than the file holds -- any count above 8 | +| `test_a_huge_num_rows_is_rejected` | a huge num_rows is rejected rather than walking off the allocation | +| `test_the_backend_survives_the_huge_num_rows` | and the backend survived it | +| `test_the_num_values_rejection_names_num_values` | the num_values rejection names num_values in its detail | +| `test_the_num_rows_rejection_names_num_rows` | the num_rows rejection names num_rows in its detail | +| `test_an_honest_list_imports_one_row` | an honest one-record list imports one row | +| `test_a_valid_three_record_list_imports_three_rows` | a valid three-record list imports three rows (not over-rejected) | +| `test_num_rows_past_a_lists_records_is_rejected` | num_rows past a list's record count is rejected, not emitted as phantom rows | +| `test_num_rows_past_a_lists_records_returns_no_phantom_rows` | and it did not silently return more rows than the file's records -- any count above 1 | +| `test_the_backend_survives_the_list_over_read` | and the backend survived the list over-read | diff --git a/test/pytest/expected_tests.txt b/test/pytest/expected_tests.txt index 8125b02e2..9d607f3ee 100644 --- a/test/pytest/expected_tests.txt +++ b/test/pytest/expected_tests.txt @@ -306,7 +306,9 @@ # guard_tests 465 -> 467 for #1339's test_residual_is_counted.py :143: the new # test_the_tally_function_records_every_suite_that_ran_or_skipped and # test_every_comm_in_the_runner_pins_its_collation, by collection over NO_CLUSTER. -guard_tests 467 +# guard_tests 467 -> 474 for #1341's test_parquet_level_width.py, the twin of parquet_level_width.sh, +# which needs no cluster: 7 tests, by collection over NO_CLUSTER. +guard_tests 474 # The complement: tests that need the driver and a throwaway cluster. Until #1016 these ran # in no CI job at all -- a quarter of the corpus, green when somebody ran them by hand and @@ -1009,4 +1011,12 @@ guard_tests 467 # 10 tests, one per check, by collection over the cluster files. # cluster_tests 2441 -> 2454 for #1341's test_encode_invariants.py, the twin of encode_invariants.sh: # 13 tests, one per check, by collection over the cluster files. -cluster_tests 2454 +# cluster_tests 2454 -> 2467 for #1341's test_read_stream.py, the twin of read_stream.sh: +# 13 tests, one per check, by collection over the cluster files. +# cluster_tests 2467 -> 2473 for #1341's test_parquet_nested.py, the twin of parquet_nested.sh: +# 6 tests, one per check, by collection over the cluster files. +# cluster_tests 2473 -> 2480 for #1341's test_parquet_nested_import.py, the twin of parquet_nested_import.sh: +# 7 tests, one per check, by collection over the cluster files. +# cluster_tests 2480 -> 2497 for #1341's test_parquet_count_bounds.py, the twin of parquet_count_bounds.sh: +# 17 tests, one per check, by collection over the cluster files. +cluster_tests 2497 diff --git a/test/pytest/test_compare_to_bash.py b/test/pytest/test_compare_to_bash.py index 27c398215..49b391bf9 100644 --- a/test/pytest/test_compare_to_bash.py +++ b/test/pytest/test_compare_to_bash.py @@ -91,6 +91,11 @@ "smoke", "sort_status", "sort_status_privilege", + "parquet_count_bounds", + "parquet_nested_import", + "parquet_nested", + "parquet_level_width", + "read_stream", "encode_invariants", "cost_written_geometry", "pg19_vacuum_options", diff --git a/test/pytest/test_harness_deps.py b/test/pytest/test_harness_deps.py index 65045baf9..bc4b66c73 100644 --- a/test/pytest/test_harness_deps.py +++ b/test/pytest/test_harness_deps.py @@ -150,6 +150,7 @@ # #1341's twin of wal_envelope.sh. It reads src/*.c as text with its own Python # reader -- no connection, no cluster, no driver -- and the classifier agrees. "test_wal_envelope.py", + "test_parquet_level_width.py", ] diff --git a/test/pytest/test_parquet_count_bounds.py b/test/pytest/test_parquet_count_bounds.py new file mode 100644 index 000000000..b0949bcd6 --- /dev/null +++ b/test/pytest/test_parquet_count_bounds.py @@ -0,0 +1,242 @@ +"""File-declared Parquet counts are bounded before use (#570, #571). The pytest twin of +`test/parquet_count_bounds.sh`, every check by the shell's name (#1341). + +The reader takes a column chunk's num_values and a row group's num_rows off the wire. +A num_values of 2^62 used to wrap the decode arrays' size to zero, and a num_rows past a +chunk's values walked the row loop off the decoded arrays. Using the shell's crafter, +which patches those footer varints in an honest pyarrow file, this asserts: the honest +file imports its 8 rows; each malformed file is rejected as a malformed row group whose +detail names the count, without returning phantom rows; a repeated (LIST) column's +num_rows past its record count is rejected too, while a valid three-record list is not; +and no backend dies on the way. + +THE BACKEND MUST NOT HAVE DIED, not merely answer now: a fresh connection answers +again after crash recovery, so the survival checks also require the server log's count +of killed backends not to have moved since setup. A phantom-row count is any count +above the file's true one, not only the one the crafter declares. + +ON A PRIVATE DATABASE. The shell skips whole without pyarrow, so every test declines by +the shell's skip name first, and only then asks for the fixture, whose crafter needs +pyarrow. The crafted files live in `server_dir`. +""" +import pathlib +import re +import subprocess +import sys + +import psycopg +import pytest + +from pgc_vacuity import query_error + +DB = "pgc_parquet_count_bounds" +CRAFTER = pathlib.Path(__file__).resolve().parents[1] / "craft_parquet_counts.py" +FILES = ("honest", "bignv", "bignr", "hugenr", "list_honest", "list_valid3", "list_bad") + + +def _crashes(cluster): + try: + text = (cluster.datadir / "server.log").read_text(errors="replace") + except OSError as exc: + return query_error(f"server log unreadable: {exc}") + return sum(1 for line in text.splitlines() if "terminated by signal" in line) + + +def _alive(cluster, dsn, before): + """-> "1" when the server answers and no backend died since setup, as the shell's alive().""" + try: + with psycopg.connect(dsn, autocommit=True, connect_timeout=10) as conn: + up = str(conn.execute("SELECT 1").fetchone()[0]) + except psycopg.Error as exc: + up = str(exc).splitlines()[0] if str(exc) else type(exc).__name__ + now = _crashes(cluster) + if not (isinstance(before, int) and isinstance(now, int)): + return "no (crash count unreadable)" + if up == "1" and now == before: + return "1" + return f"no ([{up}], {now - before} backend crashes)" + + +def _import(dsn, table, coltype, path): + """-> the imported row count as text, or "ERR: " plus "\\n".""" + try: + with psycopg.connect(dsn, autocommit=True, connect_timeout=10) as conn: + conn.execute(f"DROP TABLE IF EXISTS {table}") + conn.execute(f"CREATE TABLE {table} (a {coltype}) USING pgcolumnar") + conn.execute(f"SELECT pgcolumnar.import_parquet('{table}', '{path}')") + return str(conn.execute(f"SELECT count(*) FROM {table}").fetchone()[0]) + except psycopg.Error as exc: + detail = getattr(getattr(exc, "diag", None), "message_detail", None) or "" + return f"ERR: {str(exc).splitlines()[0] if str(exc) else type(exc).__name__}\n{detail}" + + +def _rejected(reading): + return "rejected" if "malformed row group" in reading.lower() else reading.splitlines()[0] + + +def _leak(reading, true_rows): + return f"leaked ({reading} rows)" if reading.isdigit() and int(reading) > true_rows else "ok" + + +@pytest.fixture(scope="module") +def pcb(pgc_cluster, server_dir): + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + a.execute(f"CREATE DATABASE {DB}") + try: + dsn = pgc_cluster.dsn(dbname=DB) + w = server_dir / "pcb" + w.mkdir(exist_ok=True) + w.chmod(0o777) + crafted = subprocess.run([sys.executable, str(CRAFTER), str(w)], capture_output=True, text=True) + for f in w.iterdir(): + f.chmod(0o644) + with psycopg.connect(dsn, autocommit=True) as conn: + conn.execute("CREATE EXTENSION IF NOT EXISTS pgcolumnar") + before = _crashes(pgc_cluster) + missing = [f for f in FILES if not (w / f"{f}.parquet").is_file() or (w / f"{f}.parquet").stat().st_size == 0] + r = {"crafted": "yes" if not missing else "missing: " + " ".join(missing), + "crafter": crafted.stderr.strip().splitlines()[-1:] if crafted.returncode else [], + "up before": _alive(pgc_cluster, dsn, before)} + r["honest"] = _import(dsn, "pcb", "int", w / "honest.parquet") + r["up honest"] = _alive(pgc_cluster, dsn, before) + r["bignv"] = _import(dsn, "pcb", "int", w / "bignv.parquet") + r["up bignv"] = _alive(pgc_cluster, dsn, before) + r["bignr"] = _import(dsn, "pcb", "int", w / "bignr.parquet") + r["hugenr"] = _import(dsn, "pcb", "int", w / "hugenr.parquet") + r["up hugenr"] = _alive(pgc_cluster, dsn, before) + r["list honest"] = _import(dsn, "pcl", "int[]", w / "list_honest.parquet") + r["list valid3"] = _import(dsn, "pcl", "int[]", w / "list_valid3.parquet") + r["list bad"] = _import(dsn, "pcl", "int[]", w / "list_bad.parquet") + r["up list"] = _alive(pgc_cluster, dsn, before) + yield r + finally: + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + + +def _int(v): + return int(v) if isinstance(v, str) and v.isdigit() else v + + +def test_the_crafter_produced_the_files(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(pcb["crafted"], "yes", "premise: the crafter produced the seven files") + + +def test_the_server_is_up_before_any_malformed_file(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(pcb["up before"], "1", "premise: the server is up before any malformed file is read") + + +def test_an_honest_file_imports_its_rows(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.num(_int(pcb["honest"]), 8, "an honest file imports its rows") + + +def test_the_server_is_up_after_the_honest_import(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(pcb["up honest"], "1", "the server is still up after the honest import") + + +def test_a_wrapping_num_values_is_rejected(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(_rejected(pcb["bignv"]), "rejected", "num_values 2^62 is rejected, not imported") + + +def test_the_backend_survives_the_wrapping_num_values(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(pcb["up bignv"], "1", "and the backend is still up (it did not wrap-allocate and crash)") + + +def test_num_rows_past_the_values_is_rejected(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(_rejected(pcb["bignr"]), "rejected", + "num_rows past the chunk's values is rejected, not emitted as phantom rows") + + +def test_num_rows_past_the_values_returns_no_phantom_rows(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(_leak(pcb["bignr"], 8), "ok", "and it did not silently return more rows than the file holds") + + +def test_a_huge_num_rows_is_rejected(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(_rejected(pcb["hugenr"]), "rejected", + "a huge num_rows is rejected rather than walking off the allocation") + + +def test_the_backend_survives_the_huge_num_rows(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(pcb["up hugenr"], "1", "and the backend survived it") + + +def test_the_num_values_rejection_names_num_values(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.num(1 if "num_values" in pcb["bignv"] else 0, 1, + "the num_values rejection names num_values in its detail") + + +def test_the_num_rows_rejection_names_num_rows(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.num(1 if "num_rows" in pcb["bignr"] else 0, 1, "the num_rows rejection names num_rows in its detail") + + +def test_an_honest_list_imports_one_row(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.num(_int(pcb["list honest"]), 1, "an honest one-record list imports one row") + + +def test_a_valid_three_record_list_imports_three_rows(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.num(_int(pcb["list valid3"]), 3, "a valid three-record list imports three rows (not over-rejected)") + + +def test_num_rows_past_a_lists_records_is_rejected(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(_rejected(pcb["list bad"]), "rejected", + "num_rows past a list's record count is rejected, not emitted as phantom rows") + + +def test_num_rows_past_a_lists_records_returns_no_phantom_rows(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(_leak(pcb["list bad"], 1), "ok", "and it did not silently return more rows than the file's records") + + +def test_the_backend_survives_the_list_over_read(request, expect): + if expect.require_module("pyarrow.parquet", name="pyarrow not available; Parquet count-bounds suite needs it") is None: + return + pcb = request.getfixturevalue("pcb") # after the decline: the crafter needs pyarrow + expect.text(pcb["up list"], "1", "and the backend survived the list over-read") diff --git a/test/pytest/test_parquet_level_width.py b/test/pytest/test_parquet_level_width.py new file mode 100644 index 000000000..380a7111c --- /dev/null +++ b/test/pytest/test_parquet_level_width.py @@ -0,0 +1,78 @@ +"""The Parquet level-width helper is single-sourced, unsigned and bounded (#710). The +pytest twin of `test/parquet_level_width.sh`, every check by the shell's name (#1341). + +pq_bits_for gives the bit width of the RLE/bit-pack level encoding. It used to exist +twice, in the writer and the reader, and to shift a signed 1, which is undefined once +the shift reaches 31. Neither is reachable from SQL, so this reads the source, as the +shell does: the shared helper in columnar_parquet_format.h exists, no local copy +survives in either .c file and both call the shared one, its shift is unsigned with no +signed `1 <<` left, and its loop is bounded below 32 shifts with a non-positive maximum +returning before any shift. + +INDEPENDENT OF THE SHELL. The shell greps and awks; this reads the files in Python and +matches with `re`. Both read the same tree. + +FAILS CLOSED. A count over a file that is missing, or over a body that was not found, is +0, which is the passing answer for two of these checks; each reads "no reading" instead. +""" +import pathlib +import re + +SRC = pathlib.Path(__file__).resolve().parents[2] / "src" +HDR = SRC / "columnar_parquet_format.h" +WRITER = SRC / "columnar_parquet.c" +READER = SRC / "columnar_parquet_reader.c" + + +def _text(path): + try: + return path.read_text() + except OSError: + return None + + +def _body(): + """pq_bits_for's definition, from its signature line to the first closing brace at + column 0, as the shell's awk takes it, with its comments blanked (a guard commented + out of the code is not a guard); empty when the header or the function is not + there.""" + text = _text(HDR) or "" + m = re.search(r"^pq_bits_for\(int maxval\).*?^}", text, re.M | re.S) + return re.sub(r"//[^\n]*", "", re.sub(r"/\*.*?\*/", "", m.group(0), flags=re.S)) if m else "" + + +def test_the_shared_helper_exists(expect): + expect.text("yes" if _body() else "no", "yes", "the shared level-width helper exists") + + +def test_no_local_copy_survives(expect): + writer, reader = _text(WRITER), _text(READER) + if writer is None or reader is None: + got = "no reading (a source file is missing)" + else: + got = str(sum(len(re.findall(r"^[a-z_]*bits_for\(int", t, re.M)) for t in (writer, reader))) + expect.text(got, "0", "no local copy of the helper survives in the writer or the reader") + + +def test_both_callers_reach_the_shared_helper(expect): + writer, reader = _text(WRITER) or "", _text(READER) or "" + expect.text("yes" if "pq_bits_for(" in writer and "pq_bits_for(" in reader else "no", "yes", + "and both callers reach the shared one") + + +def test_the_shift_is_unsigned(expect): + expect.num(_body().count("1u <<"), 1, "the shift is unsigned") + + +def test_no_signed_shift_remains(expect): + body = _body() + got = (str(len(re.findall(r"\(1 <<|[^u] 1 <<", body))) if body else "no reading (no body)") + expect.text(got, "0", "and no signed 1 << remains in it") + + +def test_the_loop_is_bounded(expect): + expect.num(_body().count("b < 31"), 1, "the loop is bounded below 32 shifts") + + +def test_a_non_positive_maximum_returns_first(expect): + expect.num(_body().count("maxval <= 0"), 1, "a non-positive maximum returns before shifting at all") diff --git a/test/pytest/test_parquet_nested.py b/test/pytest/test_parquet_nested.py new file mode 100644 index 000000000..15369dda2 --- /dev/null +++ b/test/pytest/test_parquet_nested.py @@ -0,0 +1,178 @@ +"""Parquet nested-type export: 1-D arrays become Parquet LIST and named composites Parquet +groups, with Dremel repetition and definition levels (gap 27). The pytest twin of +`test/parquet_nested.sh`, every check by the shell's name (#1341). + +A columnar table with int[], text[] and composite columns, NULL arrays, empty arrays, +NULL elements and NULL composites among them, is exported; pyarrow reads the file back, +the field types must be list/list/struct, and every one of the 5000 rows must equal the +value recomputed from the generator. A 70,000-row table of arrays spans several row +groups and must round-trip whole, and a multi-dimensional array is refused rather than +flattened. + +EVERY ID, ONCE. The value comparison only visits rows the file holds, so an empty file +matched; it now requires ids 1..5000. And the refusal must be the multi-dimensional one: +any error used to count, a missing table's included. + +ON A PRIVATE DATABASE, built once by a module fixture in the shell's order; each test +asserts one reading, so one red cannot stop the next. The shell skips whole without +pyarrow, so every test declines by the shell's skip name first. The files live in +`server_dir`, because the backend writes them. A failed query is a `query_error()`. +""" +import psycopg +import pytest + +from pgc_vacuity import query_error + +DB = "pgc_parquet_nested" +MULTIDIM = "does not support multi-dimensional arrays" + + +def _failed(exc): + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + + +def _value(dsn, sql): + """-> the first column of the first row as text, or the failure as a query_error().""" + try: + with psycopg.connect(dsn, autocommit=True) as conn: + row = conn.execute(sql).fetchone() + except psycopg.Error as exc: + return _failed(exc) + return "" if row is None or row[0] is None else str(row[0]) + + +def _run(dsn, sql): + """Run one statement and swallow its failure, as the shell's psql_run does: a failed + step shows up in the readings after it.""" + try: + with psycopg.connect(dsn, autocommit=True) as conn: + conn.execute(sql) + except psycopg.Error: + pass + + +def _refusal(dsn, sql): + """-> "refused: multi-dimensional", "succeeded", or the other error.""" + try: + with psycopg.connect(dsn, autocommit=True) as conn: + conn.execute(sql) + except psycopg.Error as exc: + msg = str(exc) + return "refused: multi-dimensional" if MULTIDIM in msg else f"other error: {msg.splitlines()[0] if msg else type(exc).__name__}" + return "succeeded" + + +@pytest.fixture(scope="module") +def nested(pgc_cluster, server_dir): + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + a.execute(f"CREATE DATABASE {DB}") + try: + dsn = pgc_cluster.dsn(dbname=DB) + nf, nf2 = server_dir / "nt.parquet", server_dir / "np2.parquet" + _run(dsn, "CREATE EXTENSION IF NOT EXISTS pgcolumnar") + _run(dsn, "CREATE TYPE npc AS (x int, y text)") + _run(dsn, "CREATE TABLE np (id int, ia int[], ta text[], c npc) USING pgcolumnar") + _run(dsn, "INSERT INTO np SELECT g, " + "CASE WHEN g % 10 = 0 THEN NULL WHEN g % 7 = 0 THEN '{}'::int[] " + "ELSE ARRAY[g, NULL, g + 3] END, " + "ARRAY['a' || g, 'b' || g], " + "CASE WHEN g % 5 = 0 THEN NULL ELSE ROW(g * 2, 'c' || g)::npc END " + "FROM generate_series(1, 5000) g") + r = {"file": nf, "file2": nf2, + "rows": _value(dsn, f"SELECT pgcolumnar.export_parquet('np', '{nf}')")} + _run(dsn, "CREATE TABLE np2 (id int, ia int[]) USING pgcolumnar") + _run(dsn, "INSERT INTO np2 SELECT g, ARRAY[g, g*2] FROM generate_series(1, 70000) g") + r["rows2"] = _value(dsn, f"SELECT pgcolumnar.export_parquet('np2', '{nf2}')") + _run(dsn, "CREATE TABLE npm (id int, m int[]) USING pgcolumnar") + _run(dsn, "INSERT INTO npm VALUES (1, ARRAY[[1,2],[3,4]])") + r["multidim"] = _refusal(dsn, f"SELECT pgcolumnar.export_parquet('npm', '{server_dir / 'npm.parquet'}')") + yield r + finally: + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + + +def _types(pa, pq, path): + s = pq.read_schema(str(path)) + ia, ta, c = s.field("ia").type, s.field("ta").type, s.field("c").type + return " ".join(str(v) for v in ( + pa.types.is_list(ia), ia.value_type, pa.types.is_list(ta), ta.value_type, + pa.types.is_struct(c), c.num_fields, + c.field(0).name, c.field(0).type, c.field(1).name, c.field(1).type)) + + +def _values(pq, path): + """-> "MATCH" when the file holds every id 1..5000 once and each row equals the + generator's, else what differs.""" + d = pq.read_table(str(path)).to_pydict() + if sorted(d["id"]) != list(range(1, 5001)): + return f"ROWS {len(d['id'])}, not ids 1..5000" + for i, g in enumerate(d["id"]): + ia = None if g % 10 == 0 else ([] if g % 7 == 0 else [g, None, g + 3]) + ta = [f"a{g}", f"b{g}"] + c = None if g % 5 == 0 else {"x": g * 2, "y": f"c{g}"} + if d["ia"][i] != ia or d["ta"][i] != ta or d["c"][i] != c: + return f"MISMATCH@{g}" + return "MATCH" + + +def _values2(pq, path): + d = pq.read_table(str(path)).to_pydict() + ok = all(d["ia"][i] == [d["id"][i], d["id"][i] * 2] for i in range(len(d["id"]))) + return "MATCH" if ok and len(d["id"]) == 70000 else "MISMATCH" + + +def _reading(read, pa): + """-> read(), or the failure to read the file as a query_error().""" + try: + return read() + except (OSError, KeyError, IndexError, ValueError, pa.ArrowException) as exc: + # a missing or unreadable file, or a field it lacks, is a failed reading + return query_error(f"{type(exc).__name__}: {exc}") + + +def _modules(expect): + pa = expect.require_module("pyarrow", name="pyarrow not available; skipping Parquet nested export verification") + pq = expect.require_module("pyarrow.parquet", name="pyarrow not available; skipping Parquet nested export verification") if pa is not None else None + return (pa, pq) if pq is not None else (None, None) + + +def test_the_nested_parquet_export_writes_every_row(nested, expect): + if expect.require_module("pyarrow", name="pyarrow not available; skipping Parquet nested export verification") is None: + return + expect.text(nested["rows"], "5000", "nested Parquet export rows written") + + +def test_the_parquet_fields_are_list_list_struct(nested, expect): + pa, pq = _modules(expect) + if pa is None: + return + expect.text(_reading(lambda: _types(pa, pq, nested["file"]), pa), "True int32 True string True 2 x int32 y string", + "nested Parquet field types") + + +def test_every_parquet_value_matches_the_generator(nested, expect): + pa, pq = _modules(expect) + if pa is None: + return + expect.text(_reading(lambda: _values(pq, nested["file"]), pa), "MATCH", "nested Parquet values match expectation") + + +def test_the_multi_row_group_export_writes_every_row(nested, expect): + if expect.require_module("pyarrow", name="pyarrow not available; skipping Parquet nested export verification") is None: + return + expect.text(nested["rows2"], "70000", "np2 export rows") + + +def test_the_multi_row_group_values_match(nested, expect): + pa, pq = _modules(expect) + if pa is None: + return + expect.text(_reading(lambda: _values2(pq, nested["file2"]), pa), "MATCH", "np2 values match") + + +def test_a_multidimensional_array_is_refused_for_parquet(nested, expect): + if expect.require_module("pyarrow", name="pyarrow not available; skipping Parquet nested export verification") is None: + return + expect.text(nested["multidim"], "refused: multi-dimensional", "reject multi-dim array") diff --git a/test/pytest/test_parquet_nested_import.py b/test/pytest/test_parquet_nested_import.py new file mode 100644 index 000000000..3fac5190f --- /dev/null +++ b/test/pytest/test_parquet_nested_import.py @@ -0,0 +1,165 @@ +"""Nested Parquet import round-trips arrays and composites (gap 27). The pytest twin of +`test/parquet_nested_import.sh`, every check by the shell's name (#1341). + +A table with int[], text[] and composite columns, NULL arrays, empty arrays, NULL +elements and NULL composites among them, is exported to Parquet and imported into a +fresh table, which must equal the original as a set; after a DELETE, a fresh export and +import must equal the live rows. A nested file written by pyarrow, the reference writer, +must import as the generator's rows, proving the Dremel levels are read rather than only +our own exporter's layout. + +A SET IS HASHED IN SQL, as the shell's pgc_set_hash does: md5 over the sorted row texts, +EMPTY for no rows, and a failed query is a query_error(), unique, so two failures never +agree. The row counts compare as numbers, so two failed counts no longer agree either. + +ON A PRIVATE DATABASE, built once by a module fixture in the shell's order. Only the two +pyarrow arms need pyarrow, and each declines by its own name without it, as the shell's +two skips do. The files live in `server_dir`, because the backend reads and writes them. +""" +import psycopg +import pytest + +from pgc_vacuity import query_error + +DB = "pgc_parquet_nested_import" +SEL = "SELECT id, ia::text, ta::text, c::text FROM" +SEL3 = "SELECT id, ia::text, c::text FROM" + + +def _failed(exc): + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + + +def _value(dsn, sql): + try: + with psycopg.connect(dsn, autocommit=True) as conn: + row = conn.execute(sql).fetchone() + except psycopg.Error as exc: + return _failed(exc) + return "" if row is None or row[0] is None else str(row[0]) + + +def _run(dsn, sql): + """Run one statement and swallow its failure, as the shell's psql_run does.""" + try: + with psycopg.connect(dsn, autocommit=True) as conn: + conn.execute(sql) + except psycopg.Error: + pass + + +def _set_hash(dsn, query): + return _value(dsn, "SELECT coalesce(md5(string_agg(t, chr(10) ORDER BY t)), 'EMPTY') " + f"FROM (SELECT _row::text AS t FROM ( {query} ) _row) _s") + + +def _int(v): + return int(v) if isinstance(v, str) and v.isdigit() else v + + +def _write_reference(path, pa, pq): + """Write the pyarrow nested file the shell writes, with the modules the test's + require_module returned: a fixture cannot decline, so it never imports pyarrow.""" + ids = list(range(1, 201)) + ia = [None if g % 10 == 0 else ([] if g % 7 == 0 else [g, None, g + 3]) for g in ids] + c = [None if g % 5 == 0 else {"x": g * 2, "y": "c" + str(g)} for g in ids] + t = pa.table({ + "id": pa.array(ids, pa.int32()), + "ia": pa.array(ia, pa.list_(pa.int32())), + "c": pa.array(c, pa.struct([("x", pa.int32()), ("y", pa.string())])), + }) + pq.write_table(t, str(path)) + path.chmod(0o644) + + +def _pyarrow_arm(ni, pa, pq): + """Run the shell's pyarrow arm once, on the first pyarrow test that gets here.""" + if "pyarrow import" not in ni: + dsn = ni["dsn"] + pyf = ni["dir"] / "pyarrow_nested.parquet" + _write_reference(pyf, pa, pq) + _run(dsn, "CREATE TABLE pna3 (id int, ia int[], c pnic) USING pgcolumnar") + ni["pyarrow import"] = _value(dsn, f"SELECT pgcolumnar.import_parquet('pna3', '{pyf}')") + _run(dsn, "CREATE TABLE pna3_o (id int, ia int[], c pnic)") + _run(dsn, "INSERT INTO pna3_o SELECT g, " + "CASE WHEN g % 10 = 0 THEN NULL WHEN g % 7 = 0 THEN '{}'::int[] " + "ELSE ARRAY[g, NULL, g + 3] END, " + "CASE WHEN g % 5 = 0 THEN NULL ELSE ROW(g * 2, 'c' || g)::pnic END " + "FROM generate_series(1, 200) g") + ni["pyarrow oracle"] = (_set_hash(dsn, f"{SEL3} pna3"), _set_hash(dsn, f"{SEL3} pna3_o")) + return ni + + +@pytest.fixture(scope="module") +def ni(pgc_cluster, server_dir): + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + a.execute(f"CREATE DATABASE {DB}") + try: + dsn = pgc_cluster.dsn(dbname=DB) + pf = server_dir / "ni.parquet" + _run(dsn, "CREATE EXTENSION IF NOT EXISTS pgcolumnar") + _run(dsn, "CREATE TYPE pnic AS (x int, y text)") + _run(dsn, "CREATE TABLE pna (id int, ia int[], ta text[], c pnic) USING pgcolumnar") + _run(dsn, "INSERT INTO pna SELECT g, " + "CASE WHEN g % 10 = 0 THEN NULL WHEN g % 7 = 0 THEN '{}'::int[] " + "ELSE ARRAY[g, NULL, g + 3] END, " + "ARRAY['a' || g, 'b' || g], " + "CASE WHEN g % 5 = 0 THEN NULL ELSE ROW(g * 2, 'c' || g)::pnic END " + "FROM generate_series(1, 5000) g") + r = {"export": _value(dsn, f"SELECT pgcolumnar.export_parquet('pna', '{pf}')")} + _run(dsn, "CREATE TABLE pna2 (id int, ia int[], ta text[], c pnic) USING pgcolumnar") + r["import"] = _value(dsn, f"SELECT pgcolumnar.import_parquet('pna2', '{pf}')") + r["round trip"] = (_set_hash(dsn, f"{SEL} pna2"), _set_hash(dsn, f"{SEL} pna")) + r["counts"] = (_value(dsn, "SELECT count(*) FROM pna2"), _value(dsn, "SELECT count(*) FROM pna")) + _run(dsn, "DELETE FROM pna WHERE id BETWEEN 1000 AND 2000") + _run(dsn, f"SELECT pgcolumnar.export_parquet('pna', '{pf}')") + _run(dsn, "TRUNCATE pna2") + _run(dsn, f"SELECT pgcolumnar.import_parquet('pna2', '{pf}')") + r["after delete"] = (_set_hash(dsn, f"{SEL} pna2"), _set_hash(dsn, f"{SEL} pna")) + + r["dsn"], r["dir"] = dsn, server_dir + yield r + finally: + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + + +def test_the_nested_table_exports_every_row(ni, expect): + expect.text(ni["export"], "5000", "export rows") + + +def test_the_nested_file_imports_every_row(ni, expect): + expect.text(ni["import"], "5000", "import rows") + + +def test_the_nested_parquet_import_round_trips(ni, expect): + got, want = ni["round trip"] + expect.text(got, want, "nested import round-trips (arrays + composite)") + + +def test_the_parquet_import_count_matches(ni, expect): + got, want = ni["counts"] + expect.num(_int(got), _int(want), "import count matches") + + +def test_a_fresh_round_trip_after_a_delete_matches(ni, expect): + got, want = ni["after delete"] + expect.text(got, want, "round-trips after delete") + + +def test_a_pyarrow_nested_file_imports_every_row(ni, expect): + pa = expect.require_module("pyarrow", name="pyarrow import rows") + pq = expect.require_module("pyarrow.parquet", name="pyarrow import rows") if pa else None + if pq is None: + return + expect.text(_pyarrow_arm(ni, pa, pq)["pyarrow import"], "200", "pyarrow import rows") + + +def test_a_pyarrow_nested_file_matches_the_oracle(ni, expect): + pa = expect.require_module("pyarrow", name="pyarrow nested file matches oracle") + pq = expect.require_module("pyarrow.parquet", name="pyarrow nested file matches oracle") if pa else None + if pq is None: + return + got, want = _pyarrow_arm(ni, pa, pq)["pyarrow oracle"] + expect.text(got, want, "pyarrow nested file matches oracle") diff --git a/test/pytest/test_read_stream.py b/test/pytest/test_read_stream.py new file mode 100644 index 000000000..d4dce1fa6 --- /dev/null +++ b/test/pytest/test_read_stream.py @@ -0,0 +1,116 @@ +"""Reads through PostgreSQL's read stream return what the synchronous path returns +(#1341's twin of `test/read_stream.sh`, every check by the shell's name). + +`pgcolumnar.enable_read_stream` routes a row group's block reads through a read stream +(PostgreSQL 17+), which can prefetch and, on 18+, use asynchronous I/O. Five query shapes +over a multi-stripe table are compared with a heap twin loaded from the same generator, +once with the setting on and once off, so each path is checked on its own. + +THE SETTING IS READ BACK BEFORE EACH MODE. It arrives by ALTER DATABASE SET, whose +failure the shell's q drops, so a setting that never took ran both modes down the same +path: with it misspelled and the synchronous path broken, the shell passed 11 of 11. +A comparison is a set hash of each side, and a failed side reads as its own error. +A setting that cannot be applied is the read-back's value, so those two red by name. + +WHAT THE [off] ARMS VOUCH FOR. The read-back proves the setting reached the session, +not which path a read took; nothing here observes the path. With the stream path taken +whatever the setting says, and the synchronous path copying one byte late, all 13 pass. +So the [off] arms vouch for the synchronous path only while the setting routes reads. +On PostgreSQL 15 and 16 the stream path is compiled out: [on] and [off] run the same code. + +ON A PRIVATE DATABASE, built by a module fixture in the shell's order. +""" +import hashlib + +import psycopg +import pytest + +from pgc_vacuity import query_error + +DB = "pgc_read_stream" +SHAPES = { + "full scan": "SELECT id, k, v, t FROM %T", + "filtered": "SELECT id, v FROM %T WHERE k = 7", + "range": "SELECT id, t FROM %T WHERE id BETWEEN 10000 AND 20000", + "aggregate": "SELECT k, count(*), sum(v) FROM %T GROUP BY k", + "wide sample": "SELECT * FROM %T WHERE id % 997 = 0", +} +NAMES = [ + "[on] full scan", "[on] filtered", "[on] range", "[on] aggregate", "[on] wide sample", + "[off] full scan", "[off] filtered", "[off] range", "[off] aggregate", "[off] wide sample", +] + + +def _set_hash(dsn, sql): + """-> an order-free hash of the rows, or a failed query.""" + try: + with psycopg.connect(dsn, autocommit=True) as conn: + rows = sorted(repr(r) for r in conn.execute(sql).fetchall()) + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + return f"{len(rows)}:{hashlib.md5(chr(10).join(rows).encode()).hexdigest()}" + + +def _one(dsn, sql): + try: + with psycopg.connect(dsn, autocommit=True) as conn: + return str(conn.execute(sql).fetchone()[0]) + except psycopg.Error as exc: + return query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + + +@pytest.fixture(scope="module") +def rs(pgc_cluster): + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + a.execute(f"CREATE DATABASE {DB}") + try: + dsn = pgc_cluster.dsn(dbname=DB) + with psycopg.connect(dsn, autocommit=True) as conn: + conn.execute("CREATE EXTENSION IF NOT EXISTS pgcolumnar") + conn.execute("CREATE TABLE t_heap (id int, k int, v bigint, t text) USING heap") + conn.execute("CREATE TABLE t_col (id int, k int, v bigint, t text) USING pgcolumnar") + conn.execute("SELECT pgcolumnar.set_options('t_col', stripe_row_limit => 2000, chunk_group_row_limit => 1000)") + conn.execute("INSERT INTO t_heap SELECT g, g%100, g::bigint*3, repeat('x', (g%40)+1) " + "FROM generate_series(1,50000) g") + conn.execute("INSERT INTO t_col SELECT * FROM t_heap") + r = {"stripes": _one(dsn, "SELECT count(*) FROM pgcolumnar.row_group " + "WHERE storage_id = pgcolumnar.get_storage_id('t_col')")} + for mode in ("on", "off"): + # The setting failing to apply is what the read-back checks test, so its + # error is a value they red on by name (#1522), and the comparisons still run. + failed = None + try: + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"ALTER DATABASE {DB} SET pgcolumnar.enable_read_stream = {mode}") + except psycopg.Error as exc: + failed = query_error(str(exc).splitlines()[0] if str(exc) else type(exc).__name__) + r[f"setting {mode}"] = failed or _one(dsn, "SHOW pgcolumnar.enable_read_stream") + for label, sql in SHAPES.items(): + h = _set_hash(dsn, sql.replace("%T", "t_heap")) + c = _set_hash(dsn, sql.replace("%T", "t_col")) + r[f"[{mode}] {label}"] = (c, h) + yield r + finally: + with psycopg.connect(pgc_cluster.dsn(), autocommit=True) as a: + a.execute(f"DROP DATABASE IF EXISTS {DB} WITH (FORCE)") + + +def test_multiple_stripes_present(rs, expect): + n = rs["stripes"] + got = ("yes" if int(n) >= 10 else f"no ({n})") if n.isdigit() else f"no ({n})" + expect.text(got, "yes", "multiple stripes present") + + +def test_a_fresh_session_reads_the_setting_on(rs, expect): + expect.text(rs["setting on"], "on", "[on] a fresh session reads enable_read_stream as on") + + +def test_a_fresh_session_reads_the_setting_off(rs, expect): + expect.text(rs["setting off"], "off", "[off] a fresh session reads enable_read_stream as off") + + +@pytest.mark.parametrize("name", NAMES) +def test_a_shape_matches_its_heap_twin(rs, name, expect): + got, want = rs[name] + expect.text(got, want, name) diff --git a/test/pytest/unported_suites.txt b/test/pytest/unported_suites.txt index d310c618f..2748bb1da 100644 --- a/test/pytest/unported_suites.txt +++ b/test/pytest/unported_suites.txt @@ -82,15 +82,10 @@ parallel_degree parallel_export_parquet parallel_flush_optin parallel_vector_agg -parquet_count_bounds parquet_export parquet_export_stats -parquet_level_width -parquet_nested -parquet_nested_import phase2 phase3 phase4 phase6 planner_choice_quality -read_stream diff --git a/test/read_stream.sh b/test/read_stream.sh index b41895b33..dc057876a 100755 --- a/test/read_stream.sh +++ b/test/read_stream.sh @@ -23,18 +23,33 @@ pgc_setup "${1:-/usr/local/pg17/bin/pg_config}" make_pair "id int, k int, v bigint, t text" q "SELECT pgcolumnar.set_options('t_col', stripe_row_limit => 2000, chunk_group_row_limit => 1000);" >/dev/null load_pair "SELECT g, g%100, g::bigint*3, repeat('x', (g%40)+1) FROM generate_series(1,50000) g" -check "multiple stripes present" "$([ "$(stripe_count t_col)" -ge 10 ] && echo yes || echo no)" "yes" +NSTRIPES="$(stripe_count t_col)" +check "multiple stripes present" "$([ "${NSTRIPES:-0}" -ge 10 ] 2>/dev/null && echo yes || echo "no ($NSTRIPES)")" "yes" setg() { q "ALTER DATABASE $PGC_DB SET $1 = $2;" >/dev/null; } -for mode in on off; do - setg pgcolumnar.enable_read_stream "$mode" - diff_query "[$mode] full scan" "SELECT id, k, v, t FROM %T" - diff_query "[$mode] filtered" "SELECT id, v FROM %T WHERE k = 7" - diff_query "[$mode] range" "SELECT id, t FROM %T WHERE id BETWEEN 10000 AND 20000" - diff_query "[$mode] aggregate" "SELECT k, count(*), sum(v) FROM %T GROUP BY k" - diff_query "[$mode] wide sample" "SELECT * FROM %T WHERE id % 997 = 0" -done +# Each mode by literal name (#1341). ALTER DATABASE SET reaches every new session, +# and q drops its error, so a setting that never took would run both modes down the +# same path: each mode first reads the setting back from a fresh session. +# The read-back proves the setting reached the session, not which path a read took, +# and nothing here observes the path: with the stream path taken whatever the setting +# says and the synchronous path copying one byte late, all 13 checks pass. So the +# [off] arms vouch for the synchronous path only while the setting routes reads. On +# PostgreSQL 15 and 16 the stream path is compiled out, and [on] repeats [off]. +setg pgcolumnar.enable_read_stream on +check "[on] a fresh session reads enable_read_stream as on" "$(q "SHOW pgcolumnar.enable_read_stream;")" "on" +diff_query "[on] full scan" "SELECT id, k, v, t FROM %T" +diff_query "[on] filtered" "SELECT id, v FROM %T WHERE k = 7" +diff_query "[on] range" "SELECT id, t FROM %T WHERE id BETWEEN 10000 AND 20000" +diff_query "[on] aggregate" "SELECT k, count(*), sum(v) FROM %T GROUP BY k" +diff_query "[on] wide sample" "SELECT * FROM %T WHERE id % 997 = 0" +setg pgcolumnar.enable_read_stream off +check "[off] a fresh session reads enable_read_stream as off" "$(q "SHOW pgcolumnar.enable_read_stream;")" "off" +diff_query "[off] full scan" "SELECT id, k, v, t FROM %T" +diff_query "[off] filtered" "SELECT id, v FROM %T WHERE k = 7" +diff_query "[off] range" "SELECT id, t FROM %T WHERE id BETWEEN 10000 AND 20000" +diff_query "[off] aggregate" "SELECT k, count(*), sum(v) FROM %T GROUP BY k" +diff_query "[off] wide sample" "SELECT * FROM %T WHERE id % 997 = 0" setg pgcolumnar.enable_read_stream on q "ALTER DATABASE $PGC_DB RESET ALL;" >/dev/null