Port the Java HA sender to the Rust, Python, and C/C++ clients#6
Open
javier wants to merge 27 commits into
Open
Port the Java HA sender to the Rust, Python, and C/C++ clients#6javier wants to merge 27 commits into
javier wants to merge 27 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.
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.
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).