Port the Java HA sender to the Rust, Python, and C/C++ clients

README.md

@@ -1,5 +1,21 @@
 # Sample Sender with Multiple Hosts for HA Ingestion
 
+## Client implementations
+
+This project is a QuestDB high-availability sender. The **Java** implementation in this
+repository is the **reference**. It is ported, in parity, to the three clients built on the
+same C/Rust core (the "CRusty" clients):
+
+- [`rust/`](./rust) - Rust port.
+- [`python/`](./python) - Python port, plus a pandas/polars ingestion and egress demo.
+- [`c/`](./c) - C/C++ port.
+
+Each mirrors the Java design (CSV replay loop, per-worker senders, the `qwp` / `qwpudp` /
+`ilp` transports, failover, store-and-forward, and the probe). Per-language build steps and
+client differences (for example, auto-flush is present in Java and Python but not in Rust or
+C/C++) are documented in that folder's `README.md`. The rest of this document describes the
+reference Java implementation.
+
 ## Compile
 
 `mvn  -DskipTests clean package`
@@ -46,6 +62,98 @@ java -jar target/ilp_sender-1.0-SNAPSHOT.jar \
 QWP. That is a different timeout from `--connect-timeout-ms` (below), which bounds a single
 connect attempt.
 
+## Pacing: `--delay-ms` vs `--rate`
+
+Two mutually exclusive ways to throttle generation:
+
+- `--delay-ms` (default `50`): a fixed `Thread.sleep(delay)` after **every row**, per worker.
+  Simple, but a poor throttle at speed: `sleep()` has millisecond granularity (and often
+  over-sleeps), and the fixed per-row cost caps throughput. `--delay-ms 1` yields well under
+  1000 rows/s; `--delay-ms 0` runs flat out (millions/s).
+- `--rate` (default `0` = off): a **target aggregate rate in rows/second across all workers**.
+  Each worker paces itself to its share (`rate / num-senders`) against a deadline schedule,
+  sending rows back-to-back and sleeping only when it runs *ahead* of schedule (coalescing many
+  rows into one sleep). This hits high targets a per-row delay never could: `--rate 300000`
+  holds ~300k rows/s regardless of `--num-senders`.
+
+When `--rate > 0` it **takes precedence** and `--delay-ms` is ignored (a warning is printed if
+both are set). Measured on a local QuestDB: `--rate 5000` → 4.03 s for 20k rows (~4966/s);
+`--rate 300000` → 2.03 s for 600k rows (~296k/s), steady per-second progress at the target.
+
+### Regenerating the replay CSV
+
+Two scripts regenerate the replay CSV. **Both always export from the public demo box**
+(`https://demo.questdb.io/exp`) on purpose: the demo is *continuously ingesting live market
+data*, so every export is a fresh, recent-price snapshot rather than a stale file. You then
+**recreate the table locally** from that CSV and replay it for internal demos. The sender
+always writes into a table named **`trades`** locally, regardless of which demo table the CSV
+came from.
+
+- **`regenerate_csv.sh`** — pulls the crypto **`trades`** table (BTC-USDT, ...). Its
+  timestamp is microseconds.
+  ```
+  ./regenerate_csv.sh                   # -> trades.csv, 1,000,000 most-recent rows
+  ./regenerate_csv.sh trades.csv.gz     # gzip output (matches the --csv default's .gz)
+  ./regenerate_csv.sh trades.csv 250000 # custom row count
+  ```
+- **`regenerate_csv_fx.sh`** — pulls the FX **`fx_trades`** table (EURCHF, AUDNZD, ...).
+  Two schema differences from the crypto table, both handled in the query
+  `select timestamp, symbol, side, price, quantity as amount from fx_trades order by timestamp desc limit N`:
+  its size column is **`quantity`**, aliased to **`amount`** so it matches the `trades` schema
+  the sender writes; and its designated timestamp is **`TIMESTAMP_NS`** (nanoseconds), not
+  micros (see [Nanosecond vs microsecond timestamps](#nanosecond-vs-microsecond-timestamps)).
+  ```
+  ./regenerate_csv_fx.sh                      # -> fx_trades.csv, 1,000,000 rows
+  ./regenerate_csv_fx.sh fx_trades.csv.gz     # gzipped
+  ./regenerate_csv_fx.sh fx_trades.csv 250000 # custom row count
+  ```
+
+Both export exactly `symbol, side, price, amount, timestamp` (the columns the sender reads),
+download to a temp file first so a failed fetch never clobbers a good CSV, and gzip-compress
+when the output path ends in `.gz` (the loader auto-detects `.gz`). Override `DEMO_HOST` only if
+you mirror the demo tables elsewhere.
+
+### Nanosecond vs microsecond timestamps
+
+The crypto `trades` table exports microsecond timestamps; the FX `fx_trades` table exports
+**nanosecond** timestamps (`TIMESTAMP_NS`, e.g. `...192508297Z`). This only ever matters with
+`--timestamp-from-file`; in the default replay mode the sender stamps `now()` and ignores the
+file timestamp entirely, so precision is irrelevant.
+
+**The generator sends timestamps at nanosecond resolution** (via `at(long, NANOS)`), and
+QuestDB stores them at the **target column's** resolution: a micros `TIMESTAMP` column silently
+truncates the extra digits, a `TIMESTAMP_NS` column keeps them. Cross-resolution replay is
+therefore safe in both directions and lands every row. Validated against a local QuestDB, 20k
+crypto + 20k FX rows each way (all 40k landing, no errors):
+
+| Local `trades` column | Replayed source | Stored as |
+| --- | --- | --- |
+| `timestamp` (micros) | crypto micros `...790999Z` | `...790999Z` |
+| `timestamp` (micros) | FX nanos `...192508297Z` | `...192508Z` (truncated to micros) |
+| `timestamp_ns` (nanos) | crypto micros `...790999Z` | `...790999000Z` |
+| `timestamp_ns` (nanos) | FX nanos `...192508297Z` | `...192508297Z` (**full nanos preserved**) |
+
+So if your local table is `TIMESTAMP_NS`, the FX source's nanosecond precision is preserved
+end-to-end; if it is micros, the sub-microsecond digits are truncated on store (lossless to the
+column's resolution, never an error). To recreate the local table at a given resolution,
+pre-create it before ingesting, e.g.:
+
+```sql
+-- microsecond trades table
+create table trades (symbol symbol, side symbol, price double, amount double,
+  trade_id varchar, timestamp timestamp) timestamp(timestamp) partition by day wal;
+
+-- nanosecond trades table (swap the timestamp type)
+create table trades (symbol symbol, side symbol, price double, amount double,
+  trade_id varchar, timestamp timestamp_ns) timestamp(timestamp) partition by day wal;
+```
+
+**If the table does not exist, the server auto-creates it as `TIMESTAMP_NS`** (nanosecond),
+because the generator sends nanosecond timestamps: the designated timestamp column is created
+at the resolution of the incoming value. Verified end-to-end: dropping `trades` and replaying
+the FX chunk with `--timestamp-from-file` recreated it with a `TIMESTAMP_NS` column holding the
+full `...192508297Z` value. Pre-create the table (above) if you want micros instead.
+
 ## Transport (`--protocol`)
 
 - `--protocol qwp` (default): QWP over WebSocket. Adds store-and-forward (un-acked

c/.gitignore

@@ -0,0 +1 @@
+/build

c/CMakeLists.txt

@@ -0,0 +1,30 @@
+cmake_minimum_required(VERSION 3.15)
+project(questdb_ha_sender_cpp CXX)
+
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+
+if(NOT CMAKE_BUILD_TYPE)
+    set(CMAKE_BUILD_TYPE Release)
+endif()
+
+# Path to a c-questdb-client checkout (v7.0.0+ / main, with QWP + egress).
+# Override with -DQUESTDB_CLIENT_DIR=/path/to/c-questdb-client.
+set(QUESTDB_CLIENT_DIR "/Users/j/prj/questdb/c-questdb-client"
+    CACHE PATH "c-questdb-client checkout")
+
+# The egress reader (query client / probe) needs the reader surface, and the probe
+# uses tls_verify=unsafe_off; enable both cargo features via the client's CMake options.
+set(QUESTDB_ENABLE_READER ON CACHE BOOL "" FORCE)
+set(QUESTDB_ENABLE_INSECURE_SKIP_VERIFY ON CACHE BOOL "" FORCE)
+
+add_subdirectory(${QUESTDB_CLIENT_DIR} ${CMAKE_BINARY_DIR}/questdb_client_build)
+
+find_package(ZLIB REQUIRED)
+
+add_executable(csv_parallel_sender csv_parallel_sender.cpp)
+target_include_directories(csv_parallel_sender PRIVATE ${QUESTDB_CLIENT_DIR}/include)
+target_link_libraries(csv_parallel_sender PRIVATE questdb_client ZLIB::ZLIB)
+
+find_package(Threads REQUIRED)
+target_link_libraries(csv_parallel_sender PRIVATE Threads::Threads)

c/README.md

@@ -0,0 +1,119 @@
+# C++ HA sender
+
+A C++ port of the Java/Rust/Python `CsvParallelSender`. 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 C/C++ client headers from
+[`c-questdb-client`](https://github.com/questdb/c-questdb-client) (`line_sender.hpp` for
+ingestion, `line_reader.hpp` for the egress query client) — the same C/Rust core the Rust and
+Python ports use. C++17.
+
+## Build
+
+Requires a `c-questdb-client` checkout (v7.0.0+ / `main`, which has QWP + the egress reader), a
+Rust toolchain (`cargo`, pulled in via the bundled Corrosion), a C++17 compiler, and zlib. The
+project builds the client as a CMake subdirectory (Corrosion compiles the Rust FFI), so point it
+at your checkout:
+
+```bash
+cmake -B build -S . -DQUESTDB_CLIENT_DIR=/path/to/c-questdb-client
+cmake --build build -j
+```
+
+`QUESTDB_CLIENT_DIR` defaults to `~/prj/questdb/c-questdb-client`. The CMake forces
+`QUESTDB_ENABLE_READER=ON` (the query client / probe) and `QUESTDB_ENABLE_INSECURE_SKIP_VERIFY=ON`
+(`tls_verify=unsafe_off`). The binary is `build/csv_parallel_sender`. The QWP wire protocol must
+match the target server build.
+
+## 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.
+- `ilp`: the legacy ILP/HTTP transport.
+
+## Run
+
+```bash
+./build/csv_parallel_sender \
+  --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 other ports: `--addrs`, `--token`/`--username`/`--password`, `--total-events`,
+`--delay-ms`, `--rate`, `--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`.
+(`--timestamp-from-file` and `--enterprise` are presence flags.) Confirm results server-side with
+`SELECT count() FROM trades`.
+
+## Pacing: `--delay-ms` vs `--rate`
+
+- `--delay-ms` (default `50`): a fixed sleep after **every row**, per worker.
+- `--rate` (default `0` = off): a **target aggregate rate in rows/second across all workers**.
+  Each worker paces to its share (`rate / num-senders`) against a deadline schedule, sending
+  rows back-to-back and sleeping only when it runs *ahead* — reaching high targets a per-row
+  sleep never could. When `> 0` it takes precedence over `--delay-ms` (a warning prints if both
+  are set). Measured: `--rate 5000`, 20k rows, 4 workers → ~5.0 s, steady ~5000 rows/s.
+
+## Timestamps
+
+The generator sends timestamps at **nanosecond** resolution (`timestamp_nanos`), and QuestDB
+stores them at the **target column's** resolution: a `TIMESTAMP_NS` column keeps full nanos, a
+micros `TIMESTAMP` column truncates the extra digits (silently, never an error). **If the table
+does not exist, the server auto-creates it as `TIMESTAMP_NS`.** Pre-create `trades` with a
+`timestamp` (micros) column if you want micros. The CSV parser already reads fractional seconds
+to full nanosecond precision. Verified: replaying the nanosecond FX chunk with
+`--timestamp-from-file` preserved `...192508297Z` in an auto-created `TIMESTAMP_NS` table.
+
+## Probe (QWP/WebSocket only)
+
+For `qwp`, a background thread uses the egress `reader` 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. Unlike the Python port, the C++ `reader` exposes `server_info()`
+(role/node/cluster), so the probe **does** fall back to the QWP handshake role when
+`switch status` is unavailable (OSS / missing SYSTEM ADMIN), labelled `(handshake role; ...)` —
+matching Java/Rust. The reader uses `target=any` (replica-fallback reads) and fails over across
+the `--addrs` hosts.
+
+## Differences from the other ports
+
+- **No auto-flush** (like Rust; unlike Java/Python) — the C/C++ client has none, so batching is
+  driven purely by explicit `flush()` at the same boundaries as the other ports.
+- **Reader scheme is `ws`/`wss`** (like Rust; Python required `qwpws`).
+- Row values from the CSV are validated UTF-8 via `utf8_view` at build time (throws on invalid
+  input); fixed column names use the `_cn` / `_tn` literals.
+- Threads via `std::thread`, one `line_sender` per worker (unique `sender_id` + spill dir).
+
+## Validated
+
+Against a local QuestDB (OSS), all rows accounted for server-side:
+
+| What | Result |
+| --- | --- |
+| `qwp`, 4000 rows, 1 worker | 4000 / 4000, distinct per-row timestamps; probe reported live timestamps + handshake role fallback |
+| `ilp`, 2000 rows, 2 workers | 2000 / 2000 |
+| `qwpudp` (paced) | rows land; best-effort, lossy under fast bursts |
+| `qwp`, `--rate 5000`, nanos FX, `--timestamp-from-file`, 4 workers | steady ~5000 rows/s; auto-created `TIMESTAMP_NS` table preserved `...192508297Z` |
+| 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–7500; the fallback took 7501–40000 (32500 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.