[C++] Add borrowing ingest_proto_records overload

Signed-off-by: Zlata Stefanovic <zlata.stefanovic@databricks.com>

Author: zlata-stefanovic-db

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<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.

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<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
@@ -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.

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<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
@@ -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`,

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<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

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 <chrono>
+#include <cstddef>
 #include <cstdint>
 #include <cstdlib>
 #include <iostream>
@@ -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();

cpp/include/zerobus/record.hpp

@@ -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()`.
 ///

cpp/include/zerobus/stream.hpp

@@ -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.

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 <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