Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
6b6341b
Add lazy remote Parquet table access
FrancescAlted Sep 25, 2026
7a376ae
Size remote Parquet row group conversions
FrancescAlted Sep 25, 2026
b8fd39a
Project nested remote Parquet leaves
FrancescAlted Sep 25, 2026
1024eeb
Fix Parquet table info and plan cache unification
FrancescAlted Sep 25, 2026
c5bfa56
Unify Parquet caching and speed warm reads
FrancescAlted Sep 25, 2026
a025412
Simplify remote Parquet guide
FrancescAlted Sep 25, 2026
d57805a
Fix Parquet cache locking and archive handling
FrancescAlted Sep 25, 2026
bb8083e
Skip Parquet tests without optional dependencies
FrancescAlted Sep 25, 2026
b579301
Address Parquet review findings
FrancescAlted Sep 25, 2026
2e595b3
Run Parquet tests in native CI
FrancescAlted Sep 25, 2026
4ee1453
Handle Windows paths in Parquet caches
FrancescAlted Sep 25, 2026
40af8a5
Preserve Parquet source queries and reader options
FrancescAlted Sep 25, 2026
c235b21
Redact signed queries from Parquet source descriptions
FrancescAlted Sep 25, 2026
fbdd520
Count HTTP requests before sending test responses
FrancescAlted Sep 25, 2026
0fdc62c
Consolidate remote table examples and improve writes
FrancescAlted Sep 25, 2026
c4ebceb
Move remote Parquet tables under RemoteStore
FrancescAlted Sep 25, 2026
9cedd6e
Fix Windows Parquet shared cache source identity
FrancescAlted Sep 25, 2026
d57feb6
Avoid same-tick fsspec cache identity in Windows test
FrancescAlted Sep 25, 2026
742d2d1
Parallelize remote Parquet range reads
FrancescAlted Sep 26, 2026
021127d
Respect null masks in Column reads
FrancescAlted Sep 26, 2026
e024afa
Fix reopening root RemoteCTable artifacts
FrancescAlted Sep 28, 2026
69c2aab
Recover corrupt Parquet disk cache metadata
FrancescAlted Sep 28, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions bench/remote_parquet_chicago_http_capacity_hint.jsonl
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
{"file": "chicago-taxi-flat-f32-cl20.parquet", "file_bytes": 581387271, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.203, "cold_reads": 1, "cold_bytes": 331590, "cold_heads": 0, "cold_seconds": 0.08, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.067, "cache_bytes": 356304, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi-flat-f32.parquet", "file_bytes": 686857703, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.021, "cold_reads": 1, "cold_bytes": 352939, "cold_heads": 0, "cold_seconds": 0.083, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.067, "cache_bytes": 356304, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi-flat-f64.parquet", "file_bytes": 687450036, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.023, "cold_reads": 1, "cold_bytes": 352939, "cold_heads": 0, "cold_seconds": 0.085, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.066, "cache_bytes": 356304, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi.parquet", "file_bytes": 640173859, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 26, "open_bytes": 333139, "open_heads": 1, "open_seconds": 0.37, "cold_reads": 1, "cold_bytes": 6660483, "cold_heads": 0, "cold_seconds": 1.087, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.07, "cache_bytes": 12609432, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi-flat-f32-cl20.parquet", "file_bytes": 581387271, "rows": 24314322, "column": "company", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.205, "cold_reads": 1, "cold_bytes": 1831, "cold_heads": 0, "cold_seconds": 0.089, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.063, "cache_bytes": 1008, "warm_equal": true, "groups": 25, "arrow_equal": true}
5 changes: 5 additions & 0 deletions bench/remote_parquet_chicago_http_leaf_projection.jsonl
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
{"file": "chicago-taxi-flat-f32-cl20.parquet", "file_bytes": 581387271, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.193, "cold_reads": 1, "cold_bytes": 331590, "cold_heads": 0, "cold_seconds": 0.081, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.067, "cache_bytes": 356304, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi-flat-f32.parquet", "file_bytes": 686857703, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.029, "cold_reads": 1, "cold_bytes": 352939, "cold_heads": 0, "cold_seconds": 0.087, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.061, "cache_bytes": 356304, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi-flat-f64.parquet", "file_bytes": 687450036, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.025, "cold_reads": 1, "cold_bytes": 352939, "cold_heads": 0, "cold_seconds": 0.089, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.066, "cache_bytes": 356304, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi.parquet", "file_bytes": 640173859, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 26, "open_bytes": 333139, "open_heads": 1, "open_seconds": 0.355, "cold_reads": 1, "cold_bytes": 363880, "cold_heads": 0, "cold_seconds": 0.073, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.066, "cache_bytes": 356304, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi-flat-f32-cl20.parquet", "file_bytes": 581387271, "rows": 24314322, "column": "company", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.189, "cold_reads": 1, "cold_bytes": 1831, "cold_heads": 0, "cold_seconds": 0.094, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.067, "cache_bytes": 1008, "warm_equal": true, "groups": 25, "arrow_equal": true}
5 changes: 5 additions & 0 deletions bench/remote_parquet_chicago_http_optimized.jsonl
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
{"file": "chicago-taxi-flat-f32-cl20.parquet", "file_bytes": 581387271, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.188, "cold_reads": 1, "cold_bytes": 331590, "cold_heads": 0, "cold_seconds": 17.571, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.068, "cache_bytes": 34695136, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi-flat-f32.parquet", "file_bytes": 686857703, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.022, "cold_reads": 1, "cold_bytes": 352939, "cold_heads": 0, "cold_seconds": 17.638, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.068, "cache_bytes": 34695136, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi-flat-f64.parquet", "file_bytes": 687450036, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.016, "cold_reads": 1, "cold_bytes": 352939, "cold_heads": 0, "cold_seconds": 17.862, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.069, "cache_bytes": 34695136, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi.parquet", "file_bytes": 640173859, "rows": 24314322, "column": "trip.sec", "transport": "http", "open_reads": 26, "open_bytes": 333139, "open_heads": 1, "open_seconds": 0.364, "cold_reads": 1, "cold_bytes": 6660483, "cold_heads": 0, "cold_seconds": 1.083, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.068, "cache_bytes": 12632079, "warm_equal": true, "groups": 25, "arrow_equal": true}
{"file": "chicago-taxi-flat-f32-cl20.parquet", "file_bytes": 581387271, "rows": 24314322, "column": "company", "transport": "http", "open_reads": 1, "open_bytes": 65536, "open_heads": 1, "open_seconds": 0.189, "cold_reads": 1, "cold_bytes": 1831, "cold_heads": 0, "cold_seconds": 0.819, "warm_reads": 0, "warm_bytes": 0, "warm_heads": 0, "warm_seconds": 0.068, "cache_bytes": 34695431, "warm_equal": true, "groups": 25, "arrow_equal": true}
61 changes: 61 additions & 0 deletions bench/remote_parquet_chicago_http_results.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
{
"date": "2026-09-25",
"method": "Each file served by bench/remote_parquet_files.py --http on 127.0.0.1 with byte-range responses. Reads and bytes count completed server GET responses and response bodies; HEAD requests are counted separately. Headers and TCP overhead are excluded. Default in-memory converted cache; fsspec block_size=4096, cache_type=none. Cold access reads the final row; warm access repeats it in the same process.",
"files": [
{
"file": "chicago-taxi-flat-f32-cl20.parquet",
"column": "trip.sec",
"file_bytes": 581387271,
"rows": 24314322,
"groups": 25,
"open": {"gets": 2, "heads": 2, "bytes": 32679571, "seconds": 1.253},
"cold": {"gets": 1, "heads": 0, "bytes": 331590, "seconds": 17.688},
"warm": {"gets": 0, "heads": 0, "bytes": 0, "seconds": 0.069},
"arrow_equal": true
},
{
"file": "chicago-taxi-flat-f32.parquet",
"column": "trip.sec",
"file_bytes": 686857703,
"rows": 24314322,
"groups": 25,
"open": {"gets": 2, "heads": 2, "bytes": 33057447, "seconds": 1.084},
"cold": {"gets": 1, "heads": 0, "bytes": 352939, "seconds": 17.885},
"warm": {"gets": 0, "heads": 0, "bytes": 0, "seconds": 0.069},
"arrow_equal": true
},
{
"file": "chicago-taxi-flat-f64.parquet",
"column": "trip.sec",
"file_bytes": 687450036,
"rows": 24314322,
"groups": 25,
"open": {"gets": 2, "heads": 2, "bytes": 33068884, "seconds": 1.122},
"cold": {"gets": 1, "heads": 0, "bytes": 352939, "seconds": 18.131},
"warm": {"gets": 0, "heads": 0, "bytes": 0, "seconds": 0.069},
"arrow_equal": true
},
{
"file": "chicago-taxi-flat-f32-cl20.parquet",
"column": "company",
"file_bytes": 581387271,
"rows": 24314322,
"groups": 25,
"open": {"gets": 2, "heads": 2, "bytes": 32679571, "seconds": 1.308},
"cold": {"gets": 1, "heads": 0, "bytes": 1831, "seconds": 0.813},
"warm": {"gets": 0, "heads": 0, "bytes": 0, "seconds": 0.068},
"arrow_equal": true
},
{
"file": "chicago-taxi.parquet",
"column": "trip.sec",
"file_bytes": 640173859,
"rows": 24314322,
"groups": 25,
"open": {"gets": 27, "heads": 2, "bytes": 671548047, "seconds": 3.909},
"cold": {"gets": 1, "heads": 0, "bytes": 6660483, "seconds": 1.051},
"warm": {"gets": 0, "heads": 0, "bytes": 0, "seconds": 0.067},
"arrow_equal": true
}
]
}
61 changes: 61 additions & 0 deletions bench/remote_parquet_chicago_results.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
{
"date": "2026-09-25",
"method": "Local file-handle read calls and returned bytes from bench/remote_parquet_files.py; these are not HTTP or S3 wire requests. Cold access reads the final row; warm access repeats it in the same process. Default memory cache; no disk cache.",
"files": [
{
"file": "chicago-taxi-flat-f32-cl20.parquet",
"column": "trip.sec",
"file_bytes": 581387271,
"rows": 24314322,
"groups": 25,
"open": {"reads": 2, "bytes": 32679571, "seconds": 1.242},
"cold": {"reads": 1, "bytes": 331590, "seconds": 17.69},
"warm": {"reads": 0, "bytes": 0, "seconds": 0.069},
"arrow_equal": true
},
{
"file": "chicago-taxi-flat-f32.parquet",
"column": "trip.sec",
"file_bytes": 686857703,
"rows": 24314322,
"groups": 25,
"open": {"reads": 2, "bytes": 33057447, "seconds": 1.087},
"cold": {"reads": 1, "bytes": 352939, "seconds": 17.802},
"warm": {"reads": 0, "bytes": 0, "seconds": 0.066},
"arrow_equal": true
},
{
"file": "chicago-taxi-flat-f64.parquet",
"column": "trip.sec",
"file_bytes": 687450036,
"rows": 24314322,
"groups": 25,
"open": {"reads": 2, "bytes": 33068884, "seconds": 1.099},
"cold": {"reads": 1, "bytes": 352939, "seconds": 17.988},
"warm": {"reads": 0, "bytes": 0, "seconds": 0.067},
"arrow_equal": true
},
{
"file": "chicago-taxi-flat-f32-cl20.parquet",
"column": "company",
"file_bytes": 581387271,
"rows": 24314322,
"groups": 25,
"open": {"reads": 2, "bytes": 32679571, "seconds": 1.339},
"cold": {"reads": 1, "bytes": 1831, "seconds": 0.862},
"warm": {"reads": 0, "bytes": 0, "seconds": 0.069},
"arrow_equal": true
},
{
"file": "chicago-taxi.parquet",
"column": "trip.sec",
"file_bytes": 640173859,
"rows": 24314322,
"groups": 25,
"open": {"reads": 27, "bytes": 671548047, "seconds": 3.932},
"cold": {"reads": 1, "bytes": 6660483, "seconds": 1.077},
"warm": {"reads": 0, "bytes": 0, "seconds": 0.067},
"arrow_equal": true
}
]
}
194 changes: 194 additions & 0 deletions bench/remote_parquet_files.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,194 @@
"""Measure lazy reads of local Parquet files directly or through localhost HTTP."""

import argparse
import http.server
import json
import math
import threading
import time
from contextlib import contextmanager, nullcontext
from email.utils import formatdate
from pathlib import Path

import pyarrow.compute as pc
import pyarrow.parquet as pq

import blosc2


@contextmanager
def local_http_server(path):
counts = [0, 0, 0] # GET requests, GET body bytes, HEAD requests.
lock = threading.Lock()
size = path.stat().st_size

class Ranged(http.server.BaseHTTPRequestHandler):
def log_message(self, *args):
pass

def headers_for_file(self, length):
self.send_header("Content-Length", str(length))
self.send_header("Accept-Ranges", "bytes")
self.send_header("Last-Modified", formatdate(path.stat().st_mtime, usegmt=True))

def do_HEAD(self):
with lock:
counts[2] += 1
self.send_response(200)
self.headers_for_file(size)
self.end_headers()

def do_GET(self):
span = self.headers.get("Range")
first, last = 0, size - 1
if span:
start, _, end = span.removeprefix("bytes=").partition("-")
if start:
first = int(start)
last = min(int(end), size - 1) if end else size - 1
else:
first = max(0, size - int(end))
if first > last:
self.send_response(416)
self.send_header("Content-Range", f"bytes */{size}")
self.end_headers()
return
self.send_response(206 if span else 200)
self.headers_for_file(last - first + 1)
if span:
self.send_header("Content-Range", f"bytes {first}-{last}/{size}")
self.end_headers()
sent = 0
with path.open("rb") as source:
source.seek(first)
remaining = last - first + 1
while remaining:
data = source.read(min(1 << 20, remaining))
if not data:
break
self.wfile.write(data)
sent += len(data)
remaining -= len(data)
with lock:
counts[0] += 1
counts[1] += sent

server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Ranged)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
try:

