diff --git a/cpp/CLAUDE.md b/cpp/CLAUDE.md index 827d5927..c8b9da54 100644 --- a/cpp/CLAUDE.md +++ b/cpp/CLAUDE.md @@ -177,6 +177,13 @@ Public API is everything under `include/zerobus/`: - Every FFI crossing serializes data; prefer the batch APIs (`ingest_proto_records`, `ingest_json_records`) over per-record calls in hot paths. +- Neither batch API copies payloads: `detail/proto_batch.hpp` builds the FFI's + pointer/length arrays from pointers into the caller's buffers, and + `tests/proto_batch_test.cpp` asserts that by pointer identity (a copying + regression would still behave correctly). `ingest_proto_records` also takes + `const ProtoRecordView*` + count, for records held outside a + `vector>`. JSON has no such overload: its FFI needs + NUL-terminated `const char*`, so `std::string` is already the zero-copy shape. - Ingestion is asynchronous: `ingest_*` queues and returns. Never wait per record (`wait_for_offset`/`flush` in the loop); flush once at the end or flush periodically. Examples and doc comments must follow this pattern. diff --git a/cpp/NEXT_CHANGELOG.md b/cpp/NEXT_CHANGELOG.md index 7927f801..a0835396 100644 --- a/cpp/NEXT_CHANGELOG.md +++ b/cpp/NEXT_CHANGELOG.md @@ -4,6 +4,14 @@ ### New Features and Improvements +- Added a borrowing overload of `Stream::ingest_proto_records()`, taking + `const ProtoRecordView*` and a count. Callers whose encoded records already + live elsewhere (an arena, a ring buffer, their own record type) no longer have + to copy every payload into a `std::vector>` just to + hand the batch over. `zerobus::ProtoRecordView` is a non-owning `{data, size}` + pair whose bytes must stay valid until the call returns. Existing calls are + unaffected. + ### Bug Fixes - Fixed a use-after-free in which a custom `HeadersProvider` could be destroyed @@ -25,3 +33,7 @@ ### Deprecations ### API Changes + +- New: `zerobus::ProtoRecordView` (in `zerobus/record.hpp`) and + `Stream::ingest_proto_records(const ProtoRecordView*, std::size_t)`. Additive + only — no existing signature changed. diff --git a/cpp/README.md b/cpp/README.md index bc8723fd..734ebfe6 100644 --- a/cpp/README.md +++ b/cpp/README.md @@ -229,6 +229,27 @@ stream.ingest_proto_records(batch); stream.flush(); ``` +If your encoded records already live somewhere other than a +`std::vector>` — one contiguous arena, a ring buffer, +your own record type — describe them with `ProtoRecordView` instead of copying +each payload into that container first: + +```cpp +// `arena` holds the encoded records back to back; `spans` records where each +// one starts and how long it is. +std::vector views; +for (const auto& span : spans) { + views.push_back({arena.data() + span.offset, span.size}); +} +stream.ingest_proto_records(views.data(), views.size()); +stream.flush(); +``` + +A `ProtoRecordView` borrows: the bytes it points at must stay valid until the +ingest call returns (the core copies them before it does). Build the views only +once the buffer they point into has stopped growing — a `push_back` that +reallocates invalidates every pointer taken from it earlier. + ### Arrow Flight ingestion (Beta) Stream Arrow record batches instead of proto/JSON records. Create the stream @@ -328,6 +349,7 @@ canonical version): | `zerobus::StreamOptions` / `zerobus::ArrowStreamOptions` | Stream configuration | | `zerobus::ZerobusException` | Thrown on any failure; `is_retryable()` | | `zerobus::UnackedRecord` | An unacknowledged record recovered from a failed stream | +| `zerobus::ProtoRecordView` | Non-owning `{data, size}` view of a proto record, for batch ingestion without copies | Key `Stream` methods: `ingest_proto_record`, `ingest_json_record`, `ingest_proto_records`, `ingest_json_records`, `wait_for_offset`, `flush`, diff --git a/cpp/examples/proto/README.md b/cpp/examples/proto/README.md index e2024f65..e24140dc 100644 --- a/cpp/examples/proto/README.md +++ b/cpp/examples/proto/README.md @@ -13,6 +13,7 @@ ingestion into Databricks Delta tables using the Zerobus C++ SDK. - [Batch Example](#batch-example) - [Running the Example](#running-the-example-1) - [Code Highlights](#code-highlights-1) + - [Batching Records You Already Hold](#batching-records-you-already-hold) - [Adapting for Your Custom Table](#adapting-for-your-custom-table) ## Overview @@ -165,6 +166,7 @@ zerobus::Stream stream = ``` Batch of 3 records queued; batch offset ID: 0 Batch acknowledged at offset ID: 0 +Arena batch of 2 records queued; batch offset ID: 1 Stream closed successfully. ``` @@ -195,6 +197,35 @@ if (batch_offset >= 0) { In a hot path you would queue **many** batches and `flush()` once, rather than waiting after each batch. +### Batching Records You Already Hold + +`encode_json()` hands back one `std::vector` per record, so the +batch above is a natural vector-of-vectors. If your encoded records live +somewhere else — the example packs a second batch into one contiguous arena — +use `ProtoRecordView` instead of copying each payload into that container just +to pass it: + +```cpp +std::vector views; +for (const Span& span : spans) { // spans index into `arena` + views.push_back({arena.data() + span.offset, span.size}); +} + +const std::int64_t offset = + stream.ingest_proto_records(views.data(), views.size()); +``` + +A view borrows, so two rules apply: +- **The bytes must outlive the call.** The core copies them before + `ingest_proto_records()` returns, and nothing holds the pointer afterwards. +- **Take the pointers last.** Build the views only once the buffer they point + into has stopped growing; an `insert`/`push_back` that reallocates invalidates + every pointer taken from it earlier. + +`{nullptr, 0}` is a valid empty record. A null pointer with a non-zero size is +rejected with a `ZerobusException` naming the record's index, rather than +dereferenced inside the core. + ## Adapting for Your Custom Table Because the schema is fetched from Unity Catalog at runtime, adapting to your own diff --git a/cpp/examples/proto/batch.cpp b/cpp/examples/proto/batch.cpp index 07b61f22..cfb12403 100644 --- a/cpp/examples/proto/batch.cpp +++ b/cpp/examples/proto/batch.cpp @@ -11,6 +11,9 @@ // unit. The call returns a single logical offset assigned to the whole batch; // waiting on that one offset confirms the entire batch. // +// Two ways to hand a batch over are shown: a vector of encoded records, and +// ProtoRecordViews borrowing records that already live in a caller-owned arena. +// // Configuration — every connection setting, plus the Unity Catalog table // metadata JSON, is read from the environment. Export these before running (see // ../README.md for what each one is and the full copy-pasteable block, @@ -26,6 +29,7 @@ // TIMESTAMP) #include +#include #include #include #include @@ -122,7 +126,46 @@ int main() { std::cout << "Batch acknowledged at offset ID: " << batch_offset << "\n"; } - // 6. flush() drains anything still pending, then close at a controlled + // 6. A second batch, for records the SDK does not own. + // + // encode_json() returns a vector per record, so the batch above was + // already a natural vector-of-vectors. When your records live elsewhere + // — here, packed into one arena — describe them with ProtoRecordView + // instead of copying each payload into that container to pass it. + const std::vector more_orders = { + make_order_json(4, "Dan Brown", "Laptop Stand", 1, 34.50, "pending", + now), + make_order_json(5, "Erin Page", "HD Webcam", 2, 59.99, "pending", now), + }; + + // Where each encoded record starts in the arena, and how long it is. + struct Span { + std::size_t offset; + std::size_t size; + }; + std::vector arena; + std::vector spans; + for (const std::string& order : more_orders) { + const std::vector encoded = schema.encode_json(order); + spans.push_back({arena.size(), encoded.size()}); + arena.insert(arena.end(), encoded.begin(), encoded.end()); + } + + // Take the pointers only now the arena has stopped growing: a reallocating + // insert invalidates any taken earlier. + std::vector views; + views.reserve(spans.size()); + for (const Span& span : spans) { + views.push_back({arena.data() + span.offset, span.size}); + } + + // arena must outlive this call — the views only borrow it. + const std::int64_t arena_offset = + stream.ingest_proto_records(views.data(), views.size()); + std::cout << "Arena batch of " << views.size() + << " records queued; batch offset ID: " << arena_offset << "\n"; + + // 7. flush() drains anything still pending, then close at a controlled // point. stream.flush(); stream.close(); diff --git a/cpp/include/zerobus/record.hpp b/cpp/include/zerobus/record.hpp index f72e2992..ef81f750 100644 --- a/cpp/include/zerobus/record.hpp +++ b/cpp/include/zerobus/record.hpp @@ -1,6 +1,7 @@ #ifndef ZEROBUS_RECORD_HPP #define ZEROBUS_RECORD_HPP +#include #include #include #include @@ -8,6 +9,16 @@ namespace zerobus { +/// A non-owning view of one protobuf-encoded record, for the borrowing +/// `Stream::ingest_proto_records()` overload. +/// +/// The bytes must stay valid until that call returns; the core copies them +/// before it does. `{nullptr, 0}` is a valid empty record. +struct ProtoRecordView { + const std::uint8_t* data = nullptr; + std::size_t size = 0; +}; + /// A record recovered from a stream that was closed or failed before all /// records were acknowledged. Returned by `Stream::get_unacked_records()`. /// diff --git a/cpp/include/zerobus/stream.hpp b/cpp/include/zerobus/stream.hpp index d5d565a9..c9a5dbdc 100644 --- a/cpp/include/zerobus/stream.hpp +++ b/cpp/include/zerobus/stream.hpp @@ -78,6 +78,22 @@ class Stream { std::int64_t ingest_proto_records( const std::vector>& records); + /// @overload + /// Ingest a batch of borrowed protobuf records, skipping the copy into a + /// vector of vectors when the encoded records already live elsewhere (an + /// arena, a ring buffer, your own record type). + /// + /// @param records Pointer to @p num_records views, each borrowing bytes that + /// must stay valid until this call returns. + /// @param num_records Number of views in @p records. + /// @return The single logical offset assigned to the whole batch, or -1 if + /// @p num_records is 0 (a no-op). + /// @throws ZerobusException if @p records is null with a non-zero + /// @p num_records, if a view has a null pointer with a non-zero size, + /// or if the stream is closed or ingestion fails. + std::int64_t ingest_proto_records(const ProtoRecordView* records, + std::size_t num_records); + /// Ingest a batch of JSON records, blocking until they are queued. /// /// @param records The records, each a UTF-8 JSON string. diff --git a/cpp/src/detail/proto_batch.hpp b/cpp/src/detail/proto_batch.hpp new file mode 100644 index 00000000..faba6454 --- /dev/null +++ b/cpp/src/detail/proto_batch.hpp @@ -0,0 +1,89 @@ +#ifndef ZEROBUS_DETAIL_PROTO_BATCH_HPP +#define ZEROBUS_DETAIL_PROTO_BATCH_HPP + +// Builds the parallel pointer/length arrays zerobus_stream_ingest_proto_records +// expects, without copying payloads. +// +// Kept out of stream.cpp so tests can reach it: a Stream needs a live server, +// but the invariants here — aliasing the caller's bytes, never handing the FFI +// a null payload — are testable alone (tests/proto_batch_test.cpp). Free of +// zerobus.h; the JSON equivalent stays in stream.cpp, where checked_c_str is. + +#include +#include +#include +#include + +#include "zerobus/error.hpp" +#include "zerobus/record.hpp" + +namespace zerobus { +namespace detail { + +// An empty payload still crosses the FFI as a non-null pointer with length 0, +// rather than nullptr or a dangling data() result. +inline constexpr std::uint8_t kEmptyPayloadSentinel = 0; + +inline const std::uint8_t* ptr_or_sentinel( + const std::vector& bytes) { + return bytes.empty() ? &kEmptyPayloadSentinel : bytes.data(); +} + +// Same sentinel for the raw form, so {nullptr, 0} is a valid empty record. +inline const std::uint8_t* ptr_or_sentinel(const std::uint8_t* data, + std::size_t len) { + return len == 0 ? &kEmptyPayloadSentinel : data; +} + +// The pointers alias the caller's record bytes, so a ProtoBatchView must not +// outlive the records it was built from. +struct ProtoBatchView { + std::vector ptrs; + std::vector lens; +}; + +inline ProtoBatchView make_proto_batch( + const std::vector>& records) { + ProtoBatchView v; + v.ptrs.reserve(records.size()); + v.lens.reserve(records.size()); + for (const auto& r : records) { + v.ptrs.push_back(ptr_or_sentinel(r)); + v.lens.push_back(r.size()); + } + return v; +} + +// Borrowing form. Its own loop rather than materialising the views into a +// vector> and delegating above: those copies are the cost it +// exists to avoid. +inline ProtoBatchView make_proto_batch(const ProtoRecordView* records, + std::size_t num_records) { + if (records == nullptr && num_records != 0) { + throw ZerobusException( + "ingest_proto_records called with a null record array and a non-zero " + "record count", + false); + } + ProtoBatchView v; + v.ptrs.reserve(num_records); + v.lens.reserve(num_records); + for (std::size_t i = 0; i < num_records; ++i) { + // The core would dereference a sized null payload. Name the index so the + // offending record is identifiable in a large batch. + if (records[i].data == nullptr && records[i].size != 0) { + throw ZerobusException( + "proto record at index " + std::to_string(i) + + " has a null data pointer with a non-zero size", + false); + } + v.ptrs.push_back(ptr_or_sentinel(records[i].data, records[i].size)); + v.lens.push_back(records[i].size); + } + return v; +} + +} // namespace detail +} // namespace zerobus + +#endif // ZEROBUS_DETAIL_PROTO_BATCH_HPP diff --git a/cpp/src/stream.cpp b/cpp/src/stream.cpp index 766368bb..2dc838b9 100644 --- a/cpp/src/stream.cpp +++ b/cpp/src/stream.cpp @@ -1,8 +1,8 @@ // Implementation of Stream (declared in zerobus/stream.hpp). // // A thin forwarding layer over the zerobus_stream_* C FFI entry points. The -// file-local helpers build the small parallel pointer/length arrays the batch -// entry points expect, and every fallible call routes its CResult through +// proto batch entry point's pointer/length arrays are built by +// detail/proto_batch.hpp, and every fallible call routes its CResult through // detail::ResultGuard. The destructor and move-assignment close best-effort // (swallowing errors), whereas close() surfaces them. Public API documentation // lives on the header; comments here cover only implementation details. @@ -12,6 +12,7 @@ #include #include "detail/ffi_util.hpp" +#include "detail/proto_batch.hpp" namespace zerobus { @@ -47,39 +48,6 @@ std::int64_t checked_offset(std::int64_t offset) { return offset; } -// An empty payload still needs a valid, non-null pointer to pass across the FFI -// (paired with length 0); hand out the address of a static sentinel byte rather -// than nullptr or a dangling data() result. -const std::uint8_t* ptr_or_sentinel(const std::vector& bytes) { - static const std::uint8_t kEmptyPayloadSentinel = 0; - return bytes.empty() ? &kEmptyPayloadSentinel : bytes.data(); -} - -// Same sentinel for the raw (pointer, length) form, so {nullptr, 0} is a valid -// empty record instead of a null-pointer error. -const std::uint8_t* ptr_or_sentinel(const std::uint8_t* data, std::size_t len) { - static const std::uint8_t kEmptyPayloadSentinel = 0; - return len == 0 ? &kEmptyPayloadSentinel : data; -} - -// Build the parallel pointer/length arrays the batch FFI entry points expect. -struct ProtoBatchView { - std::vector ptrs; - std::vector lens; -}; - -ProtoBatchView make_proto_batch( - const std::vector>& records) { - ProtoBatchView v; - v.ptrs.reserve(records.size()); - v.lens.reserve(records.size()); - for (const auto& r : records) { - v.ptrs.push_back(ptr_or_sentinel(r)); - v.lens.push_back(r.size()); - } - return v; -} - // JSON records cross the FFI as an array of NUL-terminated C strings, so unlike // the proto path there is no parallel length array — only the pointers. struct JsonBatchView { @@ -151,7 +119,7 @@ std::int64_t Stream::ingest_proto_record(const std::uint8_t* data, ensure_open(handle_); detail::ResultGuard guard; std::int64_t offset = zerobus_stream_ingest_proto_record( - handle_, ptr_or_sentinel(data, len), len, guard.ptr()); + handle_, detail::ptr_or_sentinel(data, len), len, guard.ptr()); guard.throw_if_error(); return checked_offset(offset); } @@ -160,7 +128,7 @@ std::int64_t Stream::ingest_proto_record(const std::uint8_t* data, // sentinel so an empty record still passes a non-null pointer. std::int64_t Stream::ingest_proto_record( const std::vector& data) { - return ingest_proto_record(ptr_or_sentinel(data), data.size()); + return ingest_proto_record(detail::ptr_or_sentinel(data), data.size()); } std::int64_t Stream::ingest_json_record(const std::string& json) { @@ -182,7 +150,24 @@ std::int64_t Stream::ingest_proto_records( if (records.empty()) { return -1; } - ProtoBatchView v = make_proto_batch(records); + detail::ProtoBatchView v = detail::make_proto_batch(records); + detail::ResultGuard guard; + std::int64_t offset = zerobus_stream_ingest_proto_records( + handle_, v.ptrs.data(), v.lens.data(), v.ptrs.size(), guard.ptr()); + guard.throw_if_error(); + return checked_offset(offset); +} + +// Borrowing overload: v points into the caller's bytes instead of copying them. +std::int64_t Stream::ingest_proto_records(const ProtoRecordView* records, + std::size_t num_records) { + ensure_open(handle_); + // No-op returning -1, as in the vector overload. Checked before the null + // guard, so a {nullptr, 0} batch is a no-op too rather than an error. + if (num_records == 0) { + return -1; + } + detail::ProtoBatchView v = detail::make_proto_batch(records, num_records); detail::ResultGuard guard; std::int64_t offset = zerobus_stream_ingest_proto_records( handle_, v.ptrs.data(), v.lens.data(), v.ptrs.size(), guard.ptr()); diff --git a/cpp/tests/CMakeLists.txt b/cpp/tests/CMakeLists.txt index c819737e..111816a0 100644 --- a/cpp/tests/CMakeLists.txt +++ b/cpp/tests/CMakeLists.txt @@ -21,6 +21,7 @@ zerobus_add_test(config_defaults_test) zerobus_add_test(arrow_config_defaults_test) zerobus_add_test(arrow_config_convert_test) zerobus_add_test(embedded_nul_test) +zerobus_add_test(proto_batch_test) zerobus_add_test(ack_callback_test) zerobus_add_test(headers_provider_test) zerobus_add_test(error_test) diff --git a/cpp/tests/proto_batch_test.cpp b/cpp/tests/proto_batch_test.cpp new file mode 100644 index 00000000..4e9f344c --- /dev/null +++ b/cpp/tests/proto_batch_test.cpp @@ -0,0 +1,160 @@ +// Exercises detail::make_proto_batch, which both ingest_proto_records overloads +// route through. +// +// The ingest call itself needs a live server, but the invariants here do not: +// the arrays must ALIAS the caller's bytes (asserted by pointer identity — a +// regression to copying would still behave correctly, so nothing else catches +// it), the FFI must never get a null payload pointer, and the two overloads +// must agree. + +#include "detail/proto_batch.hpp" + +#include +#include +#include +#include +#include + +#include "zerobus/error.hpp" +#include "zerobus/record.hpp" + +namespace { + +using zerobus::ProtoRecordView; +using zerobus::ZerobusException; +using zerobus::detail::make_proto_batch; +using zerobus::detail::ProtoBatchView; + +int g_failures = 0; + +void fail(const std::string& msg) { + std::fprintf(stderr, "FAIL: %s\n", msg.c_str()); + ++g_failures; +} + +void expect(bool condition, const std::string& msg) { + if (!condition) { + fail(msg); + } +} + +// Includes an empty record: the case where a naive data() hands the FFI null. +std::vector> sample_records() { + return {{1, 2, 3}, {42}, {}}; +} + +std::vector views_of( + const std::vector>& records) { + std::vector views; + views.reserve(records.size()); + for (const auto& r : records) { + views.push_back({r.data(), r.size()}); + } + return views; +} + +// Lengths mirror the inputs; non-empty records are aliased, not copied. +void check_matches_records( + const ProtoBatchView& v, + const std::vector>& records, + const std::string& who) { + expect(v.ptrs.size() == records.size(), who + ": wrong pointer count"); + expect(v.lens.size() == records.size(), who + ": wrong length count"); + if (v.ptrs.size() != records.size() || v.lens.size() != records.size()) { + return; // Indexing below would be out of bounds. + } + for (std::size_t i = 0; i < records.size(); ++i) { + expect(v.lens[i] == records[i].size(), + who + ": length mismatch at index " + std::to_string(i)); + expect( + v.ptrs[i] != nullptr, + who + ": null pointer handed to the FFI at index " + std::to_string(i)); + if (!records[i].empty()) { + expect(v.ptrs[i] == records[i].data(), + who + ": record " + std::to_string(i) + + " was copied instead of aliased"); + } + } +} + +} // namespace + +int main() { + const std::vector> records = sample_records(); + + const ProtoBatchView from_vector = make_proto_batch(records); + check_matches_records(from_vector, records, "vector overload"); + + const std::vector views = views_of(records); + const ProtoBatchView from_views = + make_proto_batch(views.data(), views.size()); + check_matches_records(from_views, records, "borrowing overload"); + + // Interchangeable, down to the sentinel used for the empty record. + expect(from_vector.ptrs == from_views.ptrs, + "overloads disagree on the pointer array"); + expect(from_vector.lens == from_views.lens, + "overloads disagree on the length array"); + + // A default-constructed view is {nullptr, 0}: a valid empty record, which + // still reaches the FFI as a non-null pointer. + { + const ProtoRecordView empty{}; + const ProtoBatchView v = make_proto_batch(&empty, 1); + expect(v.ptrs.size() == 1 && v.lens.size() == 1, + "{nullptr, 0} did not produce a single-record view"); + if (v.ptrs.size() == 1 && v.lens.size() == 1) { + expect(v.ptrs[0] != nullptr, + "{nullptr, 0} passed a null pointer to the FFI"); + expect(v.lens[0] == 0, "{nullptr, 0} did not produce a zero length"); + } + } + + // A zero-count batch must not touch the array at all, even a null one. + { + bool threw = false; + try { + const ProtoBatchView v = make_proto_batch(nullptr, 0); + expect(v.ptrs.empty() && v.lens.empty(), + "zero-count batch produced a non-empty view"); + } catch (const ZerobusException&) { + threw = true; + } + expect(!threw, "zero-count batch with a null array was rejected"); + } + + // A null array with records to read must be reported, not dereferenced. + { + bool threw = false; + try { + make_proto_batch(nullptr, 3); + } catch (const ZerobusException&) { + threw = true; + } + expect(threw, "null record array with a non-zero count was NOT rejected"); + } + + // Likewise a null payload claiming a non-zero size: the core would read it. + { + const std::vector bad = { + {records[0].data(), records[0].size()}, + {nullptr, 7}, + }; + bool threw = false; + try { + make_proto_batch(bad.data(), bad.size()); + } catch (const ZerobusException& e) { + threw = true; + expect(std::string(e.what()).find("index 1") != std::string::npos, + "exception message did not name the offending record index"); + } + expect(threw, "null payload with a non-zero size was NOT rejected"); + } + + if (g_failures != 0) { + std::fprintf(stderr, "%d check(s) failed.\n", g_failures); + return 1; + } + std::printf("proto batch adaptation aliases caller bytes and guards nulls\n"); + return 0; +}