diff --git a/python/README.md b/python/README.md
new file mode 100644
index 0000000..ceb0f31
--- /dev/null
+++ b/python/README.md
@@ -0,0 +1,149 @@
+# Python HA sender
+
+A Python port of the Java/Rust `CsvParallelSender`, plus a pandas/polars dataframe demo.
+It replays a CSV of trades in a loop across N worker threads over QuestDB's QWP
+(WebSocket/UDP) and ILP/HTTP transports, with a probe that reads back the latest ingested
+timestamp and the serving node's live role.
+
+Built on the QuestDB Python client (`questdb.ingress`), which wraps the same C/Rust client
+as the other ports.
+
+## Important: the client must be built from source
+
+The QWP transports (`qwpws`/`qwpwss`/`qwpudp`) **and the query client / egress reader are not
+in the PyPI release** (`questdb` 4.x on PyPI is ILP-only: `Protocol` has just
+`Tcp/Tcps/Http/Https`, and there is no `Client`/`QueryResult`). They live on the
+**`sm_qwp_dataframe_bench`** branch of
+[`py-questdb-client`](https://github.com/questdb/py-questdb-client), which bumps the bundled
+C client to the QWP + egress build. You must build that branch.
+
+Requirements: **Python ≥ 3.10** (the branch is 3.10+; the PyPI-era 3.9 venv will not work),
+a Rust toolchain (`cargo`), and a C compiler. Build steps (a git worktree keeps your
+`main` checkout untouched):
+
+```bash
+cd ~/prj/python/py-questdb-client
+git worktree add --detach /tmp/pyqdb_qwp origin/sm_qwp_dataframe_bench
+cd /tmp/pyqdb_qwp
+git submodule update --init          # bundled c-questdb-client (QWP + egress)
+
+python3.12 -m venv /tmp/pyqdb_venv
+/tmp/pyqdb_venv/bin/pip install -U pip "cython>=3.1.2" "setuptools>=80.9.0" numpy pandas polars pyarrow
+/tmp/pyqdb_venv/bin/pip install -e .   # compiles Cython + the Rust FFI
+```
+
+Verify:
+
+```python
+from questdb.ingress import Protocol, Client   # Client only exists on the QWP build
+print([p for p in dir(Protocol) if not p.startswith('_')])
+# -> ['Http', 'Https', 'QwpUdp', 'QwpWs', 'QwpWss', 'Tcp', 'Tcps']
+```
+
+Then run the scripts with that interpreter (`/tmp/pyqdb_venv/bin/python`). `pandas`, `polars`,
+and `pyarrow` are only needed for the dataframe paths.
+
+## Transports (`--protocol`)
+
+- `qwp` (default): QWP over WebSocket. Per-worker store-and-forward, transactional commit,
+  multi-host failover, and the probe. Use the WebSocket/HTTP port (`:9000`).
+- `qwpudp`: QWP over UDP, fire-and-forget datagrams to `:9007`. Ingest-only,
+  unauthenticated, single-endpoint (no failover), best-effort. **Especially lossy in
+  Python**: the Cython `row()` loop emits datagrams so fast that even a single worker can
+  overrun the server's UDP receive buffer (a full run can vanish). Pace it with `--delay-ms`
+  and keep `--batch-size` small, and treat results as best-effort.
+- `ilp`: the legacy ILP/HTTP transport.
+
+## Run
+
+```bash
+/tmp/pyqdb_venv/bin/python csv_parallel_sender.py \
+  --protocol qwp \
+  --addrs localhost:9000 \
+  --total-events 100000 \
+  --num-senders 4 \
+  --delay-ms 0 \
+  --batch-size 10000 \
+  --batches-per-transaction 10 \
+  --csv ../trades20250728.csv.gz
+```
+
+Flags mirror the Java/Rust ports: `--addrs`, `--token`/`--username`/`--password`,
+`--total-events`, `--delay-ms`, `--num-senders`, `--retry-timeout`, `--csv`,
+`--timestamp-from-file`, `--seconds-offset`, `--protocol`, `--sender-id`,
+`--store-forward-dir`, `--batch-size`, `--batches-per-transaction`, `--probe-interval-ms`,
+`--enterprise`, `--zone`. Confirm results server-side with `SELECT count() FROM trades`.
+
+## Probe (QWP/WebSocket only)
+
+For `qwp`, a background thread uses the pooled query client (`Client`) to run
+`select timestamp from trades limit -1` each interval and prints the latest ingested
+timestamp, then `switch status` for the serving node's live role. `switch status` is
+Enterprise-only (needs SYSTEM ADMIN); on OSS the probe prints
+`(live 'switch status' unavailable, ...)`. The client uses `target=any` (replica-fallback
+reads) and fails over across the `--addrs` hosts.
+
+## Dataframe demo (pandas / polars ingestion + egress)
+
+`dataframe_demo.py` shows the columnar paths the row sender does not:
+
+```bash
+/tmp/pyqdb_venv/bin/python dataframe_demo.py --addr localhost:9000 --csv ../trades20250728.csv.gz --rows 5000
+```
+
+- **pandas ingestion** via `Sender.dataframe(df, ...)` — the numpy-backed pandas planner.
+- **polars ingestion** via `Client.dataframe(df, ...)` — the pooled QWP Arrow-columnar path,
+  which takes polars / pyarrow / any Arrow C Stream source natively. (Numpy-backed pandas is
+  **not** accepted by `Client.dataframe` — it raises `UnsupportedDataFrameShapeError` — so
+  pandas goes through `Sender.dataframe`, or convert with `pyarrow.Table.from_pandas`.)
+- **egress** via `Client.query(sql).to_pandas()` and `.to_polars()` (also `.to_arrow()` /
+  `iter_pandas()`; the result exposes the Arrow PyCapsule interface for zero-copy consumers).
+
+## Differences from the Java/Rust ports
+
+- **The client is pre-release / build-from-source** (see above) — not `pip install questdb`.
+- **Ingestion and querying are split across two classes**: `Sender` (ingest) and `Client`
+  (pooled QWP: query + Arrow-columnar dataframe ingest). The probe uses `Client`.
+- **Auto-flush exists** in the Python client (unlike the Rust client) and is verified to work
+  over QWP (row-threshold flush fires mid-stream; see Validated). This port sets
+  `auto_flush=off` and flushes at the same batch boundaries as the Java/Rust ports for
+  identical cadence.
+- **Config-string scheme is `qwpws`/`qwpwss` for both the sender and the query client**
+  (the Rust reader used `ws`/`wss`; Python only accepts the `qwp*` schemes).
+- **The probe has no handshake-role fallback** — because the Python binding does not expose it.
+  Java/Rust fall back to the QWP handshake role when `switch status` is unavailable. That role
+  exists in the underlying C client (`line_reader_server_info_role`/`_role_byte`) and the Rust
+  reader (`server_info()`), but the Python `Client` on this branch does not wrap it, so this
+  port prints "unavailable" on OSS. Likewise the query path's `on_failover_reset` callback is
+  wired internally in Cython but not surfaced to Python, so there is no user-facing failover
+  narration.
+- Threads (like Java/Rust). Python's GIL is released during client I/O, but row-building is
+  Python-level, so row-by-row throughput is well below Java/Rust — use the dataframe path
+  for volume.
+
+## Validated
+
+Against a local QuestDB (OSS), all rows accounted for server-side:
+
+| What | Result |
+| --- | --- |
+| `qwp`, 3000 rows, 1 worker | 3000 / 3000, distinct per-row timestamps; probe reported live timestamps |
+| `ilp`, 2000 rows, 2 workers | 2000 / 2000 |
+| `qwpudp` (paced) | rows land; best-effort, lossy under fast bursts |
+| `dataframe_demo` | pandas + polars ingest (10000 rows), egress to pandas and polars |
+| auto-flush | `auto_flush_rows=1000`, 2500 rows, no manual flush: 2000 landed mid-stream (2 auto-flushes), close drained to 2500 |
+| QWP failover (primary crash) | see below |
+
+### QWP failover (primary crash)
+
+Two servers from the same build (primary `:9100`, fallback `:9000`), a single-worker `qwp`
+run of 40000 rows with `--addrs :9100,:9000` and the probe active. The primary was
+hard-killed mid-run:
+
+- The sender **completed all 40000 rows with exit 0**, despite its primary crashing.
+- The primary took `trade_id` seq 1–18500; the fallback took 18501–40000 (21500 rows, all
+  distinct). Combined: **contiguous 1–40000, zero loss, zero duplication** — store-and-forward
+  replayed the in-flight transaction to the new host, handing off exactly on a transaction
+  boundary.
+- The probe failed over from the primary to the fallback (visible as a second
+  `connection lost -> restored` cycle once post-failover data committed on the new host).
diff --git a/python/csv_parallel_sender.py b/python/csv_parallel_sender.py
new file mode 100644
index 0000000..43e6ae2
--- /dev/null
+++ b/python/csv_parallel_sender.py
@@ -0,0 +1,372 @@
+#!/usr/bin/env python3
+"""Parallel CSV replay sender for QuestDB, a Python port of the Java/Rust ha_sender.
+
+Replays a CSV of trades in a loop across N worker threads over one of three transports
+(``--protocol``):
+
+  * ``qwp``    - QWP over WebSocket: store-and-forward (un-acked frames spill to disk and
+                 replay after an outage), transactional commit, multi-host failover. A
+                 background probe polls the latest ingested timestamp and the serving
+                 node's live role over a QWP query client (``Client``).
+  * ``qwpudp`` - QWP over UDP: fire-and-forget datagrams to :9007. Ingest-only,
+                 unauthenticated, single-endpoint, best-effort (no ack, no failover).
+  * ``ilp``    - the legacy ILP/HTTP transport.
+
+Requires the QWP/egress build of the questdb Python client (see README.md); the PyPI
+release does not expose QWP or the query client.
+"""
+
+import argparse
+import gzip
+import csv as csvmod
+import os
+import sys
+import threading
+import time
+from datetime import datetime, timezone
+
+from questdb.ingress import Sender, Client, TimestampNanos, ServerTimestamp
+
+PROBE_QUERY = "select timestamp from trades limit -1"
+# Enterprise lifecycle status of whichever node the query client is connected to. Returns
+# the LIVE role (current_role / target_role), unlike the QWP handshake which only refreshes
+# on reconnect and so goes stale after an in-place primary<->replica switch.
+STATUS_QUERY = "switch status"
+
+
+def parse_args(argv):
+    p = argparse.ArgumentParser(description="Parallel CSV replay sender for QuestDB")
+    p.add_argument("--addrs", default="localhost:9000",
+                   help="Comma-separated host:port. QWP/WebSocket + ILP use :9000; UDP uses :9007.")
+    p.add_argument("--token", default=None, help="Bearer token (QWP/WebSocket + ILP only).")
+    p.add_argument("--username", default=None, help="Basic-auth username.")
+    p.add_argument("--password", default=None, help="Basic-auth password.")
+    p.add_argument("--total-events", type=int, default=1_000_000)
+    p.add_argument("--delay-ms", type=int, default=50)
+    p.add_argument("--num-senders", type=int, default=10)
+    p.add_argument("--retry-timeout", type=int, default=360_000,
+                   help="retry_timeout (ILP) / reconnect_max_duration_millis (QWP), in ms.")
+    p.add_argument("--csv", default="./trades20250728.csv.gz")
+    p.add_argument("--timestamp-from-file", action="store_true")
+    p.add_argument("--seconds-offset", type=int, default=0)
+    p.add_argument("--protocol", default="qwp", choices=["qwp", "qwpudp", "ilp"])
+    p.add_argument("--sender-id", default="ha_sender")
+    p.add_argument("--store-forward-dir", default="/tmp/qdb-sf")
+    p.add_argument("--batch-size", type=int, default=10_000)
+    p.add_argument("--batches-per-transaction", type=int, default=10)
+    p.add_argument("--probe-interval-ms", type=int, default=1000)
+    p.add_argument("--enterprise", action="store_true")
+    p.add_argument("--zone", default="eu-west-1")
+    return p.parse_args(argv)
+
+
+def addr_list(args):
+    return [a.strip() for a in args.addrs.split(",") if a.strip()]
+
+
+def has_token(args):
+    return bool(args.token)
+
+
+def has_basic(args):
+    return bool(args.username) and bool(args.password)
+
+
+def tls(args):
+    return has_token(args) or has_basic(args)
+
+
+def _append_auth(parts, args):
+    if has_token(args):
+        parts.append(f"token={args.token};")
+    elif has_basic(args):
+        parts.append(f"username={args.username};password={args.password};")
+
+
+def build_ingest_conf(args, worker_id):
+    """Ingestion connect string for the configured transport (same keys as the Rust port)."""
+    addrs = addr_list(args)
+    if args.protocol == "qwpudp":
+        return f"qwpudp::addr={addrs[0]};auto_flush=off;"
+
+    if args.protocol == "ilp":
+        scheme = "https" if tls(args) else "http"
+        parts = [f"{scheme}::addr={','.join(addrs)};", "auto_flush=off;"]
+        _append_auth(parts, args)
+        if tls(args):
+            parts.append("tls_verify=unsafe_off;")
+        parts.append(f"retry_timeout={args.retry_timeout};")
+        return "".join(parts)
+
+    # qwp (WebSocket)
+    scheme = "qwpwss" if tls(args) else "qwpws"
+    who = f"{args.sender_id}-{worker_id}"
+    sf = os.path.join(args.store_forward_dir, who)
+    os.makedirs(sf, exist_ok=True)
+    parts = [f"{scheme}::addr={','.join(addrs)};", "auto_flush=off;"]
+    _append_auth(parts, args)
+    if tls(args):
+        parts.append("tls_verify=unsafe_off;")
+    parts.append(f"sender_id={who};")
+    parts.append(f"sf_dir={sf};")
+    parts.append(f"reconnect_max_duration_millis={args.retry_timeout};")
+    parts.append("reconnect_initial_backoff_millis=100;")
+    parts.append("reconnect_max_backoff_millis=5000;")
+    if args.enterprise:
+        parts.append("request_durable_ack=true;")
+    return "".join(parts)
+
+
+def build_client_conf(args):
+    """Connect string for the QWP query client (probe): target=any (replica-fallback
+    reads), failover on, zone bias. Same hosts/auth as the senders."""
+    addrs = addr_list(args)
+    scheme = "qwpwss" if tls(args) else "qwpws"
+    parts = [f"{scheme}::addr={','.join(addrs)};", "target=any;", "failover=true;"]
+    _append_auth(parts, args)
+    if tls(args):
+        parts.append("tls_verify=unsafe_off;")
+    if args.zone:
+        parts.append(f"zone={args.zone};")
+    return "".join(parts)
+
+
+def load_csv(path, need_timestamp):
+    """Load the CSV (optionally gzipped). Returns a list of (symbol, side, price, amount,
+    ts_nanos) tuples; ts_nanos is 0 unless need_timestamp."""
+    opener = gzip.open if path.endswith(".gz") else open
+    rows = []
+    with opener(path, "rt", encoding="utf-8", newline="") as fh:
+        reader = csvmod.reader(fh)
+        header = next(reader, None)
+        if header is None:
+            return rows
+        idx = {name.strip(): i for i, name in enumerate(header)}
+        for col in ("symbol", "side", "price", "amount"):
+            if col not in idx:
+                raise ValueError(f"CSV missing required column: {col}")
+        if need_timestamp and "timestamp" not in idx:
+            raise ValueError("CSV missing required column: timestamp")
+        for rec in reader:
+            if not rec:
+                continue
+            ts_nanos = 0
+            if need_timestamp:
+                raw = rec[idx["timestamp"]].strip()
+                # ISO-8601, e.g. 2025-07-28T12:34:56.789012Z
+                dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+                ts_nanos = int(dt.timestamp() * 1_000_000_000)
+            rows.append((
+                rec[idx["symbol"]].strip(),
+                rec[idx["side"]].strip(),
+                float(rec[idx["price"]].strip()),
+                float(rec[idx["amount"]].strip()),
+                ts_nanos,
+            ))
+    return rows
+
+
+def run_worker(worker_id, total_events, args, rows, counts):
+    print(f"Sender {worker_id} will send {total_events} events")
+    is_qwp = args.protocol == "qwp"
+    is_udp = args.protocol == "qwpudp"
+    # Single worker on a QWP transport stamps rows client-side (monotonic, no O3); ILP or
+    # more than one worker use server-side timestamps (ServerTimestamp).
+    per_row_ts = (is_qwp or is_udp) and args.num_senders == 1
+    # Flush cadence: qwp commits every batch-size*batches-per-transaction; qwpudp/ilp flush
+    # every batch-size rows (auto_flush is off, so the buffer is drained explicitly).
+    commit_every = args.batch_size * args.batches_per_transaction if is_qwp else args.batch_size
+
+    conf = build_ingest_conf(args, worker_id)
+    n = len(rows)
+    sent = 0
+    with Sender.from_conf(conf) as sender:
+        for i in range(total_events):
+            symbol, side, price, amount, ts_nanos = rows[i % n]
+            if args.timestamp_from_file:
+                at = TimestampNanos(ts_nanos + args.seconds_offset * 1_000_000_000)
+            elif args.seconds_offset != 0:
+                at = TimestampNanos(time.time_ns() + args.seconds_offset * 1_000_000_000)
+            elif per_row_ts:
+                at = TimestampNanos.now()
+            else:
+                at = ServerTimestamp
+            sender.row(
+                "trades",
+                symbols={"symbol": symbol, "side": side},
+                columns={"price": price, "amount": amount, "trade_id": f"{worker_id}-{i + 1}"},
+                at=at,
+            )
+            sent += 1
+            counts[worker_id] = sent
+            if sent % commit_every == 0:
+                sender.flush()
+            if args.delay_ms > 0:
+                time.sleep(args.delay_ms / 1000.0)
+        sender.flush()
+        # QWP/WebSocket: drain any un-acked store-and-forward frames before closing.
+        if is_qwp:
+            sender.close_drain()
+    print(f"Sender {worker_id} finished sending {sent} events")
+
+
+def _switch_status_roles(client):
+    """Return (current_role, target_role) from `switch status`, or (None, None)."""
+    try:
+        df = client.query(STATUS_QUERY).to_pandas()
+    except Exception:
+        return None, None
+    if len(df) == 0:
+        return None, None
+    current = target = None
+    last = df.iloc[-1]
+    for col in df.columns:
+        lc = col.lower()
+        if "role" not in lc:
+            continue
+        val = last[col]
+        if "current" in lc:
+            current = val
+        elif "target" in lc:
+            target = val
+        elif current is None:
+            current = val
+    return current, target
+
+
+def run_probe(args, stop):
+    conf = build_client_conf(args)
+    try:
+        client = Client.from_conf(conf)
+    except Exception as e:
+        print(f"[query client] connect failed: {e}")
+        return
+    print("[query client] connected")
+    was_down = False
+    try:
+        while not stop.is_set():
+            try:
+                df = client.query(PROBE_QUERY).to_pandas()
+                if was_down:
+                    print("[query client] connection restored")
+                    was_down = False
+                if len(df):
+                    latest = df.iloc[-1, 0]
+                    current, target = _switch_status_roles(client)
+                    if current is not None:
+                        switching = target is not None and str(target).lower() != str(current).lower()
+                        sw = f" (switching -> {target})" if switching else ""
+                        served = f" served by role={current}{sw}"
+                    else:
+                        served = " (live 'switch status' unavailable, e.g. OSS or missing SYSTEM ADMIN)"
+                    print(f"[probe] latest trades timestamp = {latest}{served}")
+            except Exception as e:
+                if not was_down:
+                    print(f"[query client] connection lost ({e}), will retry")
+                    was_down = True
+            stop.wait(args.probe_interval_ms / 1000.0)
+    finally:
+        client.close()
+
+
+def main(argv):
+    args = parse_args(argv)
+
+    for name, val in (("--num-senders", args.num_senders), ("--total-events", args.total_events),
+                      ("--batch-size", args.batch_size),
+                      ("--batches-per-transaction", args.batches_per_transaction)):
+        if val <= 0:
+            print(f"{name} must be > 0", file=sys.stderr)
+            return 2
+
+    addrs = addr_list(args)
+    if not addrs:
+        print("--addrs must contain at least one host:port", file=sys.stderr)
+        return 2
+    if args.protocol == "qwpudp":
+        if len(addrs) > 1:
+            print("--protocol qwpudp supports a single host:port only (UDP is connectionless, "
+                  f"there is no failover); got {len(addrs)} addresses", file=sys.stderr)
+            return 2
+        if args.token or args.username or args.password:
+            print("[warn] --protocol qwpudp is unauthenticated (UDP accepts any connection); "
+                  "ignoring --token/--username/--password", file=sys.stderr)
+
+    if not os.path.exists(args.csv):
+        print(f"CSV file not found: {args.csv}", file=sys.stderr)
+        return 2
+    rows = load_csv(args.csv, args.timestamp_from_file)
+    if not rows:
+        print("CSV has no data rows.", file=sys.stderr)
+        return 2
+
+    if args.protocol == "qwp":
+        print(f"Ingestion started. Protocol: qwp (WebSocket, sender-id={args.sender_id}, "
+              f"store-and-forward={args.store_forward_dir}, batch-size={args.batch_size}, "
+              f"batches-per-transaction={args.batches_per_transaction}, "
+              f"retry-timeout-ms={args.retry_timeout}) | addrs: {','.join(addrs)}")
+    elif args.protocol == "qwpudp":
+        print(f"Ingestion started. Protocol: qwpudp (QWP/UDP datagrams, ingest-only, "
+              f"unauthenticated, no store-and-forward, no failover; query client disabled, "
+              f"batch-size={args.batch_size}) | addr: {addrs[0]}")
+    else:
+        print(f"Ingestion started. Protocol: ilp (HTTP, retry-timeout-ms={args.retry_timeout}) "
+              f"| addrs: {','.join(addrs)}")
+
+    counts = [0] * args.num_senders
+    stop = threading.Event()
+    start = time.monotonic()
+
+    def reporter():
+        last = 0
+        while not stop.is_set():
+            stop.wait(1.0)
+            now = sum(counts)
+            print(f"[progress] sent={now} rate={now - last} rows/s")
+            last = now
+
+    threads = []
+    rep = threading.Thread(target=reporter, daemon=True)
+    rep.start()
+
+    probe_thread = None
+    if args.protocol == "qwp" and args.probe_interval_ms > 0:
+        probe_thread = threading.Thread(target=run_probe, args=(args, stop), daemon=True)
+        probe_thread.start()
+
+    base = args.total_events // args.num_senders
+    rem = args.total_events % args.num_senders
+    errors = []
+
+    def worker_wrapper(wid, events):
+        try:
+            run_worker(wid, events, args, rows, counts)
+        except Exception as e:  # noqa: BLE001
+            errors.append(f"Sender {wid}: {e}")
+
+    for wid in range(args.num_senders):
+        events = base + (1 if wid < rem else 0)
+        t = threading.Thread(target=worker_wrapper, args=(wid, events))
+        t.start()
+        threads.append(t)
+
+    for t in threads:
+        t.join()
+    stop.set()
+    if probe_thread:
+        probe_thread.join(timeout=2.0)
+
+    if errors:
+        for e in errors:
+            print(f"Worker failed: {e}", file=sys.stderr)
+        return 1
+
+    elapsed = time.monotonic() - start
+    rate = args.total_events / elapsed if elapsed > 0 else 0
+    print(f"All workers completed. protocol={args.protocol} events={args.total_events} "
+          f"elapsed={elapsed:.3f} s throughput={rate:,.0f} rows/s")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))
diff --git a/python/dataframe_demo.py b/python/dataframe_demo.py
new file mode 100644
index 0000000..163958d
--- /dev/null
+++ b/python/dataframe_demo.py
@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+"""Pandas and polars ingestion + egress round-trip over QWP/WebSocket.
+
+Showcases the columnar/dataframe capabilities the row-by-row sender does not:
+
+  * pandas ingestion via ``Sender.dataframe`` (the numpy-backed pandas planner).
+  * polars ingestion via ``Client.dataframe`` (the pooled QWP Arrow-columnar path;
+    it takes polars / pyarrow / any Arrow C Stream source natively — numpy-backed
+    pandas is not accepted there, so pandas goes through ``Sender.dataframe``).
+  * egress via ``Client.query(...).to_pandas()`` and ``.to_polars()``.
+
+Usage:
+    python dataframe_demo.py [--addr localhost:9000] [--csv ../trades20250728.csv.gz] [--rows 5000]
+
+Requires the QWP/egress build of the questdb client (see README.md).
+"""
+
+import argparse
+import sys
+import time
+import urllib.parse
+import urllib.request
+
+import pandas as pd
+import polars as pl
+
+from questdb.ingress import Sender, Client
+
+TABLE = "df_demo"
+
+
+def exec_sql(addr, sql):
+    """Run a statement over the HTTP /exec endpoint (same host, HTTP port)."""
+    url = f"http://{addr}/exec?" + urllib.parse.urlencode({"query": sql})
+    with urllib.request.urlopen(url, timeout=10) as resp:
+        resp.read()
+
+
+def wait_for_count(client, table, at_least, timeout_s=30):
+    deadline = time.monotonic() + timeout_s
+    while time.monotonic() < deadline:
+        try:
+            n = int(client.query(f"select count() from {table}").to_pandas().iloc[0, 0])
+            if n >= at_least:
+                return n
+        except Exception:
+            pass
+        time.sleep(0.5)
+    return -1
+
+
+def load_pandas(csv_path, rows):
+    """Load the CSV into a pandas DataFrame with dtypes the ingestion path accepts
+    (object strings, float64, tz-aware nanosecond timestamps)."""
+    raw = pd.read_csv(csv_path, nrows=rows)
+    return pd.DataFrame({
+        "symbol": raw["symbol"].astype("object"),
+        "side": raw["side"].astype("object"),
+        "price": raw["price"].astype("float64"),
+        "amount": raw["amount"].astype("float64"),
+        "timestamp": pd.to_datetime(raw["timestamp"], utc=True).dt.as_unit("ns"),
+    })
+
+
+def main(argv):
+    ap = argparse.ArgumentParser(description="Pandas/polars ingestion + egress demo")
+    ap.add_argument("--addr", default="localhost:9000")
+    ap.add_argument("--csv", default="../trades20250728.csv.gz")
+    ap.add_argument("--rows", type=int, default=5000)
+    args = ap.parse_args(argv)
+
+    pdf = load_pandas(args.csv, args.rows)
+    print(f"loaded pandas DataFrame: {pdf.shape[0]} rows\n{pdf.dtypes}\n")
+    pldf = pl.from_pandas(pdf)
+    print(f"built polars DataFrame: {pldf.height} rows\n")
+
+    exec_sql(args.addr, f"drop table if exists {TABLE}")
+
+    # --- INGESTION ---
+    # pandas -> Sender.dataframe (numpy-backed pandas planner) over QWP/WebSocket.
+    with Sender.from_conf(f"qwpws::addr={args.addr};auto_flush=off;") as sender:
+        sender.dataframe(pdf, table_name=TABLE, symbols=["symbol", "side"], at="timestamp")
+        sender.flush()
+    print(f"[ingest] pandas via Sender.dataframe: {pdf.shape[0]} rows")
+
+    with Client.from_conf(f"qwpws::addr={args.addr};") as client:
+        # polars -> Client.dataframe (Arrow-columnar path, polars is native).
+        client.dataframe(pldf, table_name=TABLE, symbols=["symbol", "side"], at="timestamp")
+        print(f"[ingest] polars via Client.dataframe: {pldf.height} rows")
+
+        expected = pdf.shape[0] + pldf.height
+        n = wait_for_count(client, TABLE, expected)
+        print(f"[ingest] server applied {n} rows (expected {expected})\n")
+
+        # --- EGRESS ---
+        pd_out = client.query(
+            f"select symbol, count() n, round(avg(price), 2) avg_price "
+            f"from {TABLE} order by n desc limit 5"
+        ).to_pandas()
+        print("[egress] Client.query(...).to_pandas():")
+        print(pd_out.to_string(index=False))
+        print()
+
+        pl_out = client.query(
+            f"select side, count() n, round(sum(amount), 4) total_amount "
+            f"from {TABLE} group by side order by side"
+        ).to_polars()
+        print("[egress] Client.query(...).to_polars():")
+        print(pl_out)
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))
diff --git a/rust/.gitignore b/rust/.gitignore
new file mode 100644
index 0000000..ea8c4bf
--- /dev/null
+++ b/rust/.gitignore
@@ -0,0 +1 @@
+/target
diff --git a/rust/Cargo.lock b/rust/Cargo.lock
new file mode 100644
index 0000000..7421cb0
--- /dev/null
+++ b/rust/Cargo.lock
@@ -0,0 +1,1857 @@
+# This file is automatically @generated by Cargo.
+# It is not intended for manual editing.
+version = 4
+
+[[package]]
+name = "adler2"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
+
+[[package]]
+name = "aes"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0"
+dependencies = [
+ "cfg-if",
+ "cipher",
+ "cpufeatures 0.2.17",
+]
+
+[[package]]
+name = "android_system_properties"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311"
+dependencies = [
+ "libc",
+]
+
+[[package]]
+name = "anstream"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d"
+dependencies = [
+ "anstyle",
+ "anstyle-parse",
+ "anstyle-query",
+ "anstyle-wincon",
+ "colorchoice",
+ "is_terminal_polyfill",
+ "utf8parse",
+]
+
+[[package]]
+name = "anstyle"
+version = "1.0.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000"
+
+[[package]]
+name = "anstyle-parse"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e"
+dependencies = [
+ "utf8parse",
+]
+
+[[package]]
+name = "anstyle-query"
+version = "1.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc"
+dependencies = [
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "anstyle-wincon"
+version = "3.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d"
+dependencies = [
+ "anstyle",
+ "once_cell_polyfill",
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "asn1-rs"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f6fd5ddaf0351dff5b8da21b2fb4ff8e08ddd02857f0bf69c47639106c0fff0"
+dependencies = [
+ "asn1-rs-derive 0.4.0",
+ "asn1-rs-impl 0.1.0",
+ "displaydoc",
+ "nom",
+ "num-traits",
+ "rusticata-macros",
+ "thiserror 1.0.69",
+]
+
+[[package]]
+name = "asn1-rs"
+version = "0.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b7f43a50ac4fdca5df8e885c21b835997f0a1cdee65494a6847694a98652d9d8"
+dependencies = [
+ "asn1-rs-derive 0.6.0",
+ "asn1-rs-impl 0.2.0",
+ "displaydoc",
+ "nom",
+ "num-traits",
+ "rusticata-macros",
+ "thiserror 2.0.18",
+ "time",
+]
+
+[[package]]
+name = "asn1-rs-derive"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "726535892e8eae7e70657b4c8ea93d26b8553afb1ce617caee529ef96d7dee6c"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
+ "synstructure 0.12.6",
+]
+
+[[package]]
+name = "asn1-rs-derive"
+version = "0.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3109e49b1e4909e9db6515a30c633684d68cdeaa252f215214cb4fa1a5bfee2c"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+ "synstructure 0.13.2",
+]
+
+[[package]]
+name = "asn1-rs-impl"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2777730b2039ac0f95f093556e61b6d26cebed5393ca6f152717777cec3a42ed"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
+]
+
+[[package]]
+name = "asn1-rs-impl"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
+[[package]]
+name = "autocfg"
+version = "1.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
+
+[[package]]
+name = "base64"
+version = "0.22.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
+
+[[package]]
+name = "base64ct"
+version = "1.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06"
+
+[[package]]
+name = "block-buffer"
+version = "0.10.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71"
+dependencies = [
+ "generic-array",
+]
+
+[[package]]
+name = "block-padding"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a8894febbff9f758034a5b8e12d87918f56dfc64a8e1fe757d65e29041538d93"
+dependencies = [
+ "generic-array",
+]
+
+[[package]]
+name = "bumpalo"
+version = "3.20.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
+
+[[package]]
+name = "bytes"
+version = "1.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
+
+[[package]]
+name = "cbc"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "26b52a9543ae338f279b96b0b9fed9c8093744685043739079ce85cd58f289a6"
+dependencies = [
+ "cipher",
+]
+
+[[package]]
+name = "cc"
+version = "1.2.66"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f5d6cac793997bd970000024b2934968efe83b382de4fdcf4fcb46b6ee4ad996"
+dependencies = [
+ "find-msvc-tools",
+ "jobserver",
+ "libc",
+ "shlex",
+]
+
+[[package]]
+name = "cfg-if"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
+
+[[package]]
+name = "chacha20"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
+dependencies = [
+ "cfg-if",
+ "cpufeatures 0.3.0",
+ "rand_core 0.10.1",
+]
+
+[[package]]
+name = "chrono"
+version = "0.4.45"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327"
+dependencies = [
+ "iana-time-zone",
+ "js-sys",
+ "num-traits",
+ "wasm-bindgen",
+ "windows-link",
+]
+
+[[package]]
+name = "cipher"
+version = "0.4.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad"
+dependencies = [
+ "crypto-common",
+ "inout",
+]
+
+[[package]]
+name = "clap"
+version = "4.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51"
+dependencies = [
+ "clap_builder",
+ "clap_derive",
+]
+
+[[package]]
+name = "clap_builder"
+version = "4.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "714a53001bf66416adb0e2ef5ac857140e7dc3a0c48fb28b2f10762fc4b5069f"
+dependencies = [
+ "anstream",
+ "anstyle",
+ "clap_lex",
+ "strsim",
+]
+
+[[package]]
+name = "clap_derive"
+version = "4.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9"
+dependencies = [
+ "heck",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
+[[package]]
+name = "clap_lex"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9"
+
+[[package]]
+name = "cms"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7b77c319abfd5219629c45c34c89ba945ed3c5e49fcde9d16b6c3885f118a730"
+dependencies = [
+ "const-oid",
+ "der",
+ "spki",
+ "x509-cert",
+]
+
+[[package]]
+name = "colorchoice"
+version = "1.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
+
+[[package]]
+name = "const-oid"
+version = "0.9.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
+
+[[package]]
+name = "core-foundation-sys"
+version = "0.8.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
+
+[[package]]
+name = "cpufeatures"
+version = "0.2.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280"
+dependencies = [
+ "libc",
+]
+
+[[package]]
+name = "cpufeatures"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201"
+dependencies = [
+ "libc",
+]
+
+[[package]]
+name = "crc32c"
+version = "0.6.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3a47af21622d091a8f0fb295b88bc886ac74efcc613efc19f5d0b21de5c89e47"
+dependencies = [
+ "rustc_version",
+]
+
+[[package]]
+name = "crc32fast"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511"
+dependencies = [
+ "cfg-if",
+]
+
+[[package]]
+name = "crypto-common"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a"
+dependencies = [
+ "generic-array",
+ "typenum",
+]
+
+[[package]]
+name = "csv"
+version = "1.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "52cd9d68cf7efc6ddfaaee42e7288d3a99d613d4b50f76ce9827ae0c6e14f938"
+dependencies = [
+ "csv-core",
+ "itoa",
+ "ryu",
+ "serde_core",
+]
+
+[[package]]
+name = "csv-core"
+version = "0.1.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "704a3c26996a80471189265814dbc2c257598b96b8a7feae2d31ace646bb9782"
+dependencies = [
+ "memchr",
+]
+
+[[package]]
+name = "data-encoding"
+version = "2.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8"
+
+[[package]]
+name = "der"
+version = "0.7.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb"
+dependencies = [
+ "const-oid",
+ "der_derive",
+ "flagset",
+ "pem-rfc7468",
+ "zeroize",
+]
+
+[[package]]
+name = "der-parser"
+version = "10.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6"
+dependencies = [
+ "asn1-rs 0.7.2",
+ "displaydoc",
+ "nom",
+ "num-bigint",
+ "num-traits",
+ "rusticata-macros",
+]
+
+[[package]]
+name = "der_derive"
+version = "0.7.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8034092389675178f570469e6c3b0465d3d30b4505c294a6550db47f3c17ad18"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
+[[package]]
+name = "deranged"
+version = "0.5.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c"
+
+[[package]]
+name = "des"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ffdd80ce8ce993de27e9f063a444a4d53ce8e8db4c1f00cc03af5ad5a9867a1e"
+dependencies = [
+ "cipher",
+]
+
+[[package]]
+name = "digest"
+version = "0.10.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
+dependencies = [
+ "block-buffer",
+ "crypto-common",
+ "subtle",
+]
+
+[[package]]
+name = "displaydoc"
+version = "0.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
+[[package]]
+name = "dns-lookup"
+version = "3.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6e39034cee21a2f5bbb66ba0e3689819c4bb5d00382a282006e802a7ffa6c41d"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "socket2",
+ "windows-sys 0.60.2",
+]
+
+[[package]]
+name = "find-msvc-tools"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
+
+[[package]]
+name = "flagset"
+version = "0.4.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b7ac824320a75a52197e8f2d787f6a38b6718bb6897a35142d749af3c0e8f4fe"
+
+[[package]]
+name = "flate2"
+version = "1.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c"
+dependencies = [
+ "crc32fast",
+ "miniz_oxide",
+]
+
+[[package]]
+name = "futures-core"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
+
+[[package]]
+name = "futures-task"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393"
+
+[[package]]
+name = "futures-util"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "pin-project-lite",
+ "slab",
+]
+
+[[package]]
+name = "generic-array"
+version = "0.14.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
+dependencies = [
+ "typenum",
+ "version_check",
+]
+
+[[package]]
+name = "getrandom"
+version = "0.2.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "wasi",
+]
+
+[[package]]
+name = "getrandom"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "r-efi 5.3.0",
+ "wasip2",
+]
+
+[[package]]
+name = "getrandom"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "r-efi 6.0.0",
+ "rand_core 0.10.1",
+]
+
+[[package]]
+name = "heck"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
+
+[[package]]
+name = "hex"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
+
+[[package]]
+name = "hmac"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e"
+dependencies = [
+ "digest",
+]
+
+[[package]]
+name = "http"
+version = "1.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425"
+dependencies = [
+ "bytes",
+ "itoa",
+]
+
+[[package]]
+name = "httparse"
+version = "1.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
+
+[[package]]
+name = "iana-time-zone"
+version = "0.1.65"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470"
+dependencies = [
+ "android_system_properties",
+ "core-foundation-sys",
+ "iana-time-zone-haiku",
+ "js-sys",
+ "log",
+ "wasm-bindgen",
+ "windows-core",
+]
+
+[[package]]
+name = "iana-time-zone-haiku"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f"
+dependencies = [
+ "cc",
+]
+
+[[package]]
+name = "indoc"
+version = "2.0.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "79cf5c93f93228cf8efb3ba362535fb11199ac548a09ce117c9b1adc3030d706"
+dependencies = [
+ "rustversion",
+]
+
+[[package]]
+name = "inout"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01"
+dependencies = [
+ "block-padding",
+ "generic-array",
+]
+
+[[package]]
+name = "is_terminal_polyfill"
+version = "1.70.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695"
+
+[[package]]
+name = "itoa"
+version = "1.0.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
+
+[[package]]
+name = "jks"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e03966fd15eea3cb2886320a78d01e77f8aaeabd3fb01504ee6a2238876c23bc"
+dependencies = [
+ "asn1-rs 0.5.2",
+ "sha1",
+ "thiserror 1.0.69",
+]
+
+[[package]]
+name = "jobserver"
+version = "0.1.35"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3"
+dependencies = [
+ "getrandom 0.4.3",
+ "libc",
+]
+
+[[package]]
+name = "js-sys"
+version = "0.3.103"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102"
+dependencies = [
+ "cfg-if",
+ "futures-util",
+ "wasm-bindgen",
+]
+
+[[package]]
+name = "lazy_static"
+version = "1.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
+
+[[package]]
+name = "libc"
+version = "0.2.186"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66"
+
+[[package]]
+name = "log"
+version = "0.4.33"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad"
+
+[[package]]
+name = "memchr"
+version = "2.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
+
+[[package]]
+name = "memmap2"
+version = "0.9.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d1219ed1b7f229ee7104d281dd01d6802fe28bb6e95d292942c4daacdeb798c0"
+dependencies = [
+ "libc",
+]
+
+[[package]]
+name = "minimal-lexical"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
+
+[[package]]
+name = "miniz_oxide"
+version = "0.8.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316"
+dependencies = [
+ "adler2",
+ "simd-adler32",
+]
+
+[[package]]
+name = "nom"
+version = "7.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a"
+dependencies = [
+ "memchr",
+ "minimal-lexical",
+]
+
+[[package]]
+name = "num-bigint"
+version = "0.4.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c89e69e7e0f03bea5ef08013795c25018e101932225a656383bd384495ecc367"
+dependencies = [
+ "num-integer",
+ "num-traits",
+]
+
+[[package]]
+name = "num-conv"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441"
+
+[[package]]
+name = "num-integer"
+version = "0.1.46"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f"
+dependencies = [
+ "num-traits",
+]
+
+[[package]]
+name = "num-traits"
+version = "0.2.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841"
+dependencies = [
+ "autocfg",
+]
+
+[[package]]
+name = "oid-registry"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7"
+dependencies = [
+ "asn1-rs 0.7.2",
+]
+
+[[package]]
+name = "once_cell"
+version = "1.21.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
+
+[[package]]
+name = "once_cell_polyfill"
+version = "1.70.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe"
+
+[[package]]
+name = "p12-keystore"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ffb9bf5222606eb712d3bb30e01bc9420545b00859970897e70c682353a034f2"
+dependencies = [
+ "base64",
+ "cbc",
+ "cms",
+ "der",
+ "des",
+ "hex",
+ "hmac",
+ "pkcs12",
+ "pkcs5",
+ "rand 0.10.2",
+ "rc2",
+ "sha1",
+ "sha2",
+ "thiserror 2.0.18",
+ "x509-parser",
+]
+
+[[package]]
+name = "pbkdf2"
+version = "0.12.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f8ed6a7761f76e3b9f92dfb0a60a6a6477c61024b775147ff0973a02653abaf2"
+dependencies = [
+ "digest",
+ "hmac",
+]
+
+[[package]]
+name = "pem-rfc7468"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "88b39c9bfcfc231068454382784bb460aae594343fb030d46e9f50a645418412"
+dependencies = [
+ "base64ct",
+]
+
+[[package]]
+name = "percent-encoding"
+version = "2.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
+
+[[package]]
+name = "pin-project-lite"
+version = "0.2.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
+
+[[package]]
+name = "pkcs12"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "695b3df3d3cc1015f12d70235e35b6b79befc5fa7a9b95b951eab1dd07c9efc2"
+dependencies = [
+ "cms",
+ "const-oid",
+ "der",
+ "digest",
+ "spki",
+ "x509-cert",
+ "zeroize",
+]
+
+[[package]]
+name = "pkcs5"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e847e2c91a18bfa887dd028ec33f2fe6f25db77db3619024764914affe8b69a6"
+dependencies = [
+ "aes",
+ "cbc",
+ "der",
+ "pbkdf2",
+ "scrypt",
+ "sha2",
+ "spki",
+]
+
+[[package]]
+name = "pkg-config"
+version = "0.3.33"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e"
+
+[[package]]
+name = "powerfmt"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391"
+
+[[package]]
+name = "ppv-lite86"
+version = "0.2.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9"
+dependencies = [
+ "zerocopy",
+]
+
+[[package]]
+name = "proc-macro2"
+version = "1.0.106"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934"
+dependencies = [
+ "unicode-ident",
+]
+
+[[package]]
+name = "questdb-confstr"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7aceffde1cbf8e67f34cdfd70d2436396176d6ff648fa719e0231fb9856ef3e9"
+
+[[package]]
+name = "questdb-rs"
+version = "7.0.0"
+dependencies = [
+ "base64ct",
+ "bytes",
+ "crc32c",
+ "dns-lookup",
+ "indoc",
+ "itoa",
+ "jks",
+ "libc",
+ "log",
+ "memmap2",
+ "p12-keystore",
+ "questdb-confstr",
+ "rand 0.9.4",
+ "ring",
+ "rustls",
+ "rustls-pki-types",
+ "ryu",
+ "serde",
+ "serde_json",
+ "slugify",
+ "socket2",
+ "ureq",
+ "webpki-roots",
+ "windows-sys 0.60.2",
+ "zstd",
+]
+
+[[package]]
+name = "questdb_ha_sender"
+version = "1.0.0"
+dependencies = [
+ "chrono",
+ "clap",
+ "csv",
+ "flate2",
+ "questdb-rs",
+]
+
+[[package]]
+name = "quote"
+version = "1.0.46"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368"
+dependencies = [
+ "proc-macro2",
+]
+
+[[package]]
+name = "r-efi"
+version = "5.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
+
+[[package]]
+name = "r-efi"
+version = "6.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
+
+[[package]]
+name = "rand"
+version = "0.9.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea"
+dependencies = [
+ "rand_chacha",
+ "rand_core 0.9.5",
+]
+
+[[package]]
+name = "rand"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
+dependencies = [
+ "chacha20",
+ "getrandom 0.4.3",
+ "rand_core 0.10.1",
+]
+
+[[package]]
+name = "rand_chacha"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb"
+dependencies = [
+ "ppv-lite86",
+ "rand_core 0.9.5",
+]
+
+[[package]]
+name = "rand_core"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c"
+dependencies = [
+ "getrandom 0.3.4",
+]
+
+[[package]]
+name = "rand_core"
+version = "0.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69"
+
+[[package]]
+name = "rc2"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "62c64daa8e9438b84aaae55010a93f396f8e60e3911590fcba770d04643fc1dd"
+dependencies = [
+ "cipher",
+]
+
+[[package]]
+name = "ring"
+version = "0.17.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7"
+dependencies = [
+ "cc",
+ "cfg-if",
+ "getrandom 0.2.17",
+ "libc",
+ "untrusted",
+ "windows-sys 0.52.0",
+]
+
+[[package]]
+name = "rustc_version"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92"
+dependencies = [
+ "semver",
+]
+
+[[package]]
+name = "rusticata-macros"
+version = "4.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632"
+dependencies = [
+ "nom",
+]
+
+[[package]]
+name = "rustls"
+version = "0.23.41"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6b92b125634d9b795e7beca796cc790df15a7fb38323bf3196fda83292d06b1f"
+dependencies = [
+ "log",
+ "once_cell",
+ "ring",
+ "rustls-pki-types",
+ "rustls-webpki",
+ "subtle",
+ "zeroize",
+]
+
+[[package]]
+name = "rustls-pki-types"
+version = "1.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046"
+dependencies = [
+ "zeroize",
+]
+
+[[package]]
+name = "rustls-webpki"
+version = "0.103.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e"
+dependencies = [
+ "ring",
+ "rustls-pki-types",
+ "untrusted",
+]
+
+[[package]]
+name = "rustversion"
+version = "1.0.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f"
+
+[[package]]
+name = "ryu"
+version = "1.0.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
+
+[[package]]
+name = "salsa20"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "97a22f5af31f73a954c10289c93e8a50cc23d971e80ee446f1f6f7137a088213"
+dependencies = [
+ "cipher",
+]
+
+[[package]]
+name = "scrypt"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0516a385866c09368f0b5bcd1caff3366aace790fcd46e2bb032697bb172fd1f"
+dependencies = [
+ "pbkdf2",
+ "salsa20",
+ "sha2",
+]
+
+[[package]]
+name = "semver"
+version = "1.0.28"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8a7852d02fc848982e0c167ef163aaff9cd91dc640ba85e263cb1ce46fae51cd"
+
+[[package]]
+name = "serde"
+version = "1.0.228"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e"
+dependencies = [
+ "serde_core",
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_core"
+version = "1.0.228"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad"
+dependencies = [
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_derive"
+version = "1.0.228"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
+[[package]]
+name = "serde_json"
+version = "1.0.150"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9"
+dependencies = [
+ "itoa",
+ "memchr",
+ "serde",
+ "serde_core",
+ "zmij",
+]
+
+[[package]]
+name = "sha1"
+version = "0.10.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba"
+dependencies = [
+ "cfg-if",
+ "cpufeatures 0.2.17",
+ "digest",
+]
+
+[[package]]
+name = "sha2"
+version = "0.10.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283"
+dependencies = [
+ "cfg-if",
+ "cpufeatures 0.2.17",
+ "digest",
+]
+
+[[package]]
+name = "shlex"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba"
+
+[[package]]
+name = "simd-adler32"
+version = "0.3.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214"
+
+[[package]]
+name = "slab"
+version = "0.4.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5"
+
+[[package]]
+name = "slugify"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d6b8cf203d2088b831d7558f8e5151bfa420c57a34240b28cee29d0ae5f2ac8b"
+dependencies = [
+ "unidecode",
+]
+
+[[package]]
+name = "socket2"
+version = "0.6.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51"
+dependencies = [
+ "libc",
+ "windows-sys 0.61.2",
+]
+
+[[package]]
+name = "spki"
+version = "0.7.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d"
+dependencies = [
+ "base64ct",
+ "der",
+]
+
+[[package]]
+name = "strsim"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f"
+
+[[package]]
+name = "subtle"
+version = "2.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
+
+[[package]]
+name = "syn"
+version = "1.0.109"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-ident",
+]
+
+[[package]]
+name = "syn"
+version = "2.0.118"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1b9ae57f904213ebb649ce6895b8a66c66f0203b9319718f69a5612a065b1422"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-ident",
+]
+
+[[package]]
+name = "synstructure"
+version = "0.12.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f36bdaa60a83aca3921b5259d5400cbf5e90fc51931376a9bd4a0eb79aa7210f"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
+ "unicode-xid",
+]
+
+[[package]]
+name = "synstructure"
+version = "0.13.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
+[[package]]
+name = "thiserror"
+version = "1.0.69"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
+dependencies = [
+ "thiserror-impl 1.0.69",
+]
+
+[[package]]
+name = "thiserror"
+version = "2.0.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4"
+dependencies = [
+ "thiserror-impl 2.0.18",
+]
+
+[[package]]
+name = "thiserror-impl"
+version = "1.0.69"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
+[[package]]
+name = "thiserror-impl"
+version = "2.0.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
+[[package]]
+name = "time"
+version = "0.3.53"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "18dfaaeddcb932337b5e7866ee7d0ce9b76d2fd092997146f187ec09b4558a50"
+dependencies = [
+ "deranged",
+ "num-conv",
+ "powerfmt",
+ "serde_core",
+ "time-core",
+ "time-macros",
+]
+
+[[package]]
+name = "time-core"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109"
+
+[[package]]
+name = "time-macros"
+version = "0.2.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c431b87111666e491a90baa837f914fb45cd5dc3c268591b0220ff5057f2085f"
+dependencies = [
+ "num-conv",
+ "time-core",
+]
+
+[[package]]
+name = "typenum"
+version = "1.20.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"
+
+[[package]]
+name = "unicode-ident"
+version = "1.0.24"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
+
+[[package]]
+name = "unicode-xid"
+version = "0.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853"
+
+[[package]]
+name = "unidecode"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "402bb19d8e03f1d1a7450e2bd613980869438e0666331be3e073089124aa1adc"
+
+[[package]]
+name = "untrusted"
+version = "0.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1"
+
+[[package]]
+name = "ureq"
+version = "3.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d39cb1dbab692d82a977c0392ffac19e188bd9186a9f32806f0aaa859d75585a"
+dependencies = [
+ "base64",
+ "log",
+ "percent-encoding",
+ "ureq-proto",
+ "utf-8",
+]
+
+[[package]]
+name = "ureq-proto"
+version = "0.5.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d81f9efa9df032be5934a46a068815a10a042b494b6a58cb0a1a97bb5467ed6f"
+dependencies = [
+ "base64",
+ "http",
+ "httparse",
+ "log",
+]
+
+[[package]]
+name = "utf-8"
+version = "0.7.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
+
+[[package]]
+name = "utf8parse"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
+
+[[package]]
+name = "version_check"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
+
+[[package]]
+name = "wasi"
+version = "0.11.1+wasi-snapshot-preview1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
+
+[[package]]
+name = "wasip2"
+version = "1.0.4+wasi-0.2.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487"
+dependencies = [
+ "wit-bindgen",
+]
+
+[[package]]
+name = "wasm-bindgen"
+version = "0.2.126"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4"
+dependencies = [
+ "cfg-if",
+ "once_cell",
+ "rustversion",
+ "wasm-bindgen-macro",
+ "wasm-bindgen-shared",
+]
+
+[[package]]
+name = "wasm-bindgen-macro"
+version = "0.2.126"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1"
+dependencies = [
+ "quote",
+ "wasm-bindgen-macro-support",
+]
+
+[[package]]
+name = "wasm-bindgen-macro-support"
+version = "0.2.126"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e"
+dependencies = [
+ "bumpalo",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+ "wasm-bindgen-shared",
+]
+
+[[package]]
+name = "wasm-bindgen-shared"
+version = "0.2.126"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24"
+dependencies = [
+ "unicode-ident",
+]
+
+[[package]]
+name = "webpki-roots"
+version = "1.0.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bf85cb06032201fa7c6f829d7db5a7e5aa45bcc0655327713065f6f0576731bf"
+dependencies = [
+ "rustls-pki-types",
+]
+
+[[package]]
+name = "windows-core"
+version = "0.62.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb"
+dependencies = [
+ "windows-implement",
+ "windows-interface",
+ "windows-link",
+ "windows-result",
+ "windows-strings",
+]
+
+[[package]]
+name = "windows-implement"
+version = "0.60.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
+[[package]]
+name = "windows-interface"
+version = "0.59.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
+[[package]]
+name = "windows-link"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
+
+[[package]]
+name = "windows-result"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5"
+dependencies = [
+ "windows-link",
+]
+
+[[package]]
+name = "windows-strings"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091"
+dependencies = [
+ "windows-link",
+]
+
+[[package]]
+name = "windows-sys"
+version = "0.52.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
+dependencies = [
+ "windows-targets 0.52.6",
+]
+
+[[package]]
+name = "windows-sys"
+version = "0.60.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb"
+dependencies = [
+ "windows-targets 0.53.5",
+]
+
+[[package]]
+name = "windows-sys"
+version = "0.61.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
+dependencies = [
+ "windows-link",
+]
+
+[[package]]
+name = "windows-targets"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973"
+dependencies = [
+ "windows_aarch64_gnullvm 0.52.6",
+ "windows_aarch64_msvc 0.52.6",
+ "windows_i686_gnu 0.52.6",
+ "windows_i686_gnullvm 0.52.6",
+ "windows_i686_msvc 0.52.6",
+ "windows_x86_64_gnu 0.52.6",
+ "windows_x86_64_gnullvm 0.52.6",
+ "windows_x86_64_msvc 0.52.6",
+]
+
+[[package]]
+name = "windows-targets"
+version = "0.53.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3"
+dependencies = [
+ "windows-link",
+ "windows_aarch64_gnullvm 0.53.1",
+ "windows_aarch64_msvc 0.53.1",
+ "windows_i686_gnu 0.53.1",
+ "windows_i686_gnullvm 0.53.1",
+ "windows_i686_msvc 0.53.1",
+ "windows_x86_64_gnu 0.53.1",
+ "windows_x86_64_gnullvm 0.53.1",
+ "windows_x86_64_msvc 0.53.1",
+]
+
+[[package]]
+name = "windows_aarch64_gnullvm"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3"
+
+[[package]]
+name = "windows_aarch64_gnullvm"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53"
+
+[[package]]
+name = "windows_aarch64_msvc"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469"
+
+[[package]]
+name = "windows_aarch64_msvc"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006"
+
+[[package]]
+name = "windows_i686_gnu"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b"
+
+[[package]]
+name = "windows_i686_gnu"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3"
+
+[[package]]
+name = "windows_i686_gnullvm"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66"
+
+[[package]]
+name = "windows_i686_gnullvm"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c"
+
+[[package]]
+name = "windows_i686_msvc"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66"
+
+[[package]]
+name = "windows_i686_msvc"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2"
+
+[[package]]
+name = "windows_x86_64_gnu"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78"
+
+[[package]]
+name = "windows_x86_64_gnu"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499"
+
+[[package]]
+name = "windows_x86_64_gnullvm"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d"
+
+[[package]]
+name = "windows_x86_64_gnullvm"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1"
+
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.52.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
+
+[[package]]
+name = "windows_x86_64_msvc"
+version = "0.53.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"
+
+[[package]]
+name = "wit-bindgen"
+version = "0.57.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"
+
+[[package]]
+name = "x509-cert"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1301e935010a701ae5f8655edc0ad17c44bad3ac5ce8c39185f75453b720ae94"
+dependencies = [
+ "const-oid",
+ "der",
+ "spki",
+]
+
+[[package]]
+name = "x509-parser"
+version = "0.18.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d43b0f71ce057da06bc0851b23ee24f3f86190b07203dd8f567d0b706a185202"
+dependencies = [
+ "asn1-rs 0.7.2",
+ "data-encoding",
+ "der-parser",
+ "lazy_static",
+ "nom",
+ "oid-registry",
+ "rusticata-macros",
+ "thiserror 2.0.18",
+ "time",
+]
+
+[[package]]
+name = "zerocopy"
+version = "0.8.54"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b7cbbc0a705a0fd05cc3676525980d2bf5a9bc4adac6d6475209a7887cf59d19"
+dependencies = [
+ "zerocopy-derive",
+]
+
+[[package]]
+name = "zerocopy-derive"
+version = "0.8.54"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e2e817b7b52d0c7358d3246da9d69935ebb18116b2b102b4230dac079b4862f5"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.118",
+]
+
+[[package]]
+name = "zeroize"
+version = "1.9.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e"
+
+[[package]]
+name = "zmij"
+version = "1.0.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa"
+
+[[package]]
+name = "zstd"
+version = "0.13.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a"
+dependencies = [
+ "zstd-safe",
+]
+
+[[package]]
+name = "zstd-safe"
+version = "7.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d"
+dependencies = [
+ "zstd-sys",
+]
+
+[[package]]
+name = "zstd-sys"
+version = "2.0.16+zstd.1.5.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748"
+dependencies = [
+ "cc",
+ "pkg-config",
+]
diff --git a/rust/Cargo.toml b/rust/Cargo.toml
new file mode 100644
index 0000000..27aa59f
--- /dev/null
+++ b/rust/Cargo.toml
@@ -0,0 +1,27 @@
+[package]
+name = "questdb_ha_sender"
+version = "1.0.0"
+edition = "2021"
+description = "Parallel CSV replay sender for QuestDB over QWP (WebSocket/UDP) and ILP/HTTP, a Rust port of the Java ha_sender"
+
+[[bin]]
+name = "csv_parallel_sender"
+path = "src/main.rs"
+
+[dependencies]
+# The Rust/C client, built locally from the c-questdb-client checkout. Default features
+# already enable every sync sender (TCP, HTTP, QWP/UDP, QWP/WebSocket) plus webpki roots
+# and the ring crypto backend; we add the egress reader (for the probe / query client),
+# zstd result decompression, and insecure-skip-verify (for tls_verify=unsafe_off).
+questdb-rs = { path = "../../../questdb/c-questdb-client/questdb-rs", features = [
+    "sync-reader-ws",
+    "compression-zstd",
+    "insecure-skip-verify",
+] }
+clap = { version = "4", features = ["derive"] }
+csv = "1"
+flate2 = "1"
+chrono = "0.4"
+
+[profile.release]
+opt-level = 3
diff --git a/rust/README.md b/rust/README.md
new file mode 100644
index 0000000..bb8a2e2
--- /dev/null
+++ b/rust/README.md
@@ -0,0 +1,150 @@
+# Rust HA sender
+
+A Rust port of the Java `CsvParallelSender`. It replays a CSV of trades in a loop across
+N worker threads over one of three QuestDB transports, and (on QWP/WebSocket) runs a probe
+that reports the latest ingested timestamp and the serving node's live role.
+
+It is built on the Rust/C client (`questdb-rs`) from
+[`c-questdb-client`](https://github.com/questdb/c-questdb-client), which powers the C, C++,
+and Python clients too.
+
+## Transports (`--protocol`)
+
+- `qwp` (default): **QWP over WebSocket**. Store-and-forward (un-acked frames spill to disk
+  and replay after an outage), transactional commit, and multi-host failover. A background
+  probe polls the latest ingested timestamp and the serving node's live role over a QWP
+  query client (the `Reader`). Use the server's WebSocket/HTTP port (`:9000`).
+- `qwpudp`: **QWP over UDP**, fire-and-forget datagrams to `:9007`. Ingest-only,
+  unauthenticated, single-endpoint (no failover), best-effort (no ack, no store-and-forward).
+  Keep `--batch-size` small: UDP has no backpressure, so a large flush burst overflows the
+  server's receive buffer and drops datagrams wholesale.
+- `ilp`: the legacy **ILP over HTTP** transport.
+
+## Build
+
+The client is a **path dependency** on a local `c-questdb-client` checkout, expected at
+`../../../questdb/c-questdb-client/questdb-rs` (i.e. `~/prj/questdb/c-questdb-client`
+alongside `~/prj/java/questdb_java_ha_sender`). Adjust the path in `Cargo.toml` if your
+checkout is elsewhere. The QWP wire protocol must match the target server build.
+
+```
+cargo build --release
+```
+
+Requires a Rust toolchain new enough for the client's 2024 edition (1.85+; tested on 1.93).
+
+## Run
+
+```
+./target/release/csv_parallel_sender \
+  --protocol qwp \
+  --addrs localhost:9000 \
+  --total-events 5000000 \
+  --num-senders 4 \
+  --delay-ms 0 \
+  --batch-size 10000 \
+  --batches-per-transaction 10 \
+  --csv ../trades20250728.csv.gz
+```
+
+UDP (single host, small batch, no auth, no probe):
+
+```
+./target/release/csv_parallel_sender \
+  --protocol qwpudp --addrs localhost:9007 \
+  --total-events 100000 --num-senders 1 --batch-size 500 \
+  --csv ../trades20250728.csv.gz
+```
+
+QWP batch errors and UDP loss are asynchronous or silent, so always confirm the result with
+a server-side `SELECT count() FROM trades` after a run.
+
+## Flags
+
+| Flag | Default | Notes |
+| --- | --- | --- |
+| `--addrs` | `localhost:9000` | Comma-separated `host:port`. QWP/WebSocket + ILP use `:9000`; UDP uses `:9007` and accepts one host only. |
+| `--token` | | Bearer token (QWP/WebSocket + ILP). UDP rejects auth. |
+| `--username` / `--password` | | Basic auth (QWP/WebSocket + ILP). |
+| `--total-events` | `1000000` | Rows across all workers. |
+| `--delay-ms` | `50` | Per-row sleep. |
+| `--num-senders` | `10` | Worker threads (one `Sender` each). |
+| `--retry-timeout` | `360000` | `retry_timeout` (ILP) / `reconnect_max_duration_millis` (QWP). |
+| `--csv` | `./trades20250728.csv.gz` | `.csv` or `.csv.gz`; needs `symbol,side,price,amount[,timestamp]`. |
+| `--timestamp-from-file` | `false` | Use the CSV timestamp column instead of "now". |
+| `--seconds-offset` | `0` | Shift each timestamp by N seconds. |
+| `--protocol` | `qwp` | `qwp` \| `qwpudp` \| `ilp`. |
+| `--sender-id` | `ha_sender` | Store-and-forward key base; each worker gets `<sender-id>-<worker>`. |
+| `--store-forward-dir` | `/tmp/qdb-sf` | Spill directory (QWP/WebSocket). |
+| `--batch-size` | `10000` | Rows per flush. QWP commits every `batch-size × batches-per-transaction`; UDP/ILP flush every `batch-size`. |
+| `--batches-per-transaction` | `10` | QWP transaction size, in batches. |
+| `--probe-interval-ms` | `1000` | Probe poll interval (QWP/WebSocket only; `0` disables). |
+| `--enterprise` | `false` | Request durable acks (Enterprise only). |
+| `--zone` | `eu-west-1` | Preferred zone for the query client / probe. |
+
+## Timestamps
+
+- **Single worker on a QWP transport** (`qwp` or `qwpudp`, `--num-senders 1`): each row is
+  stamped with the current microsecond client-side (monotonic, so no out-of-order), giving
+  distinct per-row timestamps.
+- **ILP, or more than one worker**: rows use server-side `at_now()` (O3-safe across workers;
+  a whole batch shares one timestamp).
+- `--timestamp-from-file` / `--seconds-offset` stamp each row explicitly from the CSV.
+
+## Probe (QWP/WebSocket only)
+
+A background thread runs `select timestamp from trades limit -1` each interval and prints
+the latest ingested timestamp, then asks the serving node for its live role with
+`switch status`:
+
+```
+[query client] connected, serving node=n1 role=PRIMARY zone=eu-west-1 cluster=...
+[probe] latest trades timestamp = 2026-07-01T14:31:40.591545Z (raw=1782916300591545) served by role=PRIMARY node=n1 zone=eu-west-1
+```
+
+`switch status` gives the authoritative live role (`current_role`, plus `target_role` while
+a switch is in flight, shown as `(switching -> ROLE)`). It is Enterprise-only and needs
+SYSTEM ADMIN; on OSS the probe falls back to the QWP handshake role, labelled
+`(handshake role; live 'switch status' unavailable, may be stale)`. The query client uses
+`target=any` (so reads fall back to a replica when no primary is available) and fails over
+across the `--addrs` hosts automatically.
+
+## Differences from the Java client
+
+- **No auto-flush.** The Rust/C client deliberately has no auto-flush (it rejects every
+  `auto_flush*` key except `off`); batching is driven purely by explicit `flush()` calls.
+  This port reproduces the Java cadence by flushing at the same row boundaries, so behavior
+  matches; there is simply no background flush timer or byte cap.
+- **No `--connect-timeout-ms`.** The Rust QWP transport exposes no single-connect timeout
+  knob (only the overall `reconnect_max_duration`), so that flag is omitted here.
+- **Single bearer token only.** QWP (WebSocket/HTTP) auth is a single `token` (or
+  username/password). The split `x`/`y` key components only exist for legacy TCP-ILP ECDSA
+  auth, which this tool does not use.
+
+## Validated
+
+Smoke-tested against a local QuestDB (OSS) with all rows accounted for server-side:
+
+| Transport | Run | Result |
+| --- | --- | --- |
+| `ilp` | 2000 rows, 1 worker | 2000 / 2000 |
+| `qwpudp` | 2000 rows, 1 worker | 2000 / 2000 |
+| `qwp` | 5000 rows, 1 worker | 5000 / 5000, distinct per-row timestamps; probe reported live timestamps + role |
+| `qwp` | 20000 rows, 4 workers | 20000 / 20000, one store-and-forward dir per worker |
+
+### QWP failover (primary crash)
+
+Two servers from the same build (primary on `:9100`, fallback on `:9000`), a single-worker
+`qwp` run of 60000 rows with `--addrs :9100,:9000` and the probe active. The primary was
+hard-killed mid-run:
+
+- The sender **completed all 60000 rows with exit 0**, despite its primary crashing.
+- The primary took `trade_id` `0-1`…`0-20000`; the fallback took `0-20001`…`0-60000`
+  (40000 rows, all distinct). Combined: a **contiguous 1–60000 with zero loss and zero
+  duplication** — store-and-forward replayed the in-flight transaction to the new host and
+  the handoff landed exactly on a transaction boundary.
+- The probe kept reporting across the crash and reconnected to the fallback transparently.
+
+Note: the probe's `on_failover_reset` callback fires only on *mid-query* failover; its
+`limit -1` polls return in microseconds, so a between-poll reconnect is silent (no
+`failed over` line). Comparing `reader.current_addr()` across polls would surface it.
diff --git a/rust/src/main.rs b/rust/src/main.rs
new file mode 100644
index 0000000..ec9f9b5
--- /dev/null
+++ b/rust/src/main.rs
@@ -0,0 +1,641 @@
+//! Parallel CSV replay sender for QuestDB, a Rust port of the Java `CsvParallelSender`.
+//!
+//! It replays a CSV of trades in a loop across N worker threads over one of three
+//! transports, selected with `--protocol`:
+//!
+//! * `qwp` - QWP over WebSocket: store-and-forward (un-acked frames spill to disk and
+//!   replay after an outage), transactional commit, multi-host failover. A background probe
+//!   thread polls the latest ingested timestamp and the serving node's live role via the
+//!   query client (the `Reader`).
+//! * `qwpudp` - QWP over UDP: fire-and-forget datagrams to :9007. Ingest-only,
+//!   unauthenticated, single-endpoint, best-effort (no ack, no failover).
+//! * `ilp` - the legacy ILP/HTTP transport.
+//!
+//! Unlike the Java client, the Rust/C client has no auto-flush: batching is driven purely
+//! by when we call `flush()`. We reproduce the Java cadence explicitly (see `run_worker`).
+
+use std::io::Read;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::thread;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use clap::Parser;
+use questdb::egress::column::ColumnView;
+use questdb::egress::reader::{BatchView, Reader};
+use questdb::egress::FailoverEvent;
+use questdb::ingress::{Sender, TimestampMicros};
+
+const PROBE_QUERY: &str = "select timestamp from trades limit -1";
+// Enterprise lifecycle status of whichever node the query client is connected to. Returns
+// the LIVE role (current_role / target_role), unlike the QWP handshake SERVER_INFO which is
+// only refreshed on a reconnect and so goes stale after an in-place primary<->replica switch.
+const STATUS_QUERY: &str = "switch status";
+
+#[derive(Parser, Debug)]
+#[command(
+    name = "csv_parallel_sender",
+    about = "Parallel CSV replay sender for QuestDB (QWP WebSocket/UDP, ILP/HTTP)"
+)]
+struct Args {
+    /// Comma-separated host:port list. QWP/WebSocket + ILP use :9000; QWP/UDP uses :9007.
+    #[arg(long, default_value = "localhost:9000")]
+    addrs: String,
+
+    /// Bearer token (QWP/WebSocket + ILP only; QWP/UDP is unauthenticated).
+    #[arg(long)]
+    token: Option<String>,
+
+    /// Basic-auth username (with --password).
+    #[arg(long)]
+    username: Option<String>,
+
+    /// Basic-auth password (with --username).
+    #[arg(long)]
+    password: Option<String>,
+
+    /// Total rows to send across all workers.
+    #[arg(long, default_value_t = 1_000_000)]
+    total_events: u64,
+
+    /// Per-row sleep in milliseconds (0 = as fast as possible).
+    #[arg(long, default_value_t = 50)]
+    delay_ms: u64,
+
+    /// Number of worker threads (each is its own Sender).
+    #[arg(long, default_value_t = 10)]
+    num_senders: usize,
+
+    /// Overall "keep retrying" budget in ms: retry_timeout (ILP) / reconnect_max_duration (QWP).
+    #[arg(long, default_value_t = 360_000)]
+    retry_timeout: u64,
+
+    /// CSV path (.csv or .csv.gz) with symbol,side,price,amount[,timestamp] columns.
+    #[arg(long, default_value = "./trades20250728.csv.gz")]
+    csv: String,
+
+    /// Use the CSV's own timestamp column instead of server/client "now".
+    #[arg(long, default_value_t = false)]
+    timestamp_from_file: bool,
+
+    /// Shift each timestamp by this many seconds (works with or without --timestamp-from-file).
+    #[arg(long, default_value_t = 0)]
+    seconds_offset: i64,
+
+    /// Transport: qwp | qwpudp | ilp.
+    #[arg(long, default_value = "qwp")]
+    protocol: String,
+
+    /// Store-and-forward key base (QWP/WebSocket). Each worker gets <sender-id>-<worker>.
+    #[arg(long, default_value = "ha_sender")]
+    sender_id: String,
+
+    /// Store-and-forward spill directory (QWP/WebSocket).
+    #[arg(long, default_value = "/tmp/qdb-sf")]
+    store_forward_dir: String,
+
+    /// Rows per flush. QWP commits every batch-size*batches-per-transaction; UDP flushes
+    /// every batch-size rows (keep small for UDP; large bursts overflow the receive buffer).
+    #[arg(long, default_value_t = 10_000)]
+    batch_size: u64,
+
+    /// QWP/WebSocket transaction size, in batches.
+    #[arg(long, default_value_t = 10)]
+    batches_per_transaction: u64,
+
+    /// Probe poll interval in ms (QWP/WebSocket only; 0 disables).
+    #[arg(long, default_value_t = 1000)]
+    probe_interval_ms: u64,
+
+    /// Enterprise only: request durable acks (QWP/WebSocket).
+    #[arg(long, default_value_t = false)]
+    enterprise: bool,
+
+    /// Preferred zone for the query client / probe (QWP/WebSocket).
+    #[arg(long, default_value = "eu-west-1")]
+    zone: String,
+}
+
+impl Args {
+    fn has_token(&self) -> bool {
+        self.token.as_deref().is_some_and(|t| !t.is_empty())
+    }
+
+    fn has_basic(&self) -> bool {
+        self.username.as_deref().is_some_and(|u| !u.is_empty())
+            && self.password.as_deref().is_some_and(|p| !p.is_empty())
+    }
+
+    fn tls(&self) -> bool {
+        self.has_token() || self.has_basic()
+    }
+
+    fn addr_list(&self) -> Vec<String> {
+        self.addrs
+            .split(',')
+            .map(|s| s.trim().to_string())
+            .filter(|s| !s.is_empty())
+            .collect()
+    }
+}
+
+struct TradeRow {
+    symbol: String,
+    side: String,
+    price: f64,
+    amount: f64,
+    /// Microseconds since epoch, parsed at load time; only used with --timestamp-from-file.
+    ts_micros: i64,
+}
+
+fn now_micros() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_micros() as i64)
+        .unwrap_or(0)
+}
+
+fn main() {
+    let args = Args::parse();
+
+    if !matches!(args.protocol.as_str(), "qwp" | "qwpudp" | "ilp") {
+        eprintln!("--protocol must be 'qwp', 'qwpudp', or 'ilp', got: {}", args.protocol);
+        std::process::exit(2);
+    }
+    if args.num_senders == 0 {
+        eprintln!("--num-senders must be > 0");
+        std::process::exit(2);
+    }
+    if args.total_events == 0 {
+        eprintln!("--total-events must be > 0");
+        std::process::exit(2);
+    }
+    if args.batch_size == 0 {
+        eprintln!("--batch-size must be > 0");
+        std::process::exit(2);
+    }
+    if args.batches_per_transaction == 0 {
+        eprintln!("--batches-per-transaction must be > 0");
+        std::process::exit(2);
+    }
+
+    let addrs = args.addr_list();
+    if addrs.is_empty() {
+        eprintln!("--addrs must contain at least one host:port");
+        std::process::exit(2);
+    }
+    // QWP/UDP is single-endpoint (no failover) and unauthenticated. Enforce both upfront so
+    // the failure is a clear message here, not a per-worker exception mid-run.
+    if args.protocol == "qwpudp" {
+        if addrs.len() > 1 {
+            eprintln!(
+                "--protocol qwpudp supports a single host:port only (UDP is connectionless, \
+                 there is no failover); got {} addresses",
+                addrs.len()
+            );
+            std::process::exit(2);
+        }
+        if args.has_token() || args.username.is_some() || args.password.is_some() {
+            eprintln!(
+                "[warn] --protocol qwpudp is unauthenticated (UDP accepts any connection); \
+                 ignoring --token/--username/--password"
+            );
+        }
+    }
+
+    let rows = match load_csv(&args.csv, args.timestamp_from_file) {
+        Ok(rows) if !rows.is_empty() => rows,
+        Ok(_) => {
+            eprintln!("CSV has no data rows: {}", args.csv);
+            std::process::exit(2);
+        }
+        Err(e) => {
+            eprintln!("Failed to read CSV {}: {e}", args.csv);
+            std::process::exit(2);
+        }
+    };
+
+    match args.protocol.as_str() {
+        "qwp" => println!(
+            "Ingestion started. Protocol: qwp (WebSocket, sender-id={}, store-and-forward={}, \
+             batch-size={}, batches-per-transaction={}, retry-timeout-ms={}) | addrs: {}",
+            args.sender_id, args.store_forward_dir, args.batch_size, args.batches_per_transaction,
+            args.retry_timeout, addrs.join(",")
+        ),
+        "qwpudp" => println!(
+            "Ingestion started. Protocol: qwpudp (QWP/UDP datagrams, ingest-only, unauthenticated, \
+             no store-and-forward, no failover; query client disabled, batch-size={}) | addr: {}",
+            args.batch_size, addrs[0]
+        ),
+        _ => println!(
+            "Ingestion started. Protocol: ilp (HTTP, retry-timeout-ms={}) | addrs: {}",
+            args.retry_timeout, addrs.join(",")
+        ),
+    }
+
+    let total_sent = AtomicU64::new(0);
+    let stop = AtomicBool::new(false);
+    let start = Instant::now();
+
+    let base = args.total_events / args.num_senders as u64;
+    let rem = args.total_events % args.num_senders as u64;
+    let mut worker_errors: Vec<String> = Vec::new();
+
+    thread::scope(|scope| {
+        let args_ref = &args;
+        let rows_ref = &rows;
+        let sent_ref = &total_sent;
+        let stop_ref = &stop;
+
+        // Progress reporter: rows/s (client-side, sent), once per second.
+        scope.spawn(move || {
+            let mut last = 0u64;
+            while !stop_ref.load(Ordering::Relaxed) {
+                thread::sleep(Duration::from_millis(1000));
+                let now = sent_ref.load(Ordering::Relaxed);
+                println!("[progress] sent={now} rate={} rows/s", now - last);
+                last = now;
+            }
+        });
+
+        // Probe (QWP/WebSocket only): ingest-only transports (qwpudp, ilp) have no query path.
+        if args_ref.protocol == "qwp" && args_ref.probe_interval_ms > 0 {
+            scope.spawn(move || run_probe(args_ref, stop_ref));
+        }
+
+        let mut handles = Vec::with_capacity(args.num_senders);
+        for id in 0..args.num_senders {
+            let events = base + if (id as u64) < rem { 1 } else { 0 };
+            handles.push(scope.spawn(move || run_worker(id, events, args_ref, rows_ref, sent_ref)));
+        }
+
+        for (id, h) in handles.into_iter().enumerate() {
+            match h.join() {
+                Ok(Ok(())) => {}
+                Ok(Err(e)) => worker_errors.push(format!("Sender {id}: {e}")),
+                Err(_) => worker_errors.push(format!("Sender {id}: panicked")),
+            }
+        }
+        stop_ref.store(true, Ordering::Relaxed);
+    });
+
+    if !worker_errors.is_empty() {
+        for e in &worker_errors {
+            eprintln!("Worker failed: {e}");
+        }
+        std::process::exit(1);
+    }
+
+    let elapsed = start.elapsed().as_secs_f64();
+    let rate = if elapsed > 0.0 { args.total_events as f64 / elapsed } else { 0.0 };
+    println!(
+        "All workers completed. protocol={} events={} elapsed={:.3} s throughput={:.0} rows/s",
+        args.protocol, args.total_events, elapsed, rate
+    );
+}
+
+fn run_worker(
+    worker_id: usize,
+    total_events: u64,
+    args: &Args,
+    rows: &[TradeRow],
+    total_sent: &AtomicU64,
+) -> questdb::Result<()> {
+    println!("Sender {worker_id} will send {total_events} events");
+
+    let is_qwp = args.protocol == "qwp";
+    let is_udp = args.protocol == "qwpudp";
+    // Single worker on a QWP transport: stamp each row with the current micros client-side.
+    // A single thread is monotonic (no O3) and this avoids QWP's per-batch atNow() stamping.
+    // ILP, or more than one worker, use atNow() (server-side, O3-safe).
+    let per_row_micros = (is_qwp || is_udp) && args.num_senders == 1;
+    // Flush cadence: qwp commits every batch-size*batches-per-transaction rows; qwpudp flushes
+    // every batch-size rows (bounds the datagram burst); ilp flushes every batch-size rows too
+    // (the Rust HTTP client has no auto-flush, so the buffer must be drained explicitly).
+    let commit_every: u64 = if is_qwp {
+        args.batch_size * args.batches_per_transaction
+    } else {
+        args.batch_size
+    };
+
+    let mut sender = Sender::from_conf(build_ingest_conf(args, worker_id))?;
+    let mut buffer = sender.new_buffer();
+    let n = rows.len();
+    let mut sent: u64 = 0;
+
+    for i in 0..total_events {
+        let r = &rows[(i as usize) % n];
+        buffer
+            .table("trades")?
+            .symbol("symbol", r.symbol.as_str())?
+            .symbol("side", r.side.as_str())?
+            .column_f64("price", r.price)?
+            .column_f64("amount", r.amount)?
+            .column_str("trade_id", format!("{worker_id}-{}", i + 1))?;
+
+        if args.timestamp_from_file {
+            let ts = r.ts_micros + args.seconds_offset * 1_000_000;
+            buffer.at(TimestampMicros::new(ts))?;
+        } else if args.seconds_offset != 0 {
+            buffer.at(TimestampMicros::new(now_micros() + args.seconds_offset * 1_000_000))?;
+        } else if per_row_micros {
+            buffer.at(TimestampMicros::new(now_micros()))?;
+        } else {
+            buffer.at_now()?;
+        }
+
+        sent += 1;
+        total_sent.fetch_add(1, Ordering::Relaxed);
+
+        if sent.is_multiple_of(commit_every) {
+            sender.flush(&mut buffer)?;
+        }
+        if args.delay_ms > 0 {
+            thread::sleep(Duration::from_millis(args.delay_ms));
+        }
+    }
+
+    sender.flush(&mut buffer)?;
+    // QWP/WebSocket: drain any un-acked store-and-forward frames before closing.
+    if is_qwp {
+        sender.close_drain()?;
+    }
+    println!("Sender {worker_id} finished sending {sent} events");
+    Ok(())
+}
+
+/// Build the ingestion connect string for the configured transport.
+fn build_ingest_conf(args: &Args, worker_id: usize) -> String {
+    let addrs = args.addr_list();
+    match args.protocol.as_str() {
+        "qwpudp" => format!("qwpudp::addr={};", addrs[0]),
+        "ilp" => {
+            let scheme = if args.tls() { "https" } else { "http" };
+            let mut c = format!("{scheme}::addr={};", addrs.join(","));
+            append_auth(&mut c, args);
+            if args.tls() {
+                c.push_str("tls_verify=unsafe_off;");
+            }
+            c.push_str(&format!("retry_timeout={};", args.retry_timeout));
+            c
+        }
+        _ => {
+            // qwp (WebSocket)
+            let scheme = if args.tls() { "qwpwss" } else { "qwpws" };
+            let who = format!("{}-{worker_id}", args.sender_id);
+            let sf = format!("{}/{who}", args.store_forward_dir);
+            let _ = std::fs::create_dir_all(&sf);
+            let mut c = format!("{scheme}::addr={};", addrs.join(","));
+            append_auth(&mut c, args);
+            if args.tls() {
+                c.push_str("tls_verify=unsafe_off;");
+            }
+            c.push_str(&format!("sender_id={who};"));
+            c.push_str(&format!("sf_dir={sf};"));
+            c.push_str(&format!("reconnect_max_duration_millis={};", args.retry_timeout));
+            c.push_str("reconnect_initial_backoff_millis=100;");
+            c.push_str("reconnect_max_backoff_millis=5000;");
+            if args.enterprise {
+                c.push_str("request_durable_ack=true;");
+            }
+            c
+        }
+    }
+}
+
+fn append_auth(c: &mut String, args: &Args) {
+    if args.has_token() {
+        c.push_str(&format!("token={};", args.token.as_ref().unwrap()));
+    } else if args.has_basic() {
+        c.push_str(&format!(
+            "username={};password={};",
+            args.username.as_ref().unwrap(),
+            args.password.as_ref().unwrap()
+        ));
+    }
+}
+
+/// Connect string for the QWP query client (probe): same hosts/auth as the senders,
+/// target=any (replica-fallback reads), failover on, zone bias.
+fn build_reader_conf(args: &Args) -> String {
+    let addrs = args.addr_list();
+    let scheme = if args.tls() { "wss" } else { "ws" };
+    let mut c = format!("{scheme}::addr={};target=any;failover=true;", addrs.join(","));
+    append_auth(&mut c, args);
+    if args.tls() {
+        c.push_str("tls_verify=unsafe_off;");
+    }
+    if !args.zone.is_empty() {
+        c.push_str(&format!("zone={};", args.zone));
+    }
+    c.push_str("compression=raw;");
+    c
+}
+
+/// Probe thread: every interval, poll the latest ingested timestamp and the serving node's
+/// live role over a QWP query client. Independent of the senders; fails over automatically.
+fn run_probe(args: &Args, stop: &AtomicBool) {
+    let mut reader = match Reader::from_conf(build_reader_conf(args)) {
+        Ok(r) => r,
+        Err(e) => {
+            eprintln!("[query client] connect failed: {e}");
+            return;
+        }
+    };
+    if let Some(si) = reader.server_info() {
+        println!(
+            "[query client] connected, serving node={} role={} zone={} cluster={}",
+            or_none(&si.node_id),
+            si.role.as_str(),
+            or_none(si.zone_id.as_deref().unwrap_or("")),
+            or_none(&si.cluster_id)
+        );
+    } else {
+        println!("[query client] connected");
+    }
+
+    let mut was_down = false;
+    while !stop.load(Ordering::Relaxed) {
+        match read_latest_ts(&mut reader) {
+            Ok(latest) => {
+                if was_down {
+                    println!("[query client] connection restored");
+                    was_down = false;
+                }
+                if let Some(micros) = latest {
+                    // trades designated timestamp is microseconds by default.
+                    let ts = chrono::DateTime::from_timestamp_micros(micros)
+                        .map(|dt| dt.to_rfc3339_opts(chrono::SecondsFormat::Micros, true))
+                        .unwrap_or_else(|| micros.to_string());
+
+                    // Ask the serving node for its live role. A status-query failure must not
+                    // look like a connection loss, so it is swallowed (current stays None).
+                    let (current, target) = read_switch_status(&mut reader).unwrap_or((None, None));
+
+                    let (node, zone) = match reader.server_info() {
+                        Some(si) => (
+                            or_none(&si.node_id),
+                            or_none(si.zone_id.as_deref().unwrap_or("")),
+                        ),
+                        None => ("(none)".to_string(), "(none)".to_string()),
+                    };
+
+                    let served = match current {
+                        Some(role) => {
+                            let switching = target
+                                .as_deref()
+                                .is_some_and(|t| !t.eq_ignore_ascii_case(&role));
+                            let sw = match (&target, switching) {
+                                (Some(t), true) => format!(" (switching -> {t})"),
+                                _ => String::new(),
+                            };
+                            format!(" served by role={role}{sw} node={node} zone={zone}")
+                        }
+                        None => {
+                            let handshake = reader
+                                .server_info()
+                                .map(|si| si.role.as_str())
+                                .unwrap_or_else(|| "unknown".to_string());
+                            format!(
+                                " served by role={handshake} node={node} zone={zone} \
+                                 (handshake role; live 'switch status' unavailable, may be stale)"
+                            )
+                        }
+                    };
+                    println!("[probe] latest trades timestamp = {ts} (raw={micros}){served}");
+                }
+            }
+            Err(e) => {
+                if !was_down {
+                    println!("[query client] connection lost ({e}), will retry");
+                    was_down = true;
+                }
+            }
+        }
+        thread::sleep(Duration::from_millis(args.probe_interval_ms));
+    }
+}
+
+/// Run PROBE_QUERY and return the last non-null timestamp (micros), if any.
+fn read_latest_ts(reader: &mut Reader) -> Result<Option<i64>, questdb::egress::Error> {
+    let mut latest: Option<i64> = None;
+    let mut cursor = reader
+        .prepare(PROBE_QUERY)
+        .on_failover_reset(|ev: &FailoverEvent| {
+            println!(
+                "[query client] failed over {} -> {} (attempt {})",
+                ev.failed_addr, ev.new_addr, ev.attempts
+            );
+        })
+        .execute()?;
+    while let Some(view) = cursor.next_batch()? {
+        if let Ok(ColumnView::Timestamp(col)) = view.column(0) {
+            for r in 0..view.row_count() {
+                if !col.is_null(r) {
+                    latest = Some(col.value(r));
+                }
+            }
+        }
+    }
+    Ok(latest)
+}
+
+/// Run STATUS_QUERY and pull current_role / target_role out of the result by column name.
+fn read_switch_status(
+    reader: &mut Reader,
+) -> Result<(Option<String>, Option<String>), questdb::egress::Error> {
+    let mut current: Option<String> = None;
+    let mut target: Option<String> = None;
+    let mut cursor = reader.prepare(STATUS_QUERY).execute()?;
+    while let Some(view) = cursor.next_batch()? {
+        let names: Vec<String> = view
+            .schema()
+            .columns()
+            .iter()
+            .map(|c| c.name.to_lowercase())
+            .collect();
+        for r in 0..view.row_count() {
+            for (i, name) in names.iter().enumerate() {
+                if !name.contains("role") {
+                    continue;
+                }
+                let value = col_string(&view, i, r);
+                if name.contains("current") {
+                    current = value;
+                } else if name.contains("target") {
+                    target = value;
+                } else if current.is_none() {
+                    // A single unqualified "role" column (older servers) is the current one.
+                    current = value;
+                }
+            }
+        }
+    }
+    Ok((current, target))
+}
+
+/// Read a string value from a Symbol or Varchar column; None for null or non-string types.
+fn col_string(view: &BatchView, col: usize, row: usize) -> Option<String> {
+    match view.column(col) {
+        Ok(ColumnView::Symbol(c)) => c.resolve(row).map(|s| s.to_string()),
+        Ok(ColumnView::Varchar(c)) => c.value(row).map(|s| s.to_string()),
+        _ => None,
+    }
+}
+
+fn or_none(s: &str) -> String {
+    if s.is_empty() {
+        "(none)".to_string()
+    } else {
+        s.to_string()
+    }
+}
+
+/// Load the CSV (optionally gzipped) into memory. Requires symbol,side,price,amount and,
+/// when `need_timestamp`, a timestamp column parsed to microseconds since epoch.
+fn load_csv(path: &str, need_timestamp: bool) -> Result<Vec<TradeRow>, Box<dyn std::error::Error>> {
+    let file = std::fs::File::open(path)?;
+    let reader: Box<dyn Read> = if path.ends_with(".gz") {
+        Box::new(flate2::read::GzDecoder::new(file))
+    } else {
+        Box::new(file)
+    };
+    let mut rdr = csv::ReaderBuilder::new().has_headers(true).from_reader(reader);
+
+    let headers = rdr.headers()?.clone();
+    let idx = |name: &str| -> Result<usize, Box<dyn std::error::Error>> {
+        headers
+            .iter()
+            .position(|h| h.trim() == name)
+            .ok_or_else(|| format!("CSV missing required column: {name}").into())
+    };
+    let i_symbol = idx("symbol")?;
+    let i_side = idx("side")?;
+    let i_price = idx("price")?;
+    let i_amount = idx("amount")?;
+    let i_ts = if need_timestamp { Some(idx("timestamp")?) } else { None };
+
+    let mut out = Vec::new();
+    for rec in rdr.records() {
+        let rec = rec?;
+        if rec.is_empty() {
+            continue;
+        }
+        let ts_micros = match i_ts {
+            Some(i) => {
+                let raw = rec.get(i).unwrap_or("").trim();
+                chrono::DateTime::parse_from_rfc3339(raw)
+                    .map(|dt| dt.timestamp_micros())
+                    .map_err(|e| format!("bad timestamp {raw:?}: {e}"))?
+            }
+            None => 0,
+        };
+        out.push(TradeRow {
+            symbol: rec.get(i_symbol).unwrap_or("").trim().to_string(),
+            side: rec.get(i_side).unwrap_or("").trim().to_string(),
+            price: rec.get(i_price).unwrap_or("").trim().parse()?,
+            amount: rec.get(i_amount).unwrap_or("").trim().parse()?,
+            ts_micros,
+        });
+    }
+    Ok(out)
+}