test: objstore_s3_read twin, and every refusal names its cause (#1341)

.github/workflows/ci.yml

@@ -185,6 +185,13 @@ jobs:
       - name: fetch the release tags, for the fixture-blob comparisons
         run: git fetch --depth=1 origin '+refs/tags/v1.0-alpha*:refs/tags/v1.0-alpha*'
 
+      - name: Resolve the object-storage suite host names
+        # test_objstore_addressing.py, the twin of objstore_addressing.sh, reads
+        # through virtual-host-style S3 addressing (pgc-bucket.s3.local) against
+        # its own loopback fixture server, as the suites job's shells do. Without
+        # the names its tests decline UNAVAILABLE_ENDPOINT and the run ends 67.
+        run: echo "127.0.0.1 s3.local pgc-bucket.s3.local" | sudo tee -a /etc/hosts
+
       - name: Add the PGDG repository and refresh the package lists
         run: |
           set -euo pipefail

.github/workflows/nightly.yml

@@ -745,6 +745,13 @@ jobs:
       - name: fetch the release tags, for the fixture-blob comparisons
         run: git fetch --depth=1 origin '+refs/tags/v1.0-alpha*:refs/tags/v1.0-alpha*'
 
+      - name: Resolve the object-storage suite host names
+        # test_objstore_addressing.py, the twin of objstore_addressing.sh, reads
+        # through virtual-host-style S3 addressing (pgc-bucket.s3.local) against
+        # its own loopback fixture server, as the suites job's shells do. Without
+        # the names its tests decline UNAVAILABLE_ENDPOINT and the run ends 67.
+        run: echo "127.0.0.1 s3.local pgc-bucket.s3.local" | sudo tee -a /etc/hosts
+
       - name: Add the PGDG repository and install PostgreSQL ${{ matrix.pg }} with its headers
         run: |
           set -euo pipefail

test/craft_parquet_codecs.py

@@ -0,0 +1,56 @@
+"""Write the native_parquet_codecs fixtures into DIR (argv[1]), for both twins.
+
+c_none / c_gzip / c_zstd / c_lz4.parquet hold the same 20000 rows, each written with one
+codec; codecs.txt records the compression pyarrow actually stored for each file's first
+column (one "<codec> <COMPRESSION>" line), so a test can prove which codec it read.
+badhdr.parquet is a gzip page whose header declares uncompressed_page_size -1.
+
+The rows are the generator both twins also compute in SQL:
+    id = g, grp = g % 50, val = g * 1.5, s = 'row-' || (g % 100),  g in 0 .. 19999
+Written fresh for pgColumnar.
+"""
+import sys
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+W = sys.argv[1]
+N = 20000
+tbl = pa.table({
+    "id": pa.array(range(N), pa.int32()),
+    "grp": pa.array([i % 50 for i in range(N)], pa.int32()),
+    "val": pa.array([i * 1.5 for i in range(N)], pa.float64()),
+    "s": pa.array([f"row-{i % 100}" for i in range(N)]),
+})
+with open(f"{W}/codecs.txt", "w") as out:
+    for codec in ["none", "gzip", "zstd", "lz4"]:
+        pq.write_table(tbl, f"{W}/c_{codec}.parquet", compression=codec)
+        c = pq.ParquetFile(f"{W}/c_{codec}.parquet").metadata.row_group(0).column(0).compression
+        out.write(f"{codec} {c}\n")
+
+# A gzip page whose header's uncompressed_page_size (field 2) is rewritten to zigzag(-1).
+pq.write_table(pa.table({"a": pa.array([1, 2, 3], pa.int32())}),
+               f"{W}/badhdr.parquet", compression="gzip", use_dictionary=False)
+raw = bytearray(open(f"{W}/badhdr.parquet", "rb").read())
+pos = 4                                   # the first PageHeader follows the "PAR1" magic
+last = 0
+patched = False
+for _ in range(6):
+    b = raw[pos]
+    pos += 1
+    t, d = b & 0x0F, b >> 4
+    if t == 0:
+        break
+    fid = last + d if d else None
+    last = fid
+    start = pos
+    while raw[pos] & 0x80:
+        pos += 1
+    pos += 1
+    if fid == 2:
+        assert pos - start == 1, (start, pos)
+        raw[start] = 0x01                 # zigzag(-1)
+        patched = True
+        break
+assert patched, "did not find uncompressed_page_size"
+open(f"{W}/badhdr.parquet", "wb").write(bytes(raw))

test/craft_parquet_hardening.py

@@ -0,0 +1,145 @@
+"""craft_parquet_hardening.py DIR: write the Parquet files test/native_parquet_hardening.sh
+and its pytest twin read: the narrowing binds (out of range and in range) and a
+two-column file whose row group carries one column chunk. One writer for both
+twins, so neither can drift from the other. Needs pyarrow."""
+import sys
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+W = sys.argv[1]
+
+# ---- narrowing: out-of-range values must raise, not wrap --------------------
+# int2 bound over INT32: 40000 does not fit int16 (would wrap to -25536)
+pq.write_table(pa.table({"c": pa.array([30000, 40000], pa.int32())}),
+               f"{W}/i2_bad.parquet")
+# same bind, all values in range: must keep working
+pq.write_table(pa.table({"c": pa.array([-30000, 0, 30000], pa.int32())}),
+               f"{W}/i2_ok.parquet")
+# time bound over INT64 micros: beyond one day is not a time
+pq.write_table(pa.table({"c": pa.array([0, 86400000000 + 1], pa.int64())}),
+               f"{W}/time_bad.parquet")
+pq.write_table(pa.table({"c": pa.array([0, 3600000000], pa.int64())}),
+               f"{W}/time_ok.parquet")
+# date bound over INT32 days since epoch. INT32 max does NOT overflow the epoch
+# offset (2147483647 - 10957 stays in range); it is rejected by IS_VALID_DATE.
+pq.write_table(pa.table({"c": pa.array([0, 2147483647], pa.int32())}),
+               f"{W}/date_bad.parquet")
+# INT32 min is what overflows the subtraction itself, exercising the other branch
+pq.write_table(pa.table({"c": pa.array([0, -2147483648], pa.int32())}),
+               f"{W}/date_ovf.parquet")
+pq.write_table(pa.table({"c": pa.array([0, 19000], pa.int32())}),
+               f"{W}/date_ok.parquet")
+# timestamp bound over INT64 micros: INT64 min overflows the epoch-offset
+# subtraction. INT64 max does NOT -- it still lands inside PostgreSQL's timestamp
+# range (year ~294276), so it is deliberately not the case tested here.
+pq.write_table(pa.table({"c": pa.array([0, -9223372036854775808], pa.int64())}),
+               f"{W}/ts_bad.parquet")
+pq.write_table(pa.table({"c": pa.array([0, 1700000000000000], pa.int64())}),
+               f"{W}/ts_ok.parquet")
+
+# ---- malformed row-group chunk list ----------------------------------------
+pq.write_table(pa.table({"a": pa.array([1, 2, 3], pa.int32()),
+                         "b": pa.array([4, 5, 6], pa.int32())}),
+               f"{W}/two.parquet")
+
+# Walk the Thrift-compact footer to the exact byte holding the row group's
+# `columns` list header, then claim it holds one fewer chunk than it does.
+# Blind byte-searching is not reliable here: the same byte value occurs in the
+# data pages, so the header has to be located structurally.
+raw = bytearray(open(f"{W}/two.parquet", "rb").read())
+flen = int.from_bytes(raw[-8:-4], "little")
+base = len(raw) - 8 - flen          # start of FileMetaData
+pos = base
+
+def varint():
+    global pos
+    shift = 0; out = 0
+    while True:
+        b = raw[pos]; pos += 1
+        out |= (b & 0x7F) << shift
+        if not b & 0x80:
+            return out
+        shift += 7
+
+def zigzag():
+    u = varint()
+    return (u >> 1) ^ -(u & 1)
+
+def list_hdr():
+    global pos
+    b = raw[pos]; pos += 1
+    size = (b >> 4) & 0x0F
+    et = b & 0x0F
+    if size == 0x0F:
+        size = varint()
+    return size, et
+
+def skip(t):
+    global pos
+    if t in (1, 2):          return                 # bool: value is in the header
+    if t == 3:               pos += 1               # byte
+    elif t in (4, 5, 6):     zigzag()               # i16/i32/i64
+    elif t == 7:             pos += 8               # double
+    elif t == 8:             pos += varint()        # binary
+    elif t in (9, 10):
+        n, et = list_hdr()
+        for _ in range(n): skip(et)
+    elif t == 12:
+        last = 0
+        while True:
+            ft, _fid, last = field(last)
+            if ft == 0: break
+            skip(ft)
+
+def field(last):
+    global pos
+    b = raw[pos]; pos += 1
+    if b == 0:
+        return 0, 0, last
+    t = b & 0x0F
+    delta = b >> 4
+    fid = last + delta if delta else zigzag()
+    return t, fid, fid
+
+# FileMetaData -> field 4 (row_groups) -> first RowGroup -> field 1 (columns)
+target = None
+last = 0
+while True:
+    ft, fid, last = field(last)
+    if ft == 0:
+        break
+    if fid == 4 and ft == 9:                 # row_groups: list<RowGroup>
+        nrg, _et = list_hdr()
+        assert nrg >= 1, nrg
+        rlast = 0
+        while True:
+            rft, rfid, rlast = field(rlast)
+            if rft == 0:
+                break
+            if rfid == 1 and rft == 9:       # columns: list<ColumnChunk>
+                target = pos                 # byte holding this list's header
+                n, et = list_hdr()
+                assert n == 2, f"expected 2 column chunks, got {n}"
+                skip(et)                     # past ColumnChunk[0]
+                start2 = pos
+                skip(et)                     # past ColumnChunk[1]
+                end2 = pos
+                break
+            skip(rft)
+        break
+    skip(ft)
+
+assert target is not None, "row group columns list not found"
+# Remove ColumnChunk[1] outright and say so in the header, so the footer stays a
+# well-formed Thrift struct. Decrementing the count alone would desync the parser
+# and fail for an unrelated reason; the only defect must be that the row group
+# carries fewer chunks than the schema has leaves.
+assert (raw[target] >> 4) == 2, hex(raw[target])
+raw[target] = (1 << 4) | (raw[target] & 0x0F)
+removed = end2 - start2
+del raw[start2:end2]
+raw[-8:-4] = (flen - removed).to_bytes(4, "little")
+open(f"{W}/short_chunks.parquet", "wb").write(bytes(raw))
+print(f"  columns list at {target}: 2 chunks -> 1, removed {removed} footer bytes",
+      file=sys.stderr)

test/craft_parquet_projection.py

@@ -0,0 +1,39 @@
+"""Write the files test/native_parquet_projection.sh and its pytest twin read.
+
+    python3 test/craft_parquet_projection.py DIR
+
+DIR/proj.parquet     a, b, c, d over 5000 rows in row groups of 1000, from the same
+                     generator the suites rebuild in SQL as their oracle:
+                     a = i, b = 2i, c = 's' || (i % 20), d = i / 3.0
+DIR/corrupt.parquet  g, bad over 1000 rows, with the second column's data page
+                     smashed, so decoding `bad` fails while `g` and the footer stay
+                     intact.
+
+Both harnesses run this one crafter, so a file can never differ between them.
+"""
+import sys
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+W = sys.argv[1]
+n = 5000
+pq.write_table(pa.table({
+    "a": pa.array(range(n), pa.int32()),
+    "b": pa.array([i * 2 for i in range(n)], pa.int64()),
+    "c": pa.array([f"s{i % 20}" for i in range(n)]),
+    "d": pa.array([i / 3.0 for i in range(n)], pa.float64()),
+}), f"{W}/proj.parquet", row_group_size=1000)
+
+pq.write_table(pa.table({"g": pa.array(range(1000), pa.int32()),
+                         "bad": pa.array(range(1000), pa.int32())}),
+               f"{W}/corrupt.parquet", row_group_size=1000, use_dictionary=False)
+md = pq.ParquetFile(f"{W}/corrupt.parquet").metadata.row_group(0).column(1)
+off = md.data_page_offset
+raw = bytearray(open(f"{W}/corrupt.parquet", "rb").read())
+# smash the second column's page header and the start of its data, leaving the
+# first column and the footer intact
+for i in range(off, off + 32):
+    raw[i] = 0xFF
+open(f"{W}/corrupt.parquet", "wb").write(bytes(raw))
+print(f"  corrupted column 'bad' at offset {off}", file=sys.stderr)