def snapshot():
with lock:
return tuple(counts)

yield f"http://127.0.0.1:{server.server_port}/{path.name}", snapshot
finally:
server.shutdown()
server.server_close()
thread.join()


def measure(path, column=None, *, open_only=False, http=False):
path = Path(path)
context = local_http_server(path) if http else nullcontext((path, None))
with context as (source, server_snapshot):
started = time.perf_counter()
options = {"block_size": 4096, "cache_type": "none"} if http else None
with blosc2.open(source, storage_options=options) as remote:
snapshot = server_snapshot or (lambda: (remote.traffic.requests, remote.traffic.nbytes, 0))
opened = snapshot()
open_seconds = time.perf_counter() - started
column = column or remote.col_names[0]
if open_only:
return {
"file": path.name,
"file_bytes": path.stat().st_size,
"rows": len(remote),
"column": column,
"transport": "http" if http else "file",
"open_reads": opened[0],
"open_bytes": opened[1],
"open_heads": opened[2],
"open_seconds": round(open_seconds, 3),
}
started = time.perf_counter()
cold_value = remote[column][-1]
cold_seconds = time.perf_counter() - started
after_cold = snapshot()
cold = tuple(a - b for a, b in zip(after_cold, opened, strict=True))
started = time.perf_counter()
warm_value = remote[column][-1]
warm_seconds = time.perf_counter() - started
warm = tuple(a - b for a, b in zip(snapshot(), after_cold, strict=True))
result = {
"file": path.name,
"file_bytes": path.stat().st_size,
"rows": len(remote),
"column": column,
"transport": "http" if http else "file",
"open_reads": opened[0],
"open_bytes": opened[1],
"open_heads": opened[2],
"open_seconds": round(open_seconds, 3),
"cold_reads": cold[0],
"cold_bytes": cold[1],
"cold_heads": cold[2],
"cold_seconds": round(cold_seconds, 3),
"warm_reads": warm[0],
"warm_bytes": warm[1],
"warm_heads": warm[2],
"warm_seconds": round(warm_seconds, 3),
"cache_bytes": remote.cache_bytes,
"warm_equal": bool(
cold_value == warm_value
or (isinstance(cold_value, float) and math.isnan(cold_value) and math.isnan(warm_value))
),
}
parquet = pq.ParquetFile(path)
result["groups"] = parquet.num_row_groups
if column in parquet.schema_arrow.names:
last = parquet.read_row_group(parquet.num_row_groups - 1, columns=[column])
expected = last.column(0)[-1].as_py()
elif parquet.schema_arrow.names == [""]:
root = parquet.read_row_group(parquet.num_row_groups - 1, columns=[""]).column(0).combine_chunks()
leaf = pc.list_flatten(root)
for part in column.split("."):
leaf = leaf.field(part)
expected = leaf[-1].as_py()
else:
expected = None
if expected is not None:
result["arrow_equal"] = bool(
cold_value == expected
or (isinstance(cold_value, float) and math.isnan(cold_value) and math.isnan(expected))
)
return result


