Port the Java HA sender to the Rust, Python, and C/C++ clients - #6
Open
javier wants to merge 46 commits into
Open
Port the Java HA sender to the Rust, Python, and C/C++ clients#6javier wants to merge 46 commits into
javier wants to merge 46 commits into
Conversation
Emit trade_id = <worker>-<1-based sequence> as a VARCHAR on each row, so completeness/gaps can be checked independent of timestamps and, with DEDUP UPSERT KEYS(timestamp, trade_id) on a single-worker client-timestamped run, store-and-forward replay after a failover is idempotent (drops the at-least-once boundary duplicates). High-cardinality, so a string column, not a symbol.
Add a Row id and dedup section covering the trade_id format, the completeness check, idempotent store-and-forward replay via DEDUP UPSERT KEYS(timestamp, trade_id), and the client-side-timestamp requirement.
Sender: - Detect the QWP upgrade "HTTP/1.1 400 Bad request" that a missing/invalid ILP token produces and append an explicit auth hint wherever it surfaces (endpoint-failed, worker error, probe-stopped). - Probe now reports the LIVE serving role via `switch status` each poll instead of the QWP handshake SERVER_INFO, which only refreshes on a reconnect and so goes stale after an in-place primary<->replica switch. Watchdog: - Monitor the primary's currentRole, not just health: react to a silent demotion to REPLICA as well as an unreachable node. - Bootstrap a fresh cluster: if no node is primary, promote the first healthy one in list order. - Fail over to the first HEALTHY node in preference order; adopt a node that is already PRIMARY instead of forcing another switch. - Add a configurable grace period (QDB_WD_GRACE_PERIOD, default 5s) so a manual operator switch can settle before the watchdog acts. - Never exit on its own; keep retrying when nothing can be promoted. - Remove the now-unused get_http_code and STARTUP_RETRIES. Update README and the sample env accordingly.
Build/target: - pom default questdb.client.version -> 1.3.6-SNAPSHOT (required; QWP and the lifecycle switch features are not stable earlier, and it is not on Central). - Drop the RECONNECT_BUDGET_EXHAUSTED switch case in both senders; that kind was removed from SenderConnectionEvent.Kind in 1.3.6, so it no longer compiles. Diagnostics: - Replace the brittle "400 Bad request" string-match with the client's typed signals: QwpAuthFailedException (401/403), WebSocketUpgradeException.isRoleMismatch() (421 = endpoint is a replica, no primary to accept writes), and status-400 (missing/malformed token). The listener passes the real throwable, not its message. Timeouts: - --retry-timeout now also drives QWP (reconnectMaxDurationMillis), not just ILP's retryTimeoutMillis, so one flag means the same "keep retrying" budget on both. - New --connect-timeout-ms (default 3000, 0 = OS default): connectTimeoutMillis on the senders and connect_timeout= on the probe, so a black-holed host fails over in seconds instead of riding the OS connect timeout. ILP left unbounded/unchanged. - Watchdog: make the role/health poll timeouts configurable (QDB_WD_CHECK_CONNECT_TIMEOUT / QDB_WD_CHECK_MAX_TIME; defaults 2/3, no behaviour change). Update README and the watchdog sample env accordingly.
The probe fallback printed only "node=(none)" and dropped the role entirely when `switch status` returned nothing, which was worse than before. Restore the role/node/zone from getServerInfo() on every line (it tracks connect + failover), and append the authoritative live role from `switch status` only when present. When switch status yields no live role, log why at most once per 30s: an error status+message, a returned batch with no *role* column (names listed), or no row batch at all -- so the empty case is diagnosable instead of silent.
…dshake The probe printed getServerInfo()'s role as the headline 'served by role=', but that is the QWP handshake role, refreshed only on connect/failover. After an in-place promotion (the read connection never drops) it stayed stuck at REPLICA even though the node was serving reads as PRIMARY. Make switch status' current_role the authoritative role shown, append '(switching -> ROLE)' while a switch is in flight, and demote getServerInfo() to supplying node/zone only -- its handshake role is now a clearly labelled fallback used solely when the status query is unavailable. target=any is unchanged, so replica-fallback reads still work.
The watchdog Configuration section only named a few QDB_WD_* vars inline and deferred the rest to the env example; enumerate all 14 (required + optional) with defaults and purpose. Also update the probe example to the current 'served by role=<current_role> node=... zone=...' format, covering the '(switching -> ROLE)' in-flight case and the labelled handshake fallback.
Add --protocol qwpudp for fire-and-forget UDP datagram ingest to the UDP port (:9007). It is ingest-only, unauthenticated (auth flags are warned about and ignored), and single-endpoint with no failover, store-and-forward, TLS, or query client (the probe is skipped). The worker flushes datagrams every --batch-size rows and, for a single worker, stamps rows client-side; multiple workers use server-side atNow(). Document the transport, including its best-effort nature and the burst/packet-loss behaviour, with guidance to keep --batch-size small for UDP.
Port the Java CsvParallelSender to Rust, built on the local questdb-rs (c-questdb-client) as a path dependency. Mirrors the Java client: - qwp (WebSocket) with per-worker store-and-forward, transactional commit cadence, and multi-host failover. - qwpudp (UDP) fire-and-forget datagrams: ingest-only, unauthenticated, single-endpoint, with upfront single-address validation. - ilp (HTTP) legacy transport. - Same connection options (address list, token / basic auth, TLS, zone, enterprise durable ack, reconnect budget) and timestamp semantics (single-worker client micros vs multi-worker at_now). - A Reader-based probe (qwp only) polling the latest ingested timestamp and the serving node's live role via 'switch status', with target=any replica-fallback reads and automatic failover. Batching is driven by explicit flush() calls since the Rust client has no auto-flush by design. Validated end-to-end against a local server, including a primary-crash failover with gap-free store-and-forward handoff.
Port the Java/Rust CsvParallelSender to Python on the questdb client's QWP + egress build (the sm_qwp_dataframe_bench branch, built from source; the PyPI release is ILP-only). csv_parallel_sender.py mirrors the other ports: qwp (WebSocket) with per-worker store-and-forward and manual flush cadence, qwpudp, and ilp, plus a probe using the pooled query Client (select ... limit -1 + switch status role reporting, target=any, failover). dataframe_demo.py shows pandas ingestion (Sender.dataframe), polars ingestion (Client.dataframe), and egress to pandas/polars via Client.query(...).to_pandas()/.to_polars(). Validated against a local server: qwp/ilp/udp ingest, a primary-crash failover with gap-free store-and-forward handoff, and row-threshold auto-flush. README documents the build-from-source requirement and the Python-specific differences.
Port the Java/Rust/Python CsvParallelSender to C++ on the c-questdb-client C++ headers (line_sender.hpp ingestion, line_reader.hpp egress query client). Mirrors the other ports: qwp (WebSocket) with per-worker store-and-forward and manual flush cadence, qwpudp, and ilp, plus a probe (latest ingested timestamp + live serving role via switch status with a handshake-role fallback, target=any, failover). CMake build via Corrosion (cargo FFI), reads gzipped CSV via zlib. Validated against a local server: qwp/ilp/udp ingest and a primary-crash failover with a gap-free store-and-forward handoff.
javier
marked this pull request as ready for review
July 13, 2026 09:14
Add a Client implementations section explaining the Java implementation is the reference, ported in parity to the Rust, Python, and C/C++ clients, with links to each folder.
Add a --rate flag that targets an aggregate rows/second across all workers, pacing each worker to its share against a deadline schedule so it reaches high targets a per-row --delay-ms cannot. It takes precedence over --delay-ms. Ported to Java, Rust, Python, and C++. Send designated timestamps at nanosecond resolution everywhere. QuestDB stores at the target column resolution: TIMESTAMP_NS keeps full nanos, micros TIMESTAMP truncates, and a non-existent table is auto-created as TIMESTAMP_NS. Java now uses at long NANOS, Rust uses TimestampNanos with a nanosecond CSV parse, Python gained a nanosecond-preserving ISO parse, and C++ already sent nanos. Add regenerate_csv.sh and regenerate_csv_fx.sh to export fresh crypto and FX data from the always-on demo box. Document --rate, the nanosecond behavior, and the two scripts in each README.
Stream a table through polars, add a random enriched_rnd SYMBOL column, and write it back to enriched_<table>_demo. Reads via QueryResult.iter_arrow() one Arrow batch at a time (peak memory is a single batch, not the whole result), enriches each chunk, and ingests it via Client.dataframe. Reader and writer use separate QWP connections. Supports token/basic auth and TLS (qwpwss) for Enterprise, applied to both the QWP path and the HTTP drop/verify calls.
csv_columnar_sender.py: high-throughput Python ingestion over the columnar QWP path (Client.dataframe with polars), replacing row-by-row for volume. Streams the CSV in bounded chunks (flat memory), with --rate pacing, --num-senders, and enterprise auth/TLS. Row-by-row csv_parallel_sender.py stays for HA/failover demos (store-and-forward, probe, UDP/ILP), which the columnar path bypasses. read_bench.py: reads the last N rows as fast as possible via streaming iter_arrow() and reports rows/s plus MB/s and Gb/s (decoded Arrow payload). --readers splits the scan across N parallel connections by timestamp to get past the per-connection socket-buffer cap. boost_tcp.sh: raises net.core.wmem_max/rmem_max so the client's 4 MiB socket buffer is not clamped to ~416 KB (the per-connection throughput cap on real networks). Safe to source or exec. README: 'which script to use' guidance (columnar vs row-by-row) and network tuning.
An empty --token (typically --token "$ILP_TOKEN" with the env var unset) silently disabled auth+TLS, so the client connected plaintext to a secured server and hung retrying the handshake at sent=0, uninterruptible. Now: an empty --token (or a --username with empty --password) exits with a clear error instead of hanging. Worker threads are now daemon and joined via a polling loop, so KeyboardInterrupt is honored even while a worker is blocked in a native connect/flush.
Before starting workers, probe the server over the full path (TCP+TLS+auth+ 'select 1') in a daemon thread with a timeout. A down/unreachable/misconfigured server now fails loudly with a clear error instead of the ingest client retrying forever at sent=0. Default 10s; --connect-timeout 0 skips the check.
The progress counter only tracked rows appended to the client buffer, which with QWP store-and-forward runs far ahead of what the server has committed - so the per-second rate looked inflated and the end-of-run showed a long '0 rows/s' tail while the client drained the backlog. Now the reporter prints two counters: submitted (client-side, as before) and acknowledged (rows the server has actually committed). The acked count comes from the QWP ack watermark: each flush records its sequence via flushAndGetSequence(), and getAckedFsn() is mapped back to committed rows through a per-worker fsn->rows map - no extra query round-trips. A bounded awaitAckedFsn loop drains at the end while the acked counter climbs to the full count, so the tail shows real progress instead of zeros. The summary throughput is now labeled (acknowledged, end-to-end) and split into submit phase vs commit drain. Validated against a local QWP server: acknowledged trails submitted and converges to 100%. ILP/UDP fall back to the plain flush path.
Once all rows are submitted, silence the per-second progress and probe lines so the commit-drain tail no longer prints repeated '+0/s' / probe lines. The single final summary line (acknowledged, end-to-end + submit/drain split) is the last output. Connection/failover probe events still pass through.
qwpws/qwpwss are deprecated aliases; the client maps ws->QwpWs and wss->QwpWss. Switch the Rust and C++ ingest conf strings to ws/wss, matching their readers (which already use them) and the Java query client. Verified: both build against their client checkouts and ingest cleanly over ws against a local server. Python is left on qwpws/qwpwss - its binding's Protocol enum only exposes QwpWs/QwpWss and rejects ws/wss (ValueError: Invalid value for Protocol), even though the underlying Rust client accepts the aliases. Docs updated accordingly.
backfill.py spreads rows evenly across a historical window ([--start,--end), default yesterday 00:00 UTC .. today 13:00 UTC) at --rate-per-day density (default 500M/day -> ~771M rows over the window), ingesting them with those historical timestamps. Streams in bounded chunks (flat memory), splits the window across --num-senders workers (each a contiguous time block), columnar QWP path, with the same auth/TLS + connect-timeout preflight as the other scripts. Verified at small scale: timestamps land spread across the window. Cosmetic: the [conf] line now prints wss/ws instead of qwpwss/qwpws in csv_columnar_sender / read_bench / enrich_polars_demo / backfill. The actual connect string still uses qwpwss/qwpws (the Python binding requires them).
Target table is now configurable instead of hardcoded; auto-created if absent. Echoed in the startup line. Verified ingesting into a custom table.
Add --sample N (default 5): reader 0 wraps its first received Arrow batch as a polars DataFrame (zero-copy) and keeps its head - reusing data already streamed, no extra query and no re-scan. Rendered after the timing, so throughput is unaffected. Only one batch is ever held, so memory stays bounded.
blotter.py polls a table (or live view) at --rate Hz (default 5, max 20) and redraws a polars table in place (ANSI, no flicker). Builds 'select * from TABLE [WHERE ...] limit N' from three params: table, --where (WHERE auto-prepended if absent; trailing clauses like ORDER BY ride along; not sanitised - demo use), and --limit (default -10, magnitude clamped to 100). Query errors render in place instead of crashing, so the SELECT can be tweaked live. --once renders a single plain frame for scripting. Enterprise auth/TLS via --token/--username/--password (+ --tls-verify), plus --token-file/--token-label to read a bearer token from a file by label (keeps the secret off the command line). Verified against a remote core_price_lv live view. run_blotter.sh launches it with a default cluster address and $ILP_TOKEN, passing table/--where/--limit/--rate through; ADDR and PYTHON overridable via env.
--table replaces the positional arg. New --query runs full SQL verbatim (CTEs, window functions, ...); when set, --table/--where/--limit are ignored and a trailing ';' is stripped. Provide one of --table/--query. Multi-line SQL is collapsed to one line in the header; display is capped at 100 rows. Verified against a live view with a CTE + window query. run_blotter.sh uses --table with a commented --query example.
…resh ceiling The full-screen redraw each tick was terminal/SSH-output bound (froze past ~4 Hz over SSH), not query bound. draw() now diffs the new frame against the last and rewrites only the lines that changed - positioning each absolutely, writing it, and clearing to end-of-line - so static borders/header/rows are never re-sent. Cuts terminal output sharply and lets the refresh rate go well past the old ceiling.
A small http.server that queries QuestDB (QWP, token held server-side) and serves a self-contained dark-themed page that polls /data and renders a live table with cells flashing green/red on change. Rendering is in the browser, so it avoids the terminal/ SSH redraw bottleneck; only small JSON crosses the wire. Same query/auth params as blotter.py (--table/--query/--where/--limit, --token[-file/-label], --tls-verify). --host controls the bind address (default 127.0.0.1; --host 0.0.0.0 exposes it - the web server has no auth, so firewall the port).
protocol_version HTTP/1.1 so the browser reuses one TCP connection across polls (the per-poll q was dominated by a fresh connection/handshake each time over the laptop<->box link). daemon_threads so Ctrl+C exits immediately even with a request in flight.
Move the Python port off the 4.x `questdb.ingress.Client` split onto the 5.0 shape: - Import from top-level `questdb`; `Client.from_conf(conf)` -> `questdb.connect(conf)` returning a pooled `QuestDB` handle (`db`). - Conf schemes qwpws/qwpwss/qwpudp -> ws/wss/udp (now identical to the Rust/C ports; a conf string copies verbatim between clients). - db.query / db.dataframe / db.execute on the handle. Per script: - dataframe_demo: one handle for DDL (db.execute, replacing the urllib /exec hack), pandas + polars ingest via db.dataframe, dual egress; SYMBOL columns cast to pandas Categorical (columnar v1 requirement). - enrich_polars_demo: single pooled handle - the open read stream and each write borrow separate pooled connections. - csv_parallel_sender: keep a standalone per-worker questdb.Sender for independent store-and-forward; only the probe moves to connect(). - blotter / web_blotter / read_bench / backfill / csv_columnar_sender: Client -> connect(), scheme rename, db.* calls. - README rewritten for 5.0: build from jh_experiment_new_ilp, ws/wss/udp, the connect()/QuestDB shape, Python 3.12/3.13 + PyArrow caveat. All 8 scripts smoke-tested against a local QuestDB (ws/wss/http land 100%; udp is best-effort/lossy as documented).
A Go port of the Java/Rust/C++ CsvParallelSender on the go-questdb-client v4 QWP (WebSocket) transport. Replays a CSV across N worker goroutines, each with its own store-and-forward sender and multi-host failover, while a background probe queries the latest ingested timestamp and the serving node's role via a QwpQueryClient - ingest and query concurrently, with HA. - QWP (default): per-worker sf_dir slot, failover, acked-vs-submitted reporting (FlushAndGetSequence + AwaitAckedFsn). Empty --store-forward-dir selects memory mode (no disk durability) for raw happy-path throughput. - ILP over HTTP (--protocol ilp): ingest only (QWP-only query client). The full QWP transport is on the client's main, post-v4.2.0 and unreleased, so go.mod uses a local replace directive (documented in go/README.md). Smoke-tested against a local QuestDB: build/vet clean, ingest + probe work, memory-mode ingest ~1.33M rows/s vs disk-SF ~117k (fsync-bound) locally.
Building the 5.0 client with tls_verify=unsafe_off (self-signed enterprise certs over wss) requires the insecure-skip-verify Cargo feature, which the default build omits. Document the QUESTDB_INSECURE_SKIP_VERIFY=1 opt-in and warn against --no-build-isolation (numpy-header mismatch). Temporary until a client release ships the QWP/egress build on PyPI.
to_polars() builds SYMBOL Categoricals with a UInt32 -> Categorical cast that polars 1.43 deprecates. The warning message embeds a fresh Categories UUID on every call, so Python's once-per-location dedup never matches and the warnings re-fire at the poll rate, shredding the in-place redraw (and flooding the web_blotter log). Filter that one message; keep to_polars(), which is both the idiomatic call and the faster path on large results.
Ingestion restored itself after an outage but querying never did. Two causes, both in startProbe(): 1. QwpQueryClient.connect() ran once, outside the poll loop. A server that was down at startup threw straight past the loop into the outer catch, so the probe thread died and nothing restarted it. 2. A QwpQueryClient cannot recover on its own once a connection dies. With failover on (multiple --addrs), executeImpl leaves connected=false after the failover budget is spent and every later execute() throws "not connected; call connect() first". With failover off it is quieter: connected stays true, isConnected() keeps returning true, and each execute() short-circuits on the latched terminal failure and reports it through onError forever. Calling connect() again does not help -- it early-returns while connected is true. The client is now built inside the poll loop and torn down on any loss, so the next tick constructs a fresh one. Transport failures are distinguished from SQL errors by STATUS_INTERNAL_ERROR (0x06), which every socket-death terminal failure carries, so `switch status` failing on OSS is not mistaken for a disconnect. Verified against a 3-address failover=on config with a TCP proxy standing in for the server: connected -> connection lost -> connection restored, with the probe resuming afterwards.
A self-contained Docker Compose demo so a colleague can see the terminal blotter in action with nothing but Docker installed: - questdb.Dockerfile: QuestDB built from source at the LIVE VIEW commit (90a1b54), Java 25, with the web console (-P build-web-console). Served on host port 19000 (not 9000) so it never collides with a local QuestDB. - demo.Dockerfile: the unreleased questdb 5.0 Python client built from source (ea54b6f, Rust 1.91.1 + Cython), plus the feed and blotter. - feed.py: continuous synthetic crypto/FX bid feed, ~2000 rows/s at 20 flushes/s so the live view visibly moves. - entrypoint.sh: waits for QuestDB, creates the base table and the moving-average LIVE VIEW, runs the feed in the background and the blotter in the foreground, querying the live view directly. - vendor.sh: fallback to snapshot the client source if its branch is deleted before it merges. - README.md: run, repeat-run, and tiered teardown instructions. .dockerignore keeps the build context small.
…client Both components the demo previously built from source are now shippable: - QuestDB LIVE VIEW is in questdb/questdb:nightly (merged, commit 73685fa). Replace the from-source questdb.Dockerfile with image: questdb/questdb:nightly. - The Python client is released as questdb==5.0.0 on PyPI, with the query API the blotter uses (connect -> query -> to_polars). Replace the Rust + Cython client-build stage in demo.Dockerfile with a plain pip install. Delete questdb.Dockerfile and vendor.sh (the latter only snapshotted the unreleased client branch). Rewrite README for released components and add the one-line swap to questdb/questdb:10.0 once it ships. Build time drops from minutes to ~40s; nothing is pinned to a fragile branch SHA anymore. Verified end to end: nightly live view returns data and the 5.0.0 wheel renders the blotter with live moving averages.
Replace the per-row ingestion loop in feed.py with a vectorized path: each flush builds one Polars DataFrame (numpy-generated random walk) and ships it with a single Sender.dataframe(...) call. Over QWP/WebSocket that is a direct columnar bulk load, so a single process sustains ~1M rows/s where the row API capped far lower. New env knobs: - FEED_TARGET_RPS (default 2000, so the default demo is visually unchanged) sets the target rows/second; set 100000+ to stress ingestion. - FEED_WORKERS (default 1) forks N parallel sender processes, each targeting an equal share, to push past what one process generates. Batch size is derived as FEED_TARGET_RPS / FEED_WORKERS / FEED_FLUSH_HZ. Verified against nightly: Sender.dataframe accepts a Polars DataFrame directly (no pandas needed), single-process throughput ~560K-1.07M rows/s, and a full stack at FEED_TARGET_RPS=100000 measured ~93-98K rows/s server-side. Update README Knobs and "What it does" for the new rate controls.
Splitting the scan range by equal timestamp spans assumes rows are uniformly distributed in time. On a skewed table one reader is handed far more rows than the others, and because the slices run oldest to newest the first reader also owns the coldest end of the range, so it straggles long after the rest finish. Add --split with two modes, defaulting to rows: - rows: cut the range into --chunks equal row-count slices using LIMIT -m, -n, which takes the last m rows then drops the last n of those, giving a half-open range. Consecutive chunks tile with no gap and no overlap. Workers pull chunks off a shared queue, so no reader owns the cold end and a worker that lands on cached data just takes the next chunk. Needs no preliminary query and never has to know the designated timestamp's name. - time: the previous behaviour, kept because its boundaries are absolute timestamps computed once, so all readers agree on them even while the table is appended to. Row offsets are re-evaluated per connection and can shift under concurrent writes. The offset skip is metadata-only, not a scan: PageFrameRecordCursorImpl.skipRows walks whole page frames subtracting partitionHi minus partitionLo, and only descends into the frame where the skip lands. Measured on a 10.5B row table, skipping 10.4B rows to return a single row takes 12ms. Each worker now holds one connection for its whole chunk sequence via a db.reader lease, and passes reset_symbol_dict=False so the connection's SYMBOL dictionary stays warm instead of being rebuilt once per chunk. --chunks defaults to readers x 8. The done line now reports chunks per reader. Verified: row_chunks output parsed back into row ranges across 10 shapes, including uneven division and chunks greater than limit, tiles exactly. End to end on a 500k row table, all three paths return exactly --limit rows and the queue rebalances unevenly across workers as intended.
Minimal script: connect over QWP, run the cookbook's per-fill slippage query against fx_trades ASOF JOINed to market_data, and hand back a polars DataFrame. Address comes from QDB_ADDR and the bearer token from ILP_TOKEN, so no credential is stored in the file. Uses wss with tls_verify=unsafe_off, matching the flags read_bench already takes for the same host. The mid alias is spelled out in full inside both CASE branches. A select-list alias is not in scope for a sibling expression in the same SELECT, so referencing mid there does not resolve; wrapping the projection in a subquery is the alternative if the shorter form is wanted. Not run end to end: the target host is not reachable from this machine, so fx_trades' presence and schema are unverified. Only the Python is syntax checked.
A chunk whose row offsets fall past the end of the table still returns a valid result: a schema-only batch of 0 rows. Those chunks finish first, so a worker latched one, stored an empty DataFrame, and the is-None guard then blocked every later batch from replacing it. The sample printed a correct schema with no rows while the scan itself reported the right totals. This only became reachable with striping. Previously there was one query per reader and the sample was pinned to reader 0, so an empty leading batch was rare; with 64 chunks racing across 8 workers it is the common case whenever --limit exceeds the table's row count. Require batch.num_rows > 0 before latching. Reproduced and verified on a 100k row table with --limit 500000 and 32 chunks: before the fix, 'first 0 row(s)' shape 0x3; after, 'first 5 row(s)' shape 5x3. Regressions on --limit within the table size and on --split time both return exact row counts, and the sampled rows land on the expected chunk start.
slippage.py calls to_polars, which buffers the whole result before anything is visible. Add a variant using iter_polars, which yields one polars DataFrame per QWP batch, so rows print as they arrive. Measured on a 500k row result, warmed and run in both orders so cold start is not mistaken for streaming latency: iter_polars shows first rows at 8.5 to 18.2 ms against a 68.6 to 75.1 ms total, where to_polars shows nothing until 67.3 to 74.1 ms. Same total time, first rows roughly 8x sooner. Empty batches are skipped. A result can carry a schema-only batch of 0 rows, which is the same trap that made read_bench print an empty sample. Adds a QDB_CONF override so the script can point at a plain non-TLS instance; the wss plus ILP_TOKEN default is unchanged. Verified end to end against locally built fx_trades and market_data tables: the ASOF JOIN, the $yesterday interval, and all four computed columns resolve, and 200,000 rows arrive in 13 batches of 16,384.
fx_trades is scanned in designated-timestamp order, so the result already arrives oldest first and ORDER BY t.timestamp restates what the scan guarantees. Measured on a 12M row result before removing it: first batch at 41.3 ms with the clause and 31.2 ms without, totals 6.93s and 7.71s, i.e. within run-to-run variance either way. No sort is planned, so the clause cost nothing. Removing it is a clarity change, not a speedup. Worth recording what the same measurements did show, since the streamer looks slow next to a 60 ms query: the 60 ms is time to first result, not to all rows. First batch lands in ~41 ms, but the ASOF JOIN is per-row work done while streaming, so pulling every row costs in proportion to row count. On 12M rows a plain scan of fx_trades took 1.90s while the joined query took 8.28s on one connection; the join is roughly 77% of the time. Batch formatting is free and iter_polars over iter_arrow adds about 20%.
Splits yesterday into N contiguous half-open time ranges, gives each its own connection, and concatenates the results in range order, which is timestamp order. No preliminary query: the day's bounds are known, so the split is arithmetic. The merge is the fiddly part. SYMBOL columns arrive as polars Categorical and each connection's result carries its own Categories identity, a per-result UUID in the questdb_symbol namespace. pl.concat rejects that outright with 'Categories name mismatch ... failed to vstack column symbol', and how=vertical_relaxed does not rescue it either, failing with 'failed to determine supertype of cat and cat'. Casting every frame's categorical columns to one shared named Categories first is what makes it work, and it beats casting to String, 188ms against 216ms over 9M rows, while keeping the dtype. pl.enable_string_cache is deprecated in polars 1.42 and is not the answer here. Verified: 12,000,000 rows over 8 connections, spanning 00:00:00 to 23:59:59.99, timestamps monotonic, symbol/ecn/counterparty/side still Categorical. No speedup measured locally: 9.17s over 8 connections against roughly 8s on one, because the container's server is already CPU-saturated by the per-row ASOF JOIN, so more connections cannot help. The win, if any, is on a remote multi-core host where a single connection is round-trip bound rather than server bound. That remains unmeasured.
Combines the two previous variants. slippage_stream.py streams but on one connection; slippage_parallel.py uses N connections but shows nothing until the whole result is buffered. This fetches over N connections while printing in true timestamp order, starting as soon as the first batch lands. Each worker owns a bounded queue. All workers fetch at once, but the printer drains range 0 to completion, then range 1, and so on; since the ranges are contiguous and consumed in order, the output is exactly what a single query would have produced. The bound is what stops this being a disguised buffer-everything: a worker that runs ahead blocks once its queue is full, capping peak memory at readers x QUEUE_DEPTH batches, and the backpressure reaches the server through QWP flow control. Verified on 12M rows over 8 connections: ranges printed strictly 0 through 7, exactly 92 batches each, never interleaved. 7.76s at 1,546,460 rows/s. Tradeoff worth knowing: first rows appear after 786 ms here, against roughly 40 ms for the single-connection streamer, because eight workers contend at startup. This variant optimises total throughput, not time to first paint. As with slippage_parallel, no large speedup was measurable locally because the test server is CPU-saturated by the per-row ASOF JOIN.
Because the per-worker queues are bounded, a worker blocks once its queue fills, so the printer sets the pace for every fetcher. Rendering all ~730 batches of a 12M row result emits 3.87 MB of box-drawing tables. Piped to a file that is free, which is why it never showed up in local timings; over SSH it is slow enough to stall all eight fetchers behind it. That is the asymmetry against slippage_parallel.py, which prints once at the end and so never waits on the terminal. Render the first SLIPPAGE_FULL_FRAMES batches (default 2) as full tables and one compact line per batch after that. Output drops from 3,869,222 to 53,165 bytes, 98.6% less, while the streaming progress feel is unchanged. Set the variable to 0 for progress lines only, or high to restore rendering everything. Also raise QUEUE_DEPTH from 4 to 8 so a brief stall in the printer does not immediately block the fetchers. Honest about what is measured: locally this changes nothing (9.16s rendering everything against 9.94s with the new default, i.e. noise), because a local pipe is not the bottleneck. The 98.6% byte reduction is the measured effect; the speedup is expected on a real terminal but was not reproducible here. Ordering is unaffected: ranges still print strictly 0 through 7, 92 batches each.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This branch ports the Java
CsvParallelSender— the QWP high-availability sender (failover, store-and-forward, dual ingest/query, and the shared connection options) — to the three clients built on the C/Rust core (the "CRusty" clients): Rust, Python, and C/C++.Each port keeps the Java design (CSV replay loop, per-worker senders, timestamp / O3 semantics) and mirrors the same
--protocoltransports, connection options, and flags. Per-language differences are documented in each folder'sREADME.md.Status — all three complete
rust/) — QWP (WebSocket) with per-worker store-and-forward and multi-host failover, QWP/UDP, and ILP/HTTP; aReader-based probe (latest ingested timestamp + live serving role viaswitch status,target=anyreplica-fallback reads, automatic failover). Validated end-to-end, including a primary-crash failover with a gap-free store-and-forward handoff.python/) — the same three transports plus a pooled queryClient, and a pandas/polars ingestion + egress demo (dataframe_demo.py). Built on the client's QWP + egress branch (built from source; the PyPI release is ILP-only). Validated end-to-end, including a primary-crash failover and auto-flush.c/) — the same three transports (C++17,line_sender.hpp/line_reader.hpp), a probe with a handshake-role fallback, and a CMake build via Corrosion. Reads gzipped CSV via zlib. Validated end-to-end, including a primary-crash failover with a gap-free store-and-forward handoff.Each
README.mddocuments the per-language build and the client differences (e.g. auto-flush present in Java/Python but not Rust/C++; the handshake role exposed everywhere except the Python binding).