Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions cpp/CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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<vector<uint8_t>>`. 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.
Expand Down
12 changes: 12 additions & 0 deletions cpp/NEXT_CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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<std::vector<std::uint8_t>>` 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
Expand All @@ -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.
22 changes: 22 additions & 0 deletions cpp/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -229,6 +229,27 @@ stream.ingest_proto_records(batch);
stream.flush();
```

If your encoded records already live somewhere other than a
`std::vector<std::vector<std::uint8_t>>` — 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<zerobus::ProtoRecordView> 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
Expand Down Expand Up @@ -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`,
Expand Down
31 changes: 31 additions & 0 deletions cpp/examples/proto/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
```

Expand Down Expand Up @@ -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<std::uint8_t>` 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<zerobus::ProtoRecordView> 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
Expand Down
45 changes: 44 additions & 1 deletion cpp/examples/proto/batch.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -26,6 +29,7 @@
// TIMESTAMP)

#include <chrono>
#include <cstddef>
#include <cstdint>
#include <cstdlib>
#include <iostream>
Expand Down Expand Up @@ -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<std::string> 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<std::uint8_t> arena;
std::vector<Span> spans;
for (const std::string& order : more_orders) {
const std::vector<std::uint8_t> 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<zerobus::ProtoRecordView> 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();
Expand Down
11 changes: 11 additions & 0 deletions cpp/include/zerobus/record.hpp
Original file line number Diff line number Diff line change
@@ -1,13 +1,24 @@
#ifndef ZEROBUS_RECORD_HPP
#define ZEROBUS_RECORD_HPP

#include <cstddef>
#include <cstdint>
#include <string>
#include <utility>
#include <vector>

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()`.
///
Expand Down
16 changes: 16 additions & 0 deletions cpp/include/zerobus/stream.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,22 @@ class Stream {
std::int64_t ingest_proto_records(
const std::vector<std::vector<std::uint8_t>>& 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.
Expand Down
89 changes: 89 additions & 0 deletions cpp/src/detail/proto_batch.hpp
Original file line number Diff line number Diff line change
@@ -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 <cstddef>
#include <cstdint>
#include <string>
#include <vector>

#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<std::uint8_t>& 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<const std::uint8_t*> ptrs;
std::vector<std::uintptr_t> lens;
};

inline ProtoBatchView make_proto_batch(
const std::vector<std::vector<std::uint8_t>>& 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<vector<uint8_t>> 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
Loading
Loading