if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("paths", nargs="+", type=Path)
parser.add_argument("--column")
parser.add_argument("--open-only", action="store_true")
parser.add_argument(
"--http", action="store_true", help="serve each file over localhost HTTP with byte ranges"
)
args = parser.parse_args()
failed = False
for parquet_path in args.paths:
try:
print(
json.dumps(
measure(parquet_path, args.column, open_only=args.open_only, http=args.http), default=str
),
flush=True,
)
except Exception as error:
failed = True
print(
json.dumps({"file": parquet_path.name, "error": f"{type(error).__name__}: {error}"}),
flush=True,
)
if failed:
raise SystemExit(1)
5 changes: 5 additions & 0 deletions bench/remote_parquet_synthetic_optimized.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
scalar: file=109920 B open=1 reads/65536 B/0.004 s cold=1 reads/5364 B/0.108 s warm=0 reads/0 B/0.001 s cache=138536 B python_peak=386880 B
wide: file=1096185 B open=1 reads/65536 B/0.026 s cold=1 reads/5364 B/0.118 s warm=0 reads/0 B/0.002 s cache=138536 B python_peak=275013 B
strings: file=64961 B open=1 reads/64961 B/0.001 s cold=1 reads/6352 B/0.007 s warm=0 reads/0 B/0.002 s cache=67367 B python_peak=132149 B
dictionary: file=8231 B open=1 reads/8231 B/0.001 s cold=1 reads/701 B/0.006 s warm=0 reads/0 B/0.001 s cache=134786 B python_peak=304604 B
unnamed-root: file=55530 B open=11 reads/109198 B/0.003 s cold=1 reads/5368 B/0.003 s warm=0 reads/0 B/0.001 s cache=3283 B python_peak=8441705 B
Loading
Loading