feat: QWP ingestion protocol#130
Open
jerrinot wants to merge 185 commits into
Open
Conversation
… columns Cover all column types over QWP/UDP transport with server-side validation: TimestampMicros, TimestampNanos, datetime (TIMESTAMP / TIMESTAMP_NS), f64 arrays (contiguous + transposed), decimal into pre-created DECIMAL(18,3) table, and VARCHAR string columns. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Auto-flush: bytes threshold triggers, row count triggers, disabled mode only sends on explicit flush. Buffer lifecycle: reuse after flush, independent buffers don't interfere, flush(clear=False) retains content, standalone Buffer.qwp() clear and reuse across multiple flushes. Edge cases: multi-table rows in one flush, Unicode symbols/columns (Zürich, 你好世界, 🚀), None values silently skipped, empty flush is no-op, double close, context manager flushes on exit, ServerTimestamp vs explicit TimestampNanos, max_name_len enforcement. Stress: 500 rows exercising multiple datagrams and auto-flush. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Auto-flush interval: verify time-based trigger after 500ms. Datagram splitting: 30 rows with max_datagram_size=200, single flush forces Rust to split across multiple datagrams — all arrive. Interleave HTTP and QWP/UDP senders in same process. from_env construction via QDB_CLIENT_CONF env var. Sender reuse: close and re-create, both sessions' data arrives. Large string: 1000-char value roundtrips through QWP/UDP. Symbols-only row (no columns). Mixed timestamps: ServerTimestamp and explicit TimestampNanos in same batch. DataFrame with datetime column as designated timestamp (at='ts'). new_buffer inherits init_buf_size and max_name_len from sender. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Mirrors the existing HTTP test_decimal_pyarrow but over QWP/UDP: PyArrow decimal128(18,2) column flushed via sender.dataframe() into a pre-created DECIMAL(18,3) table, values verified server-side. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Zero, negative zero, max precision (999999999999999.999), min non-zero (0.001), NaN/Inf rejection, multiple decimal columns with different scales, and PyArrow decimal128 with null values in DataFrame. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Buffer type mismatch: ILP buffer rejected by QWP sender, QWP buffer rejected by HTTP sender — Rust layer returns clear error messages. Network edge cases: wrong port flushes silently (UDP fire-and-forget), unresolvable host fails at establish() with DNS error. Data shape: wide row (50 columns + 5 symbols), row ordering across multiple datagrams (100 rows split with max_datagram_size=200), tiny datagram (max_datagram_size=1) rejected at flush. Load: 2000 rows rapid-fire with pure auto-flush, accept >= 90% arrival (UDP may drop under load). Config: protocol_version in QWP conf string rejected. Concurrency: two senders from different threads to same port. Lifecycle: double establish rejected, establish after close rejected. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
The designated timestamp column type is TIMESTAMP_NS (not TIMESTAMP) on newer QuestDB when using TimestampNanos. Use col_types dict lookup instead of full columns list comparison to avoid brittleness. Verified: 77/77 system tests pass with QDB_REPO_PATH. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- sender.rst overview: add QWP/UDP alongside ILP/TCP and ILP/HTTP - sender.rst tips: note QWP/UDP for fire-and-forget use cases - conf.rst: note protocol_version is not applicable for QWP/UDP Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Covers: no delivery guarantee, no error feedback, buffer inspection differences (bytes() empty, len() is estimate), standalone buffer requirements (Buffer.qwp()), auto-flush byte defaults, datagram size limit errors, and protocol_version inapplicability. Also adds QWP/UDP row to the protocol version auto-detection table. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds QWP/UDP and QWP/WebSocket support across C bindings, Cython runtime and stubs, Buffer factory refactor, Sender QWP/WebSocket runtime APIs, docs/examples, manifest, CI/build updates, and extensive unit/system tests. ChangesQWP Protocol Core Implementation
Documentation and Examples
Unit and Integration Tests
Maintenance
Sequence DiagramsequenceDiagram
participant ClientPy as Python Sender API
participant IngressPy as ingress.pyx runtime
participant LineSenderC as line_sender C lib
participant Trampoline as C trampoline
ClientPy->>IngressPy: Sender.flush / publish / flush_and_get_fsn
IngressPy->>LineSenderC: call C publish/flush APIs
LineSenderC->>Trampoline: emit QWPWS error/event view (if any)
Trampoline->>IngressPy: invoke Python error handler callback
IngressPy->>ClientPy: expose diagnostics via poll_qwp_ws_error / raise IngressServerRejectionError
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Suggested reviewers
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
examples/qwp_udp.py (1)
10-14: 💤 Low valueConsider noting that 1400 is the default
max_datagram_size.The
max_datagram_size=1400parameter is set to its default value. While being explicit in examples is fine, you might consider adding a comment to note this is the default, helping users understand they can omit it or adjust it as needed.📝 Optional comment addition
with Sender( Protocol.QwpUdp, host, port, - max_datagram_size=1400) as sender: + max_datagram_size=1400) as sender: # default is 1400🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/qwp_udp.py` around lines 10 - 14, The example explicitly passes max_datagram_size=1400 which is the default; update the Sender usage (the call to Sender with Protocol.QwpUdp) to either remove the parameter or add a short inline comment (next to max_datagram_size=1400) indicating "1400 is the default" so readers know it can be omitted or tuned as needed.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@examples/qwp_udp.py`:
- Around line 10-14: The example explicitly passes max_datagram_size=1400 which
is the default; update the Sender usage (the call to Sender with
Protocol.QwpUdp) to either remove the parameter or add a short inline comment
(next to max_datagram_size=1400) indicating "1400 is the default" so readers
know it can be omitted or tuned as needed.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: d1faad2b-3030-485c-bb29-bf483b70cb9c
📒 Files selected for processing (12)
c-questdb-clientdocs/conf.rstdocs/examples.rstdocs/installation.rstdocs/sender.rstexamples.manifest.yamlexamples/qwp_udp.pysrc/questdb/ingress.pyisrc/questdb/ingress.pyxsrc/questdb/line_sender.pxdtest/system_test.pytest/test.py
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/questdb/ingress.pyx (2)
2755-2760:⚠️ Potential issue | 🟠 Major | ⚡ Quick winMake sender setup rollback-safe when buffer reservation fails.
reserve_buffer()can now raise from_new_buffer_for_sender(). If that happens afterline_sender_build()succeeds,establish()leaves_impland_optslive on the sameSender, so retryingestablish()can leak or double-initialize native state.Suggested fix
if self._impl == NULL: raise c_err_to_py(err) + line_sender_opts_free(self._opts) + self._opts = NULL if self._buffer is None: - self._buffer = self._new_buffer_for_sender() - - line_sender_opts_free(self._opts) - self._opts = NULL + try: + self._buffer = self._new_buffer_for_sender() + except Exception: + line_sender_close(self._impl) + self._impl = NULL + raise
2214-2218:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winReject unknown
qwp_ws_progressvalues explicitly.
QwpWsProgress.parse()returnsNonefor an unmatched tag, so this turns bad user input into anAttributeErroron.c_valueinstead of a config error.Suggested fix
if qwp_ws_progress is not None: - c_qwp_ws_progress = QwpWsProgress.parse(qwp_ws_progress).c_value + progress = QwpWsProgress.parse(qwp_ws_progress) + if progress is None: + raise IngressError( + IngressErrorCode.ConfigError, + f'"qwp_ws_progress" has invalid value: {qwp_ws_progress!r}') + c_qwp_ws_progress = progress.c_value if not line_sender_opts_qwpws_progress( self._opts, c_qwp_ws_progress, &err): raise c_err_to_py(err)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/questdb/ingress.pyx` around lines 2214 - 2218, QwpWsProgress.parse(qwp_ws_progress) can return None for unknown tags, which currently leads to an AttributeError when accessing .c_value; update the logic around QwpWsProgress.parse(...) in ingress.pyx to check the parse result for None, and if None raise a clear config/validation error (rather than accessing .c_value), otherwise assign c_qwp_ws_progress = parsed.c_value and call line_sender_opts_qwpws_progress(self._opts, c_qwp_ws_progress, &err) as before; reference QwpWsProgress.parse, c_qwp_ws_progress, line_sender_opts_qwpws_progress, and c_err_to_py when implementing the explicit None check and error raise.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@setup.py`:
- Line 149: The build currently unconditionally appends the Rust feature
"insecure-skip-verify" to cargo_args; change this so "insecure-skip-verify" is
only added when an explicit opt-in environment variable is set (e.g.,
RUST_INSECURE_SKIP_VERIFY or ALLOW_INSECURE_SKIP_VERIFY), otherwise append only
"confstr-ffi". Locate the cargo_args assembly in setup.py where the line adds
['--features', 'confstr-ffi,insecure-skip-verify'] and replace it with logic
that conditionally includes "insecure-skip-verify" based on os.environ; also
document the env var in the packaging README. After that, inspect the Rust
workspace (search for the feature name "insecure-skip-verify" in the Rust
submodules/crates) to confirm where the feature is defined/used and ensure it
does not alter TLS verification for default release builds when the env opt-in
is not set.
In `@src/questdb/ingress.pyx`:
- Around line 2352-2366: The code currently treats bool as an int for
retry_max_backoff, so True/False are accepted; update the type checks in the
retry_max_backoff handling (symbols: retry_max_backoff,
line_sender_opts_retry_max_backoff, cp_timedelta, _timedelta_to_millis,
c_err_to_py, _fqn) to explicitly exclude bools—for example, change the int
branch to require isinstance(retry_max_backoff, int) and not
isinstance(retry_max_backoff, bool) (or check type(...) is int), and ensure bool
values fall through to the TypeError branch so passing True/False raises a
TypeError with the same message.
---
Outside diff comments:
In `@src/questdb/ingress.pyx`:
- Around line 2214-2218: QwpWsProgress.parse(qwp_ws_progress) can return None
for unknown tags, which currently leads to an AttributeError when accessing
.c_value; update the logic around QwpWsProgress.parse(...) in ingress.pyx to
check the parse result for None, and if None raise a clear config/validation
error (rather than accessing .c_value), otherwise assign c_qwp_ws_progress =
parsed.c_value and call line_sender_opts_qwpws_progress(self._opts,
c_qwp_ws_progress, &err) as before; reference QwpWsProgress.parse,
c_qwp_ws_progress, line_sender_opts_qwpws_progress, and c_err_to_py when
implementing the explicit None check and error raise.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 52ba2082-709e-4246-808f-df47556aa5e5
📒 Files selected for processing (9)
c-questdb-clientdocs/conf.rstdocs/sender.rstsetup.pysrc/questdb/ingress.pyisrc/questdb/ingress.pyxsrc/questdb/line_sender.pxdtest/system_test.pytest/test.py
✅ Files skipped from review due to trivial changes (2)
- c-questdb-client
- docs/sender.rst
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
test/system_test.py (2)
796-814:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winDocument
Sender.new_buffer()lifecycle preconditions in public stubs/docs
Sender.new_buffer()raisesIngressErrorif called beforeSender.establish()("new_buffer() can't be called before establish()") or afterSender.close()("new_buffer() can't be called: Sender is closed."), pertest/system_test.py(lines 796-814).src/questdb/ingress.pyi(anddocs/sender.rst) still describenew_buffer()without these required preconditions—update the public contract accordingly.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@test/system_test.py` around lines 796 - 814, The public type stub and docs don't state that Sender.new_buffer() may raise IngressError when called before Sender.establish() or after Sender.close(); update src/questdb/ingress.pyi and docs/sender.rst to document these lifecycle preconditions: indicate that new_buffer() requires prior Sender.establish() and will raise qi.IngressError with message "new_buffer() can't be called before establish()" if not established, and will raise qi.IngressError with message "new_buffer() can't be called: Sender is closed" if the Sender has been closed; reference the Sender.new_buffer, Sender.establish, Sender.close methods and the IngressError exception in the public contract.
1043-1058:⚠️ Potential issue | 🟠 Major | ⚡ Quick winFix UDP auto-flush system tests to avoid false positives from context-manager flush
In
test/system_test.py, the auto-flush tests (1043-1058,1060-1073,1344-1361,1592-1607) usewith self._mk_qwpudp_sender(...) as sender:. On normalwith-exit the sender context manager callsclose(flush=True), which flushes any buffered rows even if the specific byte/row/interval/rapid-fire trigger never fires—so these tests can pass while the trigger logic is broken.Use a manual sender lifetime instead: create the sender without the context manager, call
sender.establish(), and infinallycallsender.close(flush=False)so buffered data is not flushed on exit; keep/retain the existing server-side assertions (e.g.,retry_check_table(..., min_rows=10)andcountchecks) to validate that only the intended auto-flush path publishes rows.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@test/system_test.py` around lines 1043 - 1058, The auto-flush UDP tests currently use the context manager from _mk_qwpudp_sender which calls close(flush=True) on exit and masks failures; replace the with self._mk_qwpudp_sender(...) as sender: usage by creating the sender manually, call sender.establish() before sending rows, and ensure you close the sender in a finally block with sender.close(flush=False) so the context exit doesn't flush buffered rows; keep the existing server-side assertions (e.g., qdb_plain.retry_check_table(..., min_rows=10) and resp['count'] checks) unchanged to validate that only the intended auto-flush trigger causes publishes.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@test/system_test.py`:
- Around line 165-168: The _unused_tcp_port() function currently returns a port
after closing the socket, causing a TOCTOU race; instead bind and keep the
socket reserved and return the bound socket (and/or the port) so callers can use
the reserved endpoint and close the socket only when failover is set up; update
any call sites (including the other occurrence around lines 384-386) to accept
and later close the returned socket once the sender/receiver are using the port.
---
Outside diff comments:
In `@test/system_test.py`:
- Around line 796-814: The public type stub and docs don't state that
Sender.new_buffer() may raise IngressError when called before Sender.establish()
or after Sender.close(); update src/questdb/ingress.pyi and docs/sender.rst to
document these lifecycle preconditions: indicate that new_buffer() requires
prior Sender.establish() and will raise qi.IngressError with message
"new_buffer() can't be called before establish()" if not established, and will
raise qi.IngressError with message "new_buffer() can't be called: Sender is
closed" if the Sender has been closed; reference the Sender.new_buffer,
Sender.establish, Sender.close methods and the IngressError exception in the
public contract.
- Around line 1043-1058: The auto-flush UDP tests currently use the context
manager from _mk_qwpudp_sender which calls close(flush=True) on exit and masks
failures; replace the with self._mk_qwpudp_sender(...) as sender: usage by
creating the sender manually, call sender.establish() before sending rows, and
ensure you close the sender in a finally block with sender.close(flush=False) so
the context exit doesn't flush buffered rows; keep the existing server-side
assertions (e.g., qdb_plain.retry_check_table(..., min_rows=10) and
resp['count'] checks) unchanged to validate that only the intended auto-flush
trigger causes publishes.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
Comment on lines
+165
to
+168
| def _unused_tcp_port(): | ||
| with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: | ||
| sock.bind(('127.0.0.1', 0)) | ||
| return sock.getsockname()[1] |
There was a problem hiding this comment.
Reserve the failover port instead of probing and releasing it.
This is a TOCTOU race: _unused_tcp_port() closes the socket before the sender uses the endpoint, so another process can claim that port and make the failover path nondeterministic.
Proposed fix
+import contextlib
import socket
...
- `@staticmethod`
- def _unused_tcp_port():
- with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
- sock.bind(('127.0.0.1', 0))
- return sock.getsockname()[1]
+ `@staticmethod`
+ `@contextlib.contextmanager`
+ def _reserved_tcp_port():
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+ sock.bind(('127.0.0.1', 0))
+ yield sock.getsockname()[1]
...
- endpoints = [
- (self.qdb_plain.host, self._unused_tcp_port()),
- (self.qdb_plain.host, self.qdb_plain.http_server_port)]
+ with self._reserved_tcp_port() as dead_port:
+ endpoints = [
+ (self.qdb_plain.host, dead_port),
+ (self.qdb_plain.host, self.qdb_plain.http_server_port),
+ ]Also applies to: 384-386
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@test/system_test.py` around lines 165 - 168, The _unused_tcp_port() function
currently returns a port after closing the socket, causing a TOCTOU race;
instead bind and keep the socket reserved and return the bound socket (and/or
the port) so callers can use the reserved endpoint and close the socket only
when failover is set up; update any call sites (including the other occurrence
around lines 384-386) to accept and later close the returned socket once the
sender/receiver are using the port.
Introduce a new Client class wrapping the c-questdb-client questdb_db pool. Client.dataframe() emits via column_sender_chunk_* adapters for the v1 supported subset: fixed table name, NumPy int64/float64, NumPy datetime64[ns/us] field and designated timestamp columns, pandas categorical symbols, Arrow UTF-8 string fields, and large_string columns (cast to UTF-8 before publication). - Extract dataframe_plan_t in dataframe.pxi so row and columnar emitters share argument resolution and dtype planning. - Add column_sender FFI declarations and UnsupportedDataFrameShapeError carrying structured per-column rejection details before publication. - Schema-aware chunk planning with 8-row alignment when validity is present; mandatory column_sender_sync(ok) before return, with sync-and-retry for deferred flushes hitting the in-flight reserve. - Benchmark harness (test/benchmark_pandas_columnar.py) covering Layer 1 plan/populate, Layer 2 client-ACK with pool-reuse check, and Layer 3 against a real QuestDB. - Local ACK fixture (test/qwp_ws_ack_server.py) and a fixture-backed Layer 3 runner that spins up a QuestDB jar and verifies row counts. - Deterministic seed-controlled fuzz coverage (test/test_client_dataframe_fuzz.py) exercising the planner and emitter via the ACK fixture; reproduce with QDB_CLIENT_FUZZ_SEED / QDB_CLIENT_FUZZ_ITER_SEED. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Add value-edge and multi-chunk coverage to the seeded fuzz: - Numeric generators emit IEEE-754 specials (NaN, ±Inf, ±0.0, subnormals) and int64 boundary values (INT64_MIN/MAX) with 5% probability each; string generator emits empty strings with 5% probability to exercise zero-length offset slices. - test_multi_chunk_with_nulls: 100 003 rows of nullable categorical + int64 forces multi-chunk emission across the 8-row validity-bitmap alignment boundary. - test_high_cardinality_symbol_i16: 200 categories forces pandas to allocate i16 codes, exercising the column_sender_chunk_symbol_dict_i16 path that random fuzz rarely reaches. - test_wide_frame_multi_chunk: 12 field columns + 64 001 rows hits the planner's n_cols>8 → 64 000 cap branch. - test_sequential_client_lifecycle: 30 open/use/close cycles flush any close-then-reopen lifecycle leaks; verifies accepted_connections matches cycle count. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Update the Cython pxd and Client.dataframe() implementation to track the c-questdb-client submodule's rename of the borrowed pool handle from column_sender to qwpws_conn. No behaviour change at the Python level; Client.from_conf / .dataframe / .close / .reap_idle keep their signatures. Submodule bump: c-questdb-client jh_conn_pool_refactor (7740b7a) covers the C header, Rust FFI shim, and ABI doc. See plan-conn-pool-and-writers.md for the multi-step plan this is part of (Step 1 of 5). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Add the pxd declaration for the new Arrow appender. The arrow_c_data_interface module already provides the ArrowArray / ArrowSchema struct definitions, so the new entry point can take the same types pandas / pyarrow / polars already expose via _export_to_c. No call site wires this in yet — Client.dataframe still uses the per-type appenders. Wiring lands in Step 2c. Submodule bump: c-questdb-client jh_conn_pool_refactor (632c647) brings the new FFI surface. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
c-questdb-client jh_conn_pool_refactor (6c53ea7) adds LargeUtf8 (Arrow 'U') support to column_sender_chunk_append_arrow_column. No Python-side change yet — Step 2c wires it in via the planner. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Client.dataframe()'s varchar (string) and symbol (dictionary)
columns now dispatch through column_sender_chunk_append_arrow_column
instead of the per-type column_sender_chunk_column_varchar and
symbol_dict_i* calls. Numeric / timestamp columns stay on the
per-type path — they were already direct-write to wire (see
Open Q1) and the Arrow appender adds no per-row benefit there.
What this changes:
- The Cython side no longer manually slices Arrow offsets / bytes
/ dict-codes for chunked emission; the Arrow appender does
that slicing in Rust, with one dispatch step per format.
- Categorical i8/i16/i32 dispatch lives in Rust now, not Cython.
- The v1 validator accepts both col_source_str_utf8_arrow and
col_source_str_lrg_utf8_arrow — large_string flows through the
Arrow appender's `U` path natively. (The legacy row path still
needs the planner-shared large_string → utf8 cast, so it stays;
the Rust appender's `U` support is latent capability today but
used when the cast is removed in a future change.)
No public Python API change. 929 tests pass; 4 fixed seeds × 200
fuzz iters pass.
Submodule bump: c-questdb-client jh_conn_pool_refactor (0650c40)
adds the row_offset / row_count parameters to the Arrow appender.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
- Update plan-conn-pool-and-writers.md with Step 1, 2a, 2b, 2c completion (with submodule + parent SHAs). - Correct the Step 1 claim that it resolves the round-3 dirty-sender concern — it doesn't; the rename is cosmetic at that layer. Add an explicit caveat block describing the open follow-up. - Step 2 expansion: enumerate the three sub-steps and the known gaps surfaced during implementation (dead `lrg` validator arm, unreachable `U` format path, two-place 8-row alignment). - Step 3 reframed per Q1: "no perf win, only narrower-dtype / stride / endian convenience" + the open widening-policy decisions to make before implementing. Submodule bump: c-questdb-client jh_conn_pool_refactor (f35123d) applies the Arrow mirror-type and naming polish. 929 Python tests + 836 Rust unit tests still green. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Bump c-questdb-client to 90385af ("Hide pooled sender must-close
state"), which drops the sf_column_sender_must_close and
row_sender_must_close inspection functions. The pool return path now
closes a conn that latched terminal transport/protocol state instead of
exposing that state to callers.
- Remove the sf_column_sender_must_close cdef extern from
line_sender.pxd (row_sender_must_close was never bound).
- Rewrite _dataframe_columnar_force_drop_after_error to force-drop only
when a flush left uncommitted deferred frames without a clean sync;
the FFI return path now handles latched-terminal conns, so we return
normally in the no-flush case.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Update the Python QWP/WebSocket diagnostic bindings for the new FFI category/policy enums: add NotWritable and replace DropAndContinue/Halt with Retriable/RetriableOther/Terminal. Adjust tests for the NACK v2 schema-rejection behavior: schema mismatch now surfaces as a terminal diagnostic and retains SFA state, while later already published frames may still commit.
Bump the vendored c-questdb-client submodule to f0c5467 (the
jh_conn_pool_refactor tip) and update the Cython bindings to the renamed
column-sender C ABI:
- column_sender_chunk_column_varchar -> column_sender_chunk_column_str
- column_sender_chunk_designated_timestamp_{nanos,micros,millis,seconds}
-> column_sender_chunk_at_{nanos,micros,millis,seconds}
Pure symbol renames (identical signatures). Verified: the extension
builds and links against the new FFI, and questdb.ingress imports cleanly.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
On servers before the gap-halt fix (<9.4.4) a frame pipelined behind a rejected one commits or not depending on timing, so pinning the exact table content there flakes (passed in one 9.4.3 run, timed out in another). Assert only the deterministic pre-gap row on <9.4.4, and keep the exact gap-halt result (pipelined frame held) pinned on >=9.4.4.
The macOS leg built all CPython versions serially in one ~80 min job, dominated by the per-version test phase (pandas/pyarrow install + full suite ~13 min each). Split it into four parallel jobs by Python version (cp310+cp311, cp312+cp313, cp314, cp314t), mirroring the linux_arm64 split, to cut wall-clock. Versions are paired so each fresh job cold-builds the Rust lib once (shared CARGO_TARGET_DIR) then reuses it for its second version.
Client.dataframe now always uses the direct (non-store-and-forward) column sender, independent of sf_dir, matching the Rust flush_polars_dataframe tiering. The generic borrow had silently drifted onto the always-SFA main pool, changing blocking, crash and duplication semantics under the wrapper. - declare the direct_column_sender FFI surface in line_sender.pxd and switch the dataframe helpers (capsule, numpy, bench) to it; commit boundaries via direct_column_sender_commit - drop (never commit) pipelined frames when a dataframe call fails, so a failed call no longer lands a partial prefix - skip the trailing commit when an unknown-length Arrow stream yields no batches - retry an Arrow batch once after a commit when the 127-slot deferred window fills mid-frame (server-cap splits undercount the batch counter) - document the failover contract: whole-frame re-send unless an intermediate checkpoint landed, at-least-once duplication, plain QuestDBError on server rejection, request_timeout=0 caveat - reshape the sfa dataframe system tests into direct-contract tests and delete the now-unreachable SFA dataframe fuzz suites; bump c-questdb-client for the direct C API declarations
- a failed dataframe call now drops its pipelined frames instead of best-effort-committing them on the error path (and no longer blocks on a network ack inside exception handling) - skip the trailing commit for an unknown-length Arrow stream that yields no batches - retry an Arrow batch once after a commit when the deferred window fills mid-frame (server-cap splits undercount the batch counter) - document the failover contract precisely: checkpoint-bounded re-send, eager first-chunk duplication, plain QuestDBError on server rejection, request_timeout=0 caveat - QwpAckServer fault knobs: per-connection close plans, advertised X-QWP-Max-Batch-Size, defer-aware acks; a peer dropping its connection is no longer recorded as a protocol error - test_client_dataframe_failures: five deterministic fault tests (empty stream, pre/post-checkpoint disconnect, capacity drain, reconnect-budget exhaustion) plus a seeded fault-injection fuzz sweeping random disconnect timing x server caps x frame shapes across both dataframe routes - system tests: at-least-once bounce without dedup keys; a failed call leaves only the eager first batch - rework the late-flush-error test to the drop-not-commit contract - bump c-questdb-client for the direct capacity/batch-cap fixes
…ne_id, connection events
…r_info, connection events Client.dataframe `at` now accepts the full timestamp triple: - at=ServerTimestamp (explicit opt-in; frame ships no designated ts, the server stamps rows on arrival) on both the Arrow capsule route and the numpy chunk route; - a fixed TimestampNanos/datetime scalar shared by every row (encoded as a repeated constant; resubmission stays idempotent under DEDUP); - pre-epoch scalars rejected at the API boundary. Columnar ARRAY ingest: object-dtype columns of float64 ndarray cells are promoted to Arrow list<double> (nested lists for N-D) and ship as ARRAY(DOUBLE) through the existing Rust Arrow importer — closing the last row-path-only dtype. None cells land as NULL array rows (the row path rejects those). Also fixes the numpy 2.x descr.name crash in the object-column sniff error path. pandas 3.0 readiness: frame normalization now uses positional isetitem (the name-setitem chained-assignment heuristic misfires for Cython-held frames, spamming FutureWarnings); regression tests pin the str dtype (future.infer_string), coarse datetime64 units, and a warnings-as-errors ingest sweep. UnsupportedDataFrameShapeError now folds per-column reasons into str(exc) and, for designated-timestamp failures, points at the Client-side remedies (at='<column>' / at=ServerTimestamp). Client.server_info(): snapshot of the SERVER_INFO handshake (role, epoch, capabilities, wall clock, cluster/node/zone ids) via a pooled reader borrow; ServerRole/ServerInfo mirror the Rust reader types. Connection lifecycle events: connection_listener= on Client.from_conf and Sender.from_conf delivers ConnectionEvent (connected, disconnected, reconnected, failed_over, endpoint_attempt_failed, all_endpoints_unreachable, auth_failed) on a dedicated dispatcher thread with a bounded drop-oldest inbox, plus connection_events_delivered/dropped counters — Java listener parity. test_client_dataframe_failures: bound the reconnect-budget test's dial with connect_timeout so the budget governs on macOS too (a bound-not-listening loopback port blocks ~7.8s in the OS connect there instead of refusing).
… kill The test assumed the deferred-window commit near frame 128 always completed before the server's close at frame 140. When the close landed while that sync exchange was still in flight, committed_prefix was never set, so the client legally resent the whole frame on a fresh connection and succeeded instead of raising — 'QuestDBError not raised' on CI, ~1/60 locally. max_in_flight=32 moves the first commit checkpoint to ~frame 33, far from the close, so the prefix is deterministically committed by the time the connection dies. 240 fault iterations clean (previously 1 escape at the same scale).
…re delivery-unknown The prior deflake (max_in_flight=32) was a placebo: the direct column sender's in-flight window is the hardcoded protocol constant MAX_IN_FLIGHT=128, so the conf key never reached this path and CI kept failing. Actual root cause (c-questdb-client 94bab28): DirectColumnBackend::sync is implemented as a flush of an empty commit chunk, so it inherited flush_inner's entry ack-drain classification — try_drain_acks failures map to NotDelivered (in_doubt=false). When the trailing sync of a dataframe operation raced the peer reset, whichever side won decided the outcome: reset seen by the blocking ack wait → in_doubt=true → raise (the common case on fast machines); reset seen by the non-blocking entry drain → in_doubt=false → the retry gate legally resent the whole operation and succeeded — 'QuestDBError not raised' on slow CI runners, and in production a silent duplication of the prefix the split had already committed server-side. The Rust fix rescopes classification at the semantic boundary: with frames in flight at sync entry, every failure of that call is delivery-unknown (deny_retry_after_partial; downgrade-only), so the reset-arrival race no longer changes the outcome. Data-flush paths keep their per-input classification — retrying an uncommitted deferred prefix remains sanctioned. Test: restore the original conf and document the race both of its failure surfaces must classify as in_doubt. 400 fault-injection iterations clean across both routes (previously ~7 escapes at that scale), target test 20/20, full suites green.
QwpAckServer's close_plan used a bare close() with unread client frames in the receive buffer, which takes the RST path. macOS loopback occasionally loses that RST (netstat mid-hang: the server pcb destroyed, the client socket still ESTABLISHED with empty queues), so the client sat silent until its 30s request timeout — the exact 'took 30.0s' fuzz failures on CI. shutdown(SHUT_WR) sends a sequenced, retransmitted FIN the peer reliably observes; a short drain keeps the receive queue empty so the final close() doesn't reset first. 600 fault iterations clean (was ~7/600 hitting 30s).
The e961042 rule (any sync failure with frames in flight is in-doubt) was too broad: pending deferred frames die uncommitted with the connection, so with no commit boundary at risk the transparent whole-frame replay it blocked is provably safe. On Windows the peer reset reliably surfaces at the trailing sync's entry ack drain, so every transient disconnect raised in_doubt=True to the caller instead of failing over (release CI: 10054 out of test_transient_failure_before_publication_resends_whole_frame). c-questdb-client d2a358d replaces the in-flight test with a commit_since_sync flag: set when a mid-split internal sync commits a prefix (the one case whole-frame replay would duplicate), reset on every successful sync. Forced-timing probe flips from 200/200 spurious in_doubt to the correct split (entry-drain failure resends, post-commit ack-wait failure stays in doubt); the original escape probe holds at 400/400 clean.
The failover retry re-exports the same input for the whole-frame replay. That is sound for sliceable inputs (pa.Table, pandas, polars) but a one-shot RecordBatchReader is already drained by the failed attempt, so the retry sent zero rows and reported success — silent data loss. Track when the non-sliceable stream path consumes its input and surface the original transient error instead, with a message pointing at retrying with a fresh reader. Deterministic regression test included (server FIN lands before the trailing sync via a slow batch generator).
close_plan=[0] raced the server's close against the whole in-flight operation: on a loaded runner the FIN can land only after the client has already written its commit frame, and that surface is a legitimate delivery-unknown raise, not a resend (Linux CI: 'socket closed unexpectedly while reading frame header' out of the trailing sync). Split the test into two operations with an explicit close barrier between them: operation 1 completes, the server closes its connection and the test waits for that close to finish, so nothing of operation 2 can have been delivered and the transparent whole-frame replay is the only correct outcome. 30/30 locally; the racy single-op shape remains covered probabilistically by the fault fuzz, where both outcomes are accepted.
Java's supported schemas are http, https, tcp, tcps, ws, wss, udp; the qwpws->ws rename already aligned the WebSocket pair, so the UDP scheme follows: Protocol.QwpUdp becomes Protocol.Udp with conf scheme udp:: (hard rename, no alias, same policy as ws). Tracks c-questdb-client a878b89, which renames the Rust Protocol variant, the C enum value line_sender_protocol_qwpudp, the C++ protocol::qwpudp, the example binaries, and rejects udps:: with Java's exact 'TLS is not supported for UDP.' message; the Python conf parser surfaces that same message instead of a generic invalid-protocol ValueError. Internal qwp_udp naming (modules, fixtures, doc anchors) stays, mirroring how ws kept QwpWsConfig.
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
5 participants
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.
Overview
Adds the QuestDB Wire Protocol (QWP) as a new family of transports alongside the existing ILP/TCP and ILP/HTTP, plus a query-egress path that streams results back as Arrow. End users get a lower-latency fire-and-forget option (QWP/UDP), an acknowledged streaming option with delivery confirmation (QWP/WebSocket), and the ability to read query results into pandas/polars/pyarrow.
Ingestion
qwpudp::, default port 9007) — fire-and-forget, lowest-latency datagram ingestion. New config keysmax_datagram_sizeandmulticast_ttl;protocol_versiondoes not apply. No auth/TLS/transactions and no delivery guarantee — prefer ILP/HTTP when you need reliability or error feedback.qwpws::/qwpwss::) — acknowledged streaming ingestion with frame-sequence-number (FSN) tracking. NewSendermethods:flush_and_get_fsn,flush_and_keep_and_get_fsn,published_fsn,acked_fsn,await_acked_fsn,drive_once,poll_qwp_ws_error,qwp_ws_errors_dropped,close_drain. Server diagnostics surface via aqwp_ws_error_handlercallback or are polled asQwpWsError; a terminal (halt) rejection raisesIngressServerRejectionError.Query egress
ClientwithClient.query()returning a single-useQueryResultthat streams rows as Arrow record batches over the QWP/WebSocket read endpoint. Consume viato_arrow/to_pandas/to_polars/iter_arrow/iter_pandasor the Arrow C-stream PyCapsule protocol (__arrow_c_stream__) — the last two work without pyarrow installed.Clientis a context manager exposingclose()andreap_idle()for pooled-connection lifecycle.Client.dataframe()ingests pandas / polars / pyarrow / any Arrow C Data Interface object over QWP/WebSocket (columnar path), with aschema_overrideskeyword to reclassify columns assymbol/ipv4/char/geohash.Buffer factories
Buffer.ilp()andBuffer.qwp()construct protocol-specific buffers;Sender.new_buffer()returns the buffer matching the sender's protocol. DirectBuffer(...)construction is deprecated (still works, emitsDeprecationWarning) in favour of the factories.Public API surface
All reflected in
CHANGELOG.rstanddocs/:Client,QueryResult,IngressServerRejectionError,UnsupportedDataFrameShapeError,QwpWsError,QwpWsErrorCategory,QwpWsErrorPolicy,QwpWsProgress.ProtocolmembersQwpUdp,QwpWs,QwpWss.IngressErrorCodemembersServerRejection,ArrowUnsupportedColumnKind,ArrowIngest,Cancelled; newIngressError.qwp_ws_errorproperty.tls_roots_password,retry_max_backoff(conf keyretry_max_backoff_millis),qwp_ws_progress,qwp_ws_error_handler,max_datagram_size,multicast_ttl,schema_overrides,max_rows_per_batch.Dependency & build changes
c-questdb-clientsubmodule34905ab → 5010ee9(6.1.0-109) to pull in the QWP transports, thecolumn_sendercolumnar API, and theline_readerquery API. Enables theinsecure-skip-verify,sync-reader-ws,arrowcargo features and theQUESTDB_CLIENT_HAS_ARROW/QUESTDB_CLIENT_ENABLE_ARROWmacros.schema_overrides, theto_*/iter_*helpers). The__arrow_c_stream__/to_polarspaths work without it.numpy>=1.21.0is now a hard dependency; thedataframeextra pinspandas>=1.3.5,pyarrow>=10.0.1.