diff --git a/cpp/NEXT_CHANGELOG.md b/cpp/NEXT_CHANGELOG.md index 5d6083f0..f14d946e 100644 --- a/cpp/NEXT_CHANGELOG.md +++ b/cpp/NEXT_CHANGELOG.md @@ -27,6 +27,12 @@ - Documented running the tests in `README.md`: the sanitizer runs (`make test SANITIZE=address` / `thread`) and the env-var-gated `integration_test` (which variables it needs and that it skips without them). +- Added runnable examples under `examples/` covering all three record formats — + JSON, protobuf (dynamic schema built at runtime from Unity Catalog metadata via + `ProtoSchema::from_uc_json`, no `protoc` required), and Arrow Flight (Beta) — + each with a single-record and a batch variant. They build with the SDK via + `ZEROBUS_BUILD_EXAMPLES` (the Arrow example is skipped when Apache Arrow C++ is + not installed). Includes a top-level `examples/README.md` and per-format guides. ### Internal Changes diff --git a/cpp/README.md b/cpp/README.md index 5e563262..55ae4060 100644 --- a/cpp/README.md +++ b/cpp/README.md @@ -38,8 +38,8 @@ make build # configure + build the SDK and tests make test # build + run the test suite ``` -(`make build` also builds an `examples/` directory once one is added; none -ships yet.) +(`make build` also builds the runnable examples under `examples/`; see +[Examples](#examples).) Or drive CMake directly: @@ -365,10 +365,15 @@ the FFI defaults. ## Examples -Runnable examples will live under `examples/`, covering the three ingestion -paths (JSON, dynamic proto, and static proto). Until they land, the -[Quickstart](#quickstart) above demonstrates the JSON and dynamic-proto paths -end to end. +Runnable examples live under [`examples/`](examples/README.md), covering all +three record formats — JSON, protobuf (dynamic schema from Unity Catalog), and +Arrow Flight (Beta) — each with a single-record and a batch variant. Every +example reads its OAuth secrets (`DATABRICKS_CLIENT_ID` / +`DATABRICKS_CLIENT_SECRET`) from the environment and keeps its non-secret +connection info as placeholder constants at the top of the source file. They +build as part of the top-level CMake build (`ZEROBUS_BUILD_EXAMPLES`, on by +default) into `build/examples/`. See the [examples README](examples/README.md) +for setup and per-format guides. ## A note on credentials diff --git a/cpp/examples/CMakeLists.txt b/cpp/examples/CMakeLists.txt new file mode 100644 index 00000000..b2576bf8 --- /dev/null +++ b/cpp/examples/CMakeLists.txt @@ -0,0 +1,61 @@ +# Example binaries for the Zerobus C++ SDK. +# +# Built when ZEROBUS_BUILD_EXAMPLES is on (the default for a top-level build). +# Each example links the zerobus::zerobus target, which transitively pulls in +# the Rust C FFI archive and the platform system libraries it needs. +# +# Every example reads its two OAuth secrets (DATABRICKS_CLIENT_ID / +# DATABRICKS_CLIENT_SECRET) from the environment; the proto examples additionally +# read the Unity Catalog table metadata JSON (ZEROBUS_UC_TABLE_JSON). All other +# connection info is placeholder constants at the top of each source file. See +# each file's header for the variables it expects and a sample invocation. +# +# The JSON and proto examples need no protobuf toolchain: the proto examples use +# dynamic ProtoSchema::from_uc_json(), which builds the descriptor at runtime. +# Only the Arrow example has an external dependency (Apache Arrow C++), and it is +# skipped gracefully when Arrow is not installed. + +# --------------------------------------------------------------------------- +# JSON examples +# --------------------------------------------------------------------------- +add_executable(json_single json/single.cpp) +target_link_libraries(json_single PRIVATE zerobus::zerobus) + +add_executable(json_batch json/batch.cpp) +target_link_libraries(json_batch PRIVATE zerobus::zerobus) + +# --------------------------------------------------------------------------- +# Protocol Buffers examples (dynamic schema — no protoc required) +# --------------------------------------------------------------------------- +add_executable(proto_single proto/single.cpp) +target_link_libraries(proto_single PRIVATE zerobus::zerobus) + +add_executable(proto_batch proto/batch.cpp) +target_link_libraries(proto_batch PRIVATE zerobus::zerobus) + +if(DEFINED ZEROBUS_FFI_BUILD_TARGET) + add_dependencies(json_single ${ZEROBUS_FFI_BUILD_TARGET}) + add_dependencies(json_batch ${ZEROBUS_FFI_BUILD_TARGET}) + add_dependencies(proto_single ${ZEROBUS_FFI_BUILD_TARGET}) + add_dependencies(proto_batch ${ZEROBUS_FFI_BUILD_TARGET}) +endif() + +# --------------------------------------------------------------------------- +# Arrow Flight example (arrow_ingest) — Beta +# --------------------------------------------------------------------------- +# Uses the Apache Arrow C++ library to build RecordBatches and serialize them to +# Arrow IPC bytes for ArrowStream::ingest_batch. Skipped if Arrow is not +# installed — a machine without libarrow still gets every other example. +find_package(Arrow QUIET) +if(Arrow_FOUND) + add_executable(arrow_ingest arrow/arrow_ingest.cpp) + target_link_libraries(arrow_ingest PRIVATE zerobus::zerobus Arrow::arrow_shared) + if(DEFINED ZEROBUS_FFI_BUILD_TARGET) + add_dependencies(arrow_ingest ${ZEROBUS_FFI_BUILD_TARGET}) + endif() +else() + message(STATUS + "Zerobus examples: Apache Arrow C++ not found; skipping arrow_ingest " + "example (install libarrow-dev to enable). The JSON and proto examples " + "still build.") +endif() diff --git a/cpp/examples/README.md b/cpp/examples/README.md new file mode 100644 index 00000000..7ae74c11 --- /dev/null +++ b/cpp/examples/README.md @@ -0,0 +1,317 @@ +# Zerobus C++ SDK Examples + +This directory contains examples demonstrating how to use the Zerobus C++ SDK to +ingest data into Databricks Delta tables. + +## Table of Contents + +- [Overview](#overview) +- [JSON Examples](json/README.md) +- [Protocol Buffers Examples](proto/README.md) +- [Arrow Flight Examples](arrow/README.md) (Beta) +- [Prerequisites](#prerequisites) + - [Create a Databricks Table](#1-create-a-databricks-table) + - [Set Up OAuth Service Principal](#2-set-up-oauth-service-principal) + - [Configure Credentials](#3-configure-credentials) +- [Building the Examples](#building-the-examples) +- [Common Code Patterns](#common-code-patterns) +- [Single-Record vs Batch Ingestion](#single-record-vs-batch-ingestion) +- [Choosing JSON vs Protocol Buffers](#choosing-json-vs-protocol-buffers) +- [Troubleshooting](#troubleshooting) + +## Overview + +The SDK supports two record wire formats on a standard `Stream`, plus a separate +columnar `ArrowStream`: + +**Serialization Formats:** +- **[JSON](json/README.md)** — Simpler, no schema generation required. Great for + getting started. +- **[Protocol Buffers](proto/README.md)** — Type-safe binary encoding. The + examples build the descriptor at runtime from Unity Catalog metadata + (`ProtoSchema::from_uc_json`), so **no `.proto` file or `protoc` is required**. +- **[Arrow Flight](arrow/README.md)** (Beta) — Columnar Arrow `RecordBatch` + ingestion over the Arrow Flight protocol. + +**Ingestion Methods:** +- **Single-record** (`ingest_json_record` / `ingest_proto_record`) — ingest + records one at a time. +- **Batch** (`ingest_json_records` / `ingest_proto_records`) — ingest multiple + records at once with all-or-nothing semantics. Preferred in hot paths: it + amortizes the per-call FFI crossing. +- **Arrow batch** (`ingest_batch`) — ingest one Arrow `RecordBatch` (one or many + rows) over Arrow Flight. + +**Available Examples:** + +| Example | Format | Method | Binary | +|---------|--------|--------|--------| +| [JSON Single](json/README.md#single-record-example) | JSON | Single-record | `json_single` | +| [JSON Batch](json/README.md#batch-example) | JSON | Batch | `json_batch` | +| [Proto Single](proto/README.md#single-record-example) | Protocol Buffers | Single-record | `proto_single` | +| [Proto Batch](proto/README.md#batch-example) | Protocol Buffers | Batch | `proto_batch` | +| [Arrow](arrow/README.md) | Arrow Flight (Beta) | `RecordBatch` | `arrow_ingest` | + +## Prerequisites + +### 1. Create a Databricks Table + +Create a table in your Databricks workspace: + +```sql +CREATE TABLE catalog.schema.orders ( + id INT, + customer_name STRING, + product_name STRING, + quantity INT, + price DOUBLE, + status STRING, + created_at TIMESTAMP, + updated_at TIMESTAMP +); +``` + +Replace `catalog.schema.orders` with your actual catalog, schema, and table name. + +### 2. Set Up OAuth Service Principal + +1. In your Databricks workspace, go to **Settings** > **Identity and Access**. +2. Create a service principal or use an existing one. +3. Generate OAuth credentials (client ID and secret). +4. Grant the service principal these permissions on your table: + - `SELECT` — read table schema + - `MODIFY` — write data to the table + - `USE CATALOG` and `USE SCHEMA` — access catalog and schema + +### 3. Configure Credentials + +Each example keeps its non-secret connection info as placeholder constants at +the top of the source file. Edit them to match your workspace: + +```cpp +constexpr const char* kTableName = ""; // catalog.schema.orders +constexpr const char* kWorkspaceUrl = "https://.cloud.databricks.com"; +constexpr const char* kServerEndpoint = + "https://.zerobus..cloud.databricks.com"; +``` + +Each file has a commented Azure variant alongside the AWS default — uncomment +the pair for your cloud. + +The two genuinely secret values are **not** in source; they are read from the +environment at runtime, so no credential is ever baked in: + +```bash +export DATABRICKS_CLIENT_ID="" +export DATABRICKS_CLIENT_SECRET="" +``` + +The proto examples additionally read the Unity Catalog table metadata JSON from +the environment (see the [Protocol Buffers README](proto/README.md)): + +```bash +export ZEROBUS_UC_TABLE_JSON="$(curl -s \ + -H "Authorization: Bearer $DATABRICKS_TOKEN" \ + "$WORKSPACE_URL/api/2.1/unity-catalog/tables/")" +``` + +**How to get these values:** +- **kWorkspaceUrl** — your Databricks workspace URL (Unity Catalog endpoint). +- **kTableName** — full table name in the form `catalog.schema.table`. +- **DATABRICKS_CLIENT_ID / DATABRICKS_CLIENT_SECRET** — OAuth 2.0 credentials from your + service principal. +- **kServerEndpoint** — Zerobus ingestion endpoint (usually + `https://.zerobus..cloud.databricks.com`). + +## Building the Examples + +The examples build as part of the SDK's top-level CMake build, gated on the +`ZEROBUS_BUILD_EXAMPLES` option (on by default for a top-level build). From +`cpp/`: + +```bash +make build # configure + build the SDK, tests, and examples +``` + +Or drive CMake directly: + +```bash +cmake -S . -B build -DCMAKE_BUILD_TYPE=Release +cmake --build build -j +``` + +The binaries land in `build/examples/`: + +```bash +./build/examples/json_single +./build/examples/json_batch +./build/examples/proto_single +./build/examples/proto_batch +./build/examples/arrow_ingest # only if Apache Arrow C++ is installed +``` + +The JSON and proto examples need no extra dependencies. The Arrow example +requires the Apache Arrow C++ library (`find_package(Arrow)`); if it is not +installed the `arrow_ingest` target is skipped and the other four still build. + +## Common Code Patterns + +All examples follow the same general flow. + +### 1. Initialize the SDK + +```cpp +zerobus::Sdk sdk = zerobus::Sdk::builder() + .endpoint(kServerEndpoint) + .unity_catalog_url(kWorkspaceUrl) + .application_name("my-app") + .build(); +``` + +### 2. Create a Stream + +**JSON:** +```cpp +zerobus::TableProperties props; +props.table_name = kTableName; // empty descriptor => JSON stream + +zerobus::StreamOptions options; +options.record_type = zerobus::RecordType::Json; + +zerobus::Stream stream = + sdk.create_stream(props, client_id, client_secret, options); +``` + +**Protocol Buffers (dynamic schema):** +```cpp +zerobus::ProtoSchema schema = zerobus::ProtoSchema::from_uc_json(uc_table_json); + +zerobus::TableProperties props; +props.table_name = kTableName; +props.descriptor_proto = schema.descriptor_bytes(); + +zerobus::StreamOptions options; +options.record_type = zerobus::RecordType::Proto; + +zerobus::Stream stream = + sdk.create_stream(props, client_id, client_secret, options); +``` + +**Arrow Flight (Beta):** +```cpp +zerobus::ArrowStream stream = + sdk.create_arrow_stream(kTableName, schema_ipc_bytes, client_id, + client_secret); +``` + +### 3. Ingest and Acknowledge + +> **The cardinal performance rule.** `ingest_*` returns as soon as the record is +> **queued** — sending and acknowledgement happen on background tasks. **Never +> call `wait_for_offset()` / `flush()` after every record.** Waiting per record +> forces one server round-trip per record and collapses throughput. Ingest in a +> loop, then `flush()` once at the end (or periodically for a continuous stream). + +```cpp +for (const auto& record : records) { + stream.ingest_json_record(record); // queue only — do NOT wait here +} +stream.flush(); // wait once for all pending acks +``` + +`wait_for_offset()` behaves the same way: acks are monotonic, so waiting on the +**last** offset returned by a run of ingests confirms all prior ones too. + +### 4. Close the Stream + +```cpp +stream.close(); // flush + close at a controlled point; surfaces final errors +``` + +Prefer calling `close()` explicitly rather than relying on the destructor: it +flushes pending records and surfaces any error by throwing, whereas the +destructor swallows it. + +## Single-Record vs Batch Ingestion + +| Aspect | Single-Record | Batch | +|--------|---------------|-------| +| **Method** | `ingest_json_record()` / `ingest_proto_record()` | `ingest_json_records()` / `ingest_proto_records()` | +| **Use case** | Records arrive one at a time | Multiple records ready at once | +| **Semantics** | Each record independent | All-or-nothing (atomic) | +| **Acknowledgment** | Per record | Per batch (offset of last record) | +| **Throughput** | Lower | Higher (amortizes the FFI crossing) | + +**Single-record:** +```cpp +for (const auto& record : records) { + stream.ingest_json_record(record); +} +stream.flush(); +``` + +**Batch:** +```cpp +std::int64_t last = stream.ingest_json_records(records); +if (last >= 0) { + stream.wait_for_offset(last); // one wait confirms the whole batch +} +``` + +An empty batch is a no-op and returns `-1`. + +## Choosing JSON vs Protocol Buffers + +| Feature | JSON | Protocol Buffers (dynamic) | +|---------|------|----------------------------| +| **Setup** | Simple — no schema | Fetches the descriptor from Unity Catalog at runtime | +| **Build deps** | none | none (`ProtoSchema`, no `protoc`) | +| **Type Safety** | Runtime validation | Runtime validation against the fetched schema | +| **Performance** | Text-based | Efficient binary encoding | +| **Best For** | Prototyping, flexible schemas | Production, high-throughput | + +**Recommendation:** Start with JSON for quick prototyping, then move to +Protocol Buffers for production where binary encoding and performance matter. +Both paths need no protobuf toolchain in your build. + +## Troubleshooting + +### Failed to create the stream + +**Possible causes:** +- Invalid credentials (client ID or secret). +- Service principal lacks permissions on the table. +- Incorrect workspace URL or endpoint. +- Table doesn't exist. + +**Solution:** Verify your credentials and table permissions, and confirm the +endpoint and workspace URL. + +### `environment variable ... is not set` + +The example exits with code 2 before touching the SDK if a required environment +variable is missing. Export `DATABRICKS_CLIENT_ID` and `DATABRICKS_CLIENT_SECRET` (and +`ZEROBUS_UC_TABLE_JSON` for the proto examples). + +### Invalid token + +**Possible causes:** OAuth credentials expired or invalid, or an incorrect Unity +Catalog endpoint. **Solution:** regenerate your service principal credentials and +verify the workspace URL. + +### JSON parsing / encoding errors + +**Possible causes:** the JSON record doesn't match the table schema, invalid JSON +syntax, or a type mismatch (e.g. a string where a number is expected). For proto, +`ProtoSchema::encode_json` throws if a record can't be encoded. **Solution:** +verify your record structure matches the Databricks table schema exactly. + +### `arrow_ingest` target missing + +Apache Arrow C++ is not installed. Install it (e.g. `libarrow-dev` on Debian/ +Ubuntu) and re-run CMake; the target appears automatically. + +## Additional Resources + +- [Main C++ SDK Documentation](../README.md) +- [Databricks Unity Catalog Documentation](https://docs.databricks.com/unity-catalog/index.html) diff --git a/cpp/examples/arrow/README.md b/cpp/examples/arrow/README.md new file mode 100644 index 00000000..1f8c523f --- /dev/null +++ b/cpp/examples/arrow/README.md @@ -0,0 +1,135 @@ +# Arrow Flight Example + +This directory contains an example demonstrating Arrow Flight-based data +ingestion into Databricks Delta tables using the Zerobus C++ SDK. + +> **Beta**: Arrow Flight ingestion is in Beta. The API is stabilising but may +> still change before reaching GA. + +## Table of Contents + +- [Overview](#overview) +- [Running the Example](#running-the-example) +- [Code Highlights](#code-highlights) +- [IPC Compression](#ipc-compression) +- [Adapting for Your Custom Table](#adapting-for-your-custom-table) + +## Overview + +Arrow Flight is a third record format option alongside JSON and Protocol Buffers: +it sends Apache Arrow `RecordBatch` data directly to Zerobus over the Arrow +Flight protocol. It is the best fit when your workload is naturally columnar or +batched — analytics pipelines, gateways aggregating short windows of rows, or +applications that already produce Arrow data via the Apache Arrow C++ library. + +**Features:** +- Columnar Arrow data sent over the Arrow Flight protocol +- Per-batch acknowledgments with the same recovery semantics as standard streams + +Send multiple rows per `RecordBatch`. Start with natural application-sized +batches; sending one row per call works but negates most of the performance +advantage of Arrow. For sparse, one-row-at-a-time traffic, the JSON or Protocol +Buffers examples are usually a better fit. + +**Dependency:** the Apache Arrow C++ library. The example's CMake target is built +only when `find_package(Arrow)` succeeds; otherwise it is skipped and the other +examples still build. On Debian/Ubuntu, install `libarrow-dev`. + +## Running the Example + +1. Edit the placeholder constants at the top of `arrow_ingest.cpp` (table, + endpoint, workspace URL) — see [Prerequisites](../README.md#prerequisites). + +2. Export the OAuth secrets and run: + ```bash + export DATABRICKS_CLIENT_ID="" + export DATABRICKS_CLIENT_SECRET="" + ./build/examples/arrow_ingest + ``` + +The example ingests 10 `RecordBatch`es of 100 rows each, then `flush()`es and +`close()`s at the end. + +**Expected output:** +``` +Queued 10 batches (1000 rows); last offset ID: 9 +Flushed — all batches acknowledged. +Stream closed successfully. +``` + +## Code Highlights + +**Building an Arrow Flight stream.** The stream is created from schema-only Arrow +IPC bytes (an IPC stream with just the schema message), which tell the server +what the record batches will look like: + +```cpp +std::shared_ptr schema = orders_schema(); +std::vector schema_ipc = serialize_schema_ipc(schema); + +zerobus::ArrowStream stream = + sdk.create_arrow_stream(kTableName, schema_ipc, client_id, client_secret); +``` + +**Ingest many `RecordBatch`es, then flush once.** Each `ingest_batch()` queues +one self-contained Arrow IPC stream (schema + one record-batch message) and +returns immediately with the assigned offset — there is no per-batch wait: + +```cpp +for (int b = 0; b < kBatches; ++b) { + std::shared_ptr batch = make_batch(...); + last_offset = stream.ingest_batch(serialize_ipc(batch)); // queue only +} + +stream.flush(); // wait once for all pending batches +stream.close(); +``` + +**Batch semantics:** +- **All-or-nothing per `RecordBatch`** — a batch is acknowledged as a unit. +- **Single acknowledgment** — one offset ID for the whole `RecordBatch`. +- **Schema validation** — the `RecordBatch` schema must match the schema + configured on the stream. The server validates on the first batch and fails + fast with a descriptive error on a mismatch. + +## IPC Compression + +Arrow IPC payloads can be compressed on the wire. Enable compression only when +network bandwidth limits throughput — it reduces bytes on the wire at the cost of +client CPU. Set `ArrowStreamOptions::ipc_compression` and pass the options to +`create_arrow_stream`: + +```cpp +zerobus::ArrowStreamOptions opts; +opts.ipc_compression = zerobus::IpcCompression::Zstd; // or Lz4Frame + +zerobus::ArrowStream stream = sdk.create_arrow_stream( + kTableName, schema_ipc, client_id, client_secret, opts); +``` + +- `Lz4Frame` — fast, low CPU overhead, modest compression ratio. +- `Zstd` — higher compression ratio, more CPU per batch. +- `NoCompression` (the default) — no compression. + +## Adapting for Your Custom Table + +To ingest into your own table, change the Arrow schema and the array values to +match its columns. + +1. **Update the Arrow schema** (must match the Delta table column names and types + exactly): Delta `STRING` → `arrow::large_utf8()`, `INT` → `arrow::int32()`, + `DOUBLE` → `arrow::float64()`, `TIMESTAMP` → + `arrow::timestamp(arrow::TimeUnit::MICRO, "UTC")`. These mirror the canonical + Arrow schema the Arrow Flight server derives from Delta — note `STRING` maps to + `large_utf8` (64-bit offsets), not `utf8`. + ```cpp + std::shared_ptr orders_schema() { + return arrow::schema({ + arrow::field("your_field_1", arrow::large_utf8()), + arrow::field("your_field_2", arrow::int32()), + }); + } + ``` +2. **Update `make_batch`** to populate builders matching the new schema. +3. **Update the constants** at the top of the source file: `kTableName`, + `kWorkspaceUrl`, and `kServerEndpoint`. diff --git a/cpp/examples/arrow/arrow_ingest.cpp b/cpp/examples/arrow/arrow_ingest.cpp new file mode 100644 index 00000000..b60269b3 --- /dev/null +++ b/cpp/examples/arrow/arrow_ingest.cpp @@ -0,0 +1,248 @@ +// Arrow Flight ingestion with the Zerobus C++ SDK (Beta). +// +// Arrow Flight is a third record format alongside JSON and protobuf: it streams +// columnar Apache Arrow RecordBatches to Zerobus over the Arrow Flight +// protocol. It is the best fit for workloads that are naturally columnar or +// batched — analytics pipelines, gateways aggregating short windows of rows, or +// apps that already produce Arrow data. +// +// Unlike proto/JSON streams, batches are supplied as Arrow IPC bytes (a schema +// message plus one record-batch message, in Arrow's IPC stream format), which +// the Apache Arrow C++ library builds for us. Send multiple rows per batch; +// one row per call works but negates most of Arrow's advantage. +// +// This example ingests several multi-row batches with ingest_batch(), which +// QUEUES each batch and returns immediately with its offset — it does NOT wait +// for the server ack. Loop and flush() once at the end; never wait per batch +// (one round-trip each collapses throughput). See the cardinal rule in +// zerobus.hpp. +// +// Configuration: +// * Edit the placeholder constants below to match your workspace and table. +// * The two OAuth secrets are read from the environment: +// +// export DATABRICKS_CLIENT_ID="" +// export DATABRICKS_CLIENT_SECRET="" +// +// ./build/examples/arrow_ingest +// +// Target table (see ../README.md for the CREATE TABLE statement): +// orders(id INT, customer_name STRING, product_name STRING, quantity INT, +// price DOUBLE, status STRING, created_at TIMESTAMP, updated_at +// TIMESTAMP) +// +// Dependencies: Apache Arrow C++ (found via find_package(Arrow) in the +// example's CMake target). If Arrow is not installed the target is skipped and +// the other examples still build. + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "zerobus/zerobus.hpp" + +namespace { + +// Change these constants to match your workspace and table. +constexpr const char* kTableName = + ""; // catalog.schema.orders + +// The values below are for AWS. For Azure, replace the +// `.cloud.databricks.com` hosts with `.azuredatabricks.net`. +constexpr const char* kWorkspaceUrl = + "https://.cloud.databricks.com"; +constexpr const char* kServerEndpoint = + "https://.zerobus..cloud.databricks.com"; + +constexpr int kBatches = 10; +constexpr int kRowsPerBatch = 100; + +std::string require_env(const char* name) { + const char* value = std::getenv(name); + if (value == nullptr || *value == '\0') { + std::cerr << "error: environment variable " << name << " is not set.\n" + << "Set DATABRICKS_CLIENT_ID and DATABRICKS_CLIENT_SECRET before " + "running this example.\n"; + std::exit(2); + } + return value; +} + +std::int64_t now_micros() { + return std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); +} + +// The Arrow schema must match the target Delta table's columns by name and +// type. This mirrors the canonical Arrow schema the Databricks Arrow Flight +// server derives from a Delta table: Delta STRING -> large_utf8, INT -> int32, +// DOUBLE -> float64, TIMESTAMP -> timestamp(microsecond, "UTC"). The server +// validates the record-batch schema on the first batch and fails fast with a +// descriptive error on a mismatch. +std::shared_ptr orders_schema() { + auto utc_micros = arrow::timestamp(arrow::TimeUnit::MICRO, "UTC"); + return arrow::schema({ + arrow::field("id", arrow::int32()), + arrow::field("customer_name", arrow::large_utf8()), + arrow::field("product_name", arrow::large_utf8()), + arrow::field("quantity", arrow::int32()), + arrow::field("price", arrow::float64()), + arrow::field("status", arrow::large_utf8()), + arrow::field("created_at", utc_micros), + arrow::field("updated_at", utc_micros), + }); +} + +// Abort with a clear message if an Arrow call fails, rather than silently +// ingesting partial data. +void check(const arrow::Status& status) { + if (!status.ok()) throw std::runtime_error(status.ToString()); +} + +// Build one RecordBatch containing `n` rows. Values are derived from +// `start_seed` so successive batches carry distinct data on the wire. +std::shared_ptr make_batch( + const std::shared_ptr& schema, int start_seed, int n, + std::int64_t ts) { + arrow::Int32Builder id_b; + arrow::LargeStringBuilder customer_b; + arrow::LargeStringBuilder product_b; + arrow::Int32Builder quantity_b; + arrow::DoubleBuilder price_b; + arrow::LargeStringBuilder status_b; + arrow::TimestampBuilder created_b(schema->field(6)->type(), + arrow::default_memory_pool()); + arrow::TimestampBuilder updated_b(schema->field(7)->type(), + arrow::default_memory_pool()); + + for (int i = 0; i < n; ++i) { + const int s = start_seed + i; + check(id_b.Append(s)); + check(customer_b.Append("Customer " + std::to_string(s))); + check(product_b.Append("Product " + std::to_string(s % 7))); + check(quantity_b.Append(1 + (s % 5))); + check(price_b.Append(9.99 + (s % 100))); + check(status_b.Append("pending")); + check(created_b.Append(ts)); + check(updated_b.Append(ts)); + } + + std::vector> columns(8); + check(id_b.Finish(&columns[0])); + check(customer_b.Finish(&columns[1])); + check(product_b.Finish(&columns[2])); + check(quantity_b.Finish(&columns[3])); + check(price_b.Finish(&columns[4])); + check(status_b.Finish(&columns[5])); + check(created_b.Finish(&columns[6])); + check(updated_b.Finish(&columns[7])); + + return arrow::RecordBatch::Make(schema, n, columns); +} + +// Serialize a RecordBatch into a self-contained Arrow IPC stream (schema +// message + one record-batch message) — exactly what ArrowStream::ingest_batch +// expects. Each ingest carries its own schema, so no prior state is required. +std::vector serialize_ipc( + const std::shared_ptr& batch) { + auto out_r = arrow::io::BufferOutputStream::Create(); + check(out_r.status()); + auto out = *out_r; + auto writer_r = arrow::ipc::MakeStreamWriter(out, batch->schema()); + check(writer_r.status()); + auto writer = *writer_r; + check(writer->WriteRecordBatch(*batch)); + check(writer->Close()); + auto buf_r = out->Finish(); + check(buf_r.status()); + auto buf = *buf_r; + return std::vector(buf->data(), buf->data() + buf->size()); +} + +// Build the schema-only Arrow IPC bytes: a stream containing just the schema +// message and no record batches. That is what Sdk::create_arrow_stream expects. +std::vector serialize_schema_ipc( + const std::shared_ptr& schema) { + auto out_r = arrow::io::BufferOutputStream::Create(); + check(out_r.status()); + auto out = *out_r; + auto writer_r = arrow::ipc::MakeStreamWriter(out, schema); + check(writer_r.status()); + auto writer = *writer_r; + check(writer->Close()); + auto buf_r = out->Finish(); + check(buf_r.status()); + auto buf = *buf_r; + return std::vector(buf->data(), buf->data() + buf->size()); +} + +} // namespace + +int main() { + const std::string client_id = require_env("DATABRICKS_CLIENT_ID"); + const std::string client_secret = require_env("DATABRICKS_CLIENT_SECRET"); + + try { + // 1. Build the SDK. + zerobus::Sdk sdk = zerobus::Sdk::builder() + .endpoint(kServerEndpoint) + .unity_catalog_url(kWorkspaceUrl) + .application_name("arrow-ingest") + .build(); + + // 2. Open an Arrow stream. The schema-only IPC bytes tell the server what + // the record batches will look like. Optional IPC compression trades + // client CPU for fewer bytes on the wire; enable it only when network + // bandwidth limits throughput: + // zerobus::ArrowStreamOptions opts; + // opts.ipc_compression = zerobus::IpcCompression::Zstd; // or + // Lz4Frame + const std::shared_ptr schema = orders_schema(); + const std::vector schema_ipc = serialize_schema_ipc(schema); + + zerobus::ArrowStream stream = sdk.create_arrow_stream( + kTableName, schema_ipc, client_id, client_secret); + + // 3. Ingest a series of multi-row batches. Each ingest_batch() queues one + // Arrow IPC stream and returns immediately with the assigned offset — + // there is NO per-batch wait here. The loop-then-flush pattern is the + // whole point of the async API. + const std::int64_t ts = now_micros(); + std::int64_t last_offset = -1; + for (int b = 0; b < kBatches; ++b) { + std::shared_ptr batch = + make_batch(schema, b * kRowsPerBatch, kRowsPerBatch, ts); + last_offset = stream.ingest_batch(serialize_ipc(batch)); + } + std::cout << "Queued " << kBatches << " batches (" + << kBatches * kRowsPerBatch + << " rows); last offset ID: " << last_offset << "\n"; + + // 4. One flush drains every pending batch to a durable server ack, then + // close at a controlled point. + stream.flush(); + std::cout << "Flushed — all batches acknowledged.\n"; + + stream.close(); + std::cout << "Stream closed successfully.\n"; + } catch (const zerobus::ZerobusException& e) { + std::cerr << "Zerobus error: " << e.what() + << " (retryable=" << (e.is_retryable() ? "true" : "false") + << ")\n"; + return 1; + } catch (const std::exception& e) { + std::cerr << "Arrow error: " << e.what() << "\n"; + return 1; + } + + return 0; +} diff --git a/cpp/examples/json/README.md b/cpp/examples/json/README.md new file mode 100644 index 00000000..9bf6e6d0 --- /dev/null +++ b/cpp/examples/json/README.md @@ -0,0 +1,142 @@ +# JSON Examples + +This directory contains examples demonstrating JSON-based data ingestion into +Databricks Delta tables using the Zerobus C++ SDK. + +## Table of Contents + +- [Overview](#overview) +- [Single-Record Example](#single-record-example) + - [Running the Example](#running-the-example) + - [Code Highlights](#code-highlights) +- [Batch Example](#batch-example) + - [Running the Example](#running-the-example-1) + - [Code Highlights](#code-highlights-1) +- [Adapting for Your Custom Table](#adapting-for-your-custom-table) + +## Overview + +JSON examples are recommended for getting started — they're simpler and don't +require any schema handling. The server maps each record's JSON fields onto the +table's columns by name. + +**Features:** +- No schema generation required +- Easy to understand and modify +- Great for quick prototyping + +**Available examples:** +- **`single.cpp`** — ingest records one at a time using `ingest_json_record()` +- **`batch.cpp`** — ingest multiple records at once using `ingest_json_records()` + +Both open a JSON stream by setting `StreamOptions::record_type` to +`RecordType::Json` and leaving `TableProperties::descriptor_proto` empty. + +## Single-Record Example + +### Running the Example + +1. Edit the placeholder constants at the top of `single.cpp` (table, endpoint, + workspace URL) — see [Prerequisites](../README.md#prerequisites). + +2. Export the OAuth secrets and run: + ```bash + export DATABRICKS_CLIENT_ID="" + export DATABRICKS_CLIENT_SECRET="" + ./build/examples/json_single + ``` + +**Expected output:** +``` +Record 1 queued with offset ID: 0 +Record 2 queued with offset ID: 1 +Record 3 queued with offset ID: 2 +All records acknowledged. +Stream closed successfully. +``` + +### Code Highlights + +Each `ingest_json_record()` returns as soon as the record is queued; the example +ingests all of them and calls `flush()` ONCE at the end rather than waiting per +record (waiting per record forces a server round-trip each time): + +```cpp +std::int64_t offset = stream.ingest_json_record(record1); // queue only +offset = stream.ingest_json_record(record2); // queue only +offset = stream.ingest_json_record(record3); // queue only + +stream.flush(); // the single wait point — confirm all queued records at once +``` + +**Building a JSON stream:** +```cpp +zerobus::TableProperties props; +props.table_name = kTableName; // empty descriptor => JSON stream + +zerobus::StreamOptions options; +options.record_type = zerobus::RecordType::Json; + +zerobus::Stream stream = + sdk.create_stream(props, client_id, client_secret, options); +``` + +## Batch Example + +### Running the Example + +1. Edit the placeholder constants at the top of `batch.cpp` — see + [Prerequisites](../README.md#prerequisites). + +2. Export the OAuth secrets and run: + ```bash + export DATABRICKS_CLIENT_ID="" + export DATABRICKS_CLIENT_SECRET="" + ./build/examples/json_batch + ``` + +**Expected output:** +``` +Batch of 3 records queued; last offset ID: 2 +Batch acknowledged through offset ID: 2 +Stream closed successfully. +``` + +### Code Highlights + +`ingest_json_records()` hands a whole vector of records to the SDK in a single +FFI crossing and returns the offset of the **last** record. Waiting on that one +offset confirms the whole batch, because acks are monotonic: + +```cpp +const std::vector batch = { record1, record2, record3 }; + +const std::int64_t last_offset = stream.ingest_json_records(batch); +if (last_offset >= 0) { + stream.wait_for_offset(last_offset); // one wait confirms the batch +} +``` + +**Batch semantics:** +- **All-or-nothing** — the entire batch succeeds or fails as a unit. +- **Single acknowledgment** — one offset (the last record's) for the whole batch. +- **Empty batches** — a no-op; `ingest_json_records()` returns `-1`. + +In a hot path you would queue **many** batches and `flush()` once, rather than +waiting after each batch. + +## Adapting for Your Custom Table + +JSON examples require no schema generation. To use your own table: + +1. **Update the record shape.** Change the JSON your records produce (the + `make_order_json` helper, or the raw JSON literals) to match your table's + columns and types: + ```cpp + std::string record = R"({"your_field_1": "value", "your_field_2": 123})"; + ``` +2. **Update the constants** at the top of the source file: `kTableName`, + `kWorkspaceUrl`, and `kServerEndpoint`. + +> **Tip.** Delta `TIMESTAMP` columns are int64 microseconds since the Unix epoch +> (UTC) — the examples fill them with `now_micros()`. diff --git a/cpp/examples/json/batch.cpp b/cpp/examples/json/batch.cpp new file mode 100644 index 00000000..3f1391f0 --- /dev/null +++ b/cpp/examples/json/batch.cpp @@ -0,0 +1,141 @@ +// Batch JSON ingestion with the Zerobus C++ SDK. +// +// This example opens a JSON stream and ingests records with the BATCH API, +// ingest_json_records(), which hands a whole vector of records to the SDK in a +// single FFI crossing. Batching is the right choice in hot paths: each FFI +// crossing has a fixed cost that batching amortizes, and a batch is +// acknowledged all-or-nothing as a unit. +// +// The batch call returns the offset of the LAST record in the batch. Because +// acks are monotonic, waiting on that single offset confirms the whole batch. +// +// Configuration: +// * Edit the placeholder constants below to match your workspace and table. +// * The two OAuth secrets are read from the environment: +// +// export DATABRICKS_CLIENT_ID="" +// export DATABRICKS_CLIENT_SECRET="" +// +// ./build/examples/json_batch +// +// Target table (see ../README.md for the CREATE TABLE statement): +// orders(id INT, customer_name STRING, product_name STRING, quantity INT, +// price DOUBLE, status STRING, created_at TIMESTAMP, updated_at +// TIMESTAMP) + +#include +#include +#include +#include +#include +#include + +#include "zerobus/zerobus.hpp" + +namespace { + +// Change these constants to match your workspace and table. +constexpr const char* kTableName = + ""; // catalog.schema.orders + +// The values below are for AWS. For Azure, replace the +// `.cloud.databricks.com` hosts with `.azuredatabricks.net`. +constexpr const char* kWorkspaceUrl = + "https://.cloud.databricks.com"; +constexpr const char* kServerEndpoint = + "https://.zerobus..cloud.databricks.com"; + +std::string require_env(const char* name) { + const char* value = std::getenv(name); + if (value == nullptr || *value == '\0') { + std::cerr << "error: environment variable " << name << " is not set.\n" + << "Set DATABRICKS_CLIENT_ID and DATABRICKS_CLIENT_SECRET before " + "running this example.\n"; + std::exit(2); + } + return value; +} + +std::int64_t now_micros() { + return std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); +} + +std::string make_order_json(int id, const std::string& customer, + const std::string& product, int quantity, + double price, const std::string& status, + std::int64_t ts) { + return "{\"id\": " + std::to_string(id) + ", \"customer_name\": \"" + + customer + "\", \"product_name\": \"" + product + + "\", \"quantity\": " + std::to_string(quantity) + + ", \"price\": " + std::to_string(price) + ", \"status\": \"" + status + + "\", \"created_at\": " + std::to_string(ts) + + ", \"updated_at\": " + std::to_string(ts) + "}"; +} + +} // namespace + +int main() { + const std::string client_id = require_env("DATABRICKS_CLIENT_ID"); + const std::string client_secret = require_env("DATABRICKS_CLIENT_SECRET"); + + try { + // 1. Build the SDK. + zerobus::Sdk sdk = zerobus::Sdk::builder() + .endpoint(kServerEndpoint) + .unity_catalog_url(kWorkspaceUrl) + .application_name("json-batch") + .build(); + + // 2. Open a JSON stream. + zerobus::TableProperties props; + props.table_name = kTableName; + + zerobus::StreamOptions options; + options.record_type = zerobus::RecordType::Json; + + zerobus::Stream stream = + sdk.create_stream(props, client_id, client_secret, options); + + const std::int64_t now = now_micros(); + + // 3. Build a batch and hand it over in one call. ingest_json_records() + // queues the whole vector and returns the offset of the LAST record. + const std::vector batch = { + make_order_json(1, "Alice Smith", "Wireless Mouse", 2, 25.99, "pending", + now), + make_order_json(2, "Bob Johnson", "Mechanical Keyboard", 1, 89.99, + "shipped", now), + make_order_json(3, "Carol Williams", "USB-C Hub", 3, 45.00, "delivered", + now), + }; + + const std::int64_t last_offset = stream.ingest_json_records(batch); + std::cout << "Batch of " << batch.size() + << " records queued; last offset ID: " << last_offset << "\n"; + + // 4. Confirm the batch. Waiting on the last offset is enough — acks are + // monotonic, so offset N acked implies every offset <= N is acked. In a + // hot path you would queue many batches and flush() once instead of + // waiting after each. + if (last_offset >= 0) { + stream.wait_for_offset(last_offset); + std::cout << "Batch acknowledged through offset ID: " << last_offset + << "\n"; + } + + // 5. flush() drains anything still pending, then close at a controlled + // point. + stream.flush(); + stream.close(); + std::cout << "Stream closed successfully.\n"; + } catch (const zerobus::ZerobusException& e) { + std::cerr << "Zerobus error: " << e.what() + << " (retryable=" << (e.is_retryable() ? "true" : "false") + << ")\n"; + return 1; + } + + return 0; +} diff --git a/cpp/examples/json/single.cpp b/cpp/examples/json/single.cpp new file mode 100644 index 00000000..62d66809 --- /dev/null +++ b/cpp/examples/json/single.cpp @@ -0,0 +1,150 @@ +// Single-record JSON ingestion with the Zerobus C++ SDK. +// +// This example opens a JSON stream to a Delta table and ingests a handful of +// records ONE AT A TIME with ingest_json_record(), then flushes ONCE at the +// end. That is the correct pattern: ingest_json_record() returns as soon as the +// record is queued; sending and acknowledgement happen on background tasks. +// Calling wait_for_offset()/flush() after every record would force a full +// server round-trip per record and collapse throughput. For high volume, prefer +// the batch API in batch.cpp. +// +// Configuration: +// * Edit the placeholder constants below (table, endpoint, workspace URL) to +// match your Databricks workspace. +// * The two OAuth secrets are read from the environment so no credential is +// ever baked into source: +// +// export DATABRICKS_CLIENT_ID="" +// export DATABRICKS_CLIENT_SECRET="" +// +// ./build/examples/json_single +// +// Target table (see ../README.md for the CREATE TABLE statement): +// orders(id INT, customer_name STRING, product_name STRING, quantity INT, +// price DOUBLE, status STRING, created_at TIMESTAMP, updated_at +// TIMESTAMP) + +#include +#include +#include +#include +#include + +#include "zerobus/zerobus.hpp" + +namespace { + +// Change these constants to match your workspace and table. +constexpr const char* kTableName = + ""; // catalog.schema.orders + +// The values below are for AWS. For Azure, replace the +// `.cloud.databricks.com` hosts with `.azuredatabricks.net`. +constexpr const char* kWorkspaceUrl = + "https://.cloud.databricks.com"; +constexpr const char* kServerEndpoint = + "https://.zerobus..cloud.databricks.com"; + +// Read a required environment variable or exit with a clear message. Exiting +// (rather than throwing) keeps a misconfigured environment distinct from a +// genuine SDK ZerobusException below. +std::string require_env(const char* name) { + const char* value = std::getenv(name); + if (value == nullptr || *value == '\0') { + std::cerr << "error: environment variable " << name << " is not set.\n" + << "Set DATABRICKS_CLIENT_ID and DATABRICKS_CLIENT_SECRET before " + "running this example.\n"; + std::exit(2); + } + return value; +} + +// Delta TIMESTAMP is an int64 count of microseconds since the Unix epoch (UTC). +std::int64_t now_micros() { + return std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); +} + +// Build one order record as a JSON string matching the table columns. +std::string make_order_json(int id, const std::string& customer, + const std::string& product, int quantity, + double price, const std::string& status, + std::int64_t ts) { + return "{\"id\": " + std::to_string(id) + ", \"customer_name\": \"" + + customer + "\", \"product_name\": \"" + product + + "\", \"quantity\": " + std::to_string(quantity) + + ", \"price\": " + std::to_string(price) + ", \"status\": \"" + status + + "\", \"created_at\": " + std::to_string(ts) + + ", \"updated_at\": " + std::to_string(ts) + "}"; +} + +} // namespace + +int main() { + const std::string client_id = require_env("DATABRICKS_CLIENT_ID"); + const std::string client_secret = require_env("DATABRICKS_CLIENT_SECRET"); + + try { + // 1. Build the SDK — an authenticated connection factory. TLS is on by + // default; the builder is consumed by build(). + zerobus::Sdk sdk = zerobus::Sdk::builder() + .endpoint(kServerEndpoint) + .unity_catalog_url(kWorkspaceUrl) + .application_name("json-single") + .build(); + + // 2. Open a JSON stream. record_type must be Json to match the payloads, + // and descriptor_proto is left empty (no schema needed for JSON — the + // server maps each record's fields onto the table's columns by name). + zerobus::TableProperties props; + props.table_name = kTableName; + + zerobus::StreamOptions options; + options.record_type = zerobus::RecordType::Json; + + zerobus::Stream stream = + sdk.create_stream(props, client_id, client_secret, options); + + const std::int64_t now = now_micros(); + + // 3. Ingest records one at a time. Each call queues the record and returns + // immediately with the assigned offset — there is NO per-record wait + // here. The single wait point is the flush() below. + std::int64_t offset = stream.ingest_json_record(make_order_json( + 1, "Alice Smith", "Wireless Mouse", 2, 25.99, "pending", now)); + std::cout << "Record 1 queued with offset ID: " << offset << "\n"; + + offset = stream.ingest_json_record(make_order_json( + 2, "Bob Johnson", "Mechanical Keyboard", 1, 89.99, "shipped", now)); + std::cout << "Record 2 queued with offset ID: " << offset << "\n"; + + // A raw JSON literal works exactly the same — any UTF-8 JSON string that + // matches the table schema is accepted. + offset = stream.ingest_json_record( + R"({"id": 3, "customer_name": "Carol Williams", "product_name": "USB-C Hub", )" + R"("quantity": 3, "price": 45.00, "status": "delivered", )" + "\"created_at\": " + + std::to_string(now) + ", \"updated_at\": " + std::to_string(now) + "}"); + std::cout << "Record 3 queued with offset ID: " << offset << "\n"; + + // 4. Flush once at the end: block until every queued record is durably + // acknowledged by the server. This is the right place to wait — not + // after each individual ingest above. + stream.flush(); + std::cout << "All records acknowledged.\n"; + + // 5. Close at a controlled point rather than leaving it to the destructor. + // close() surfaces any final error by throwing; ~Stream() would swallow + // it. + stream.close(); + std::cout << "Stream closed successfully.\n"; + } catch (const zerobus::ZerobusException& e) { + std::cerr << "Zerobus error: " << e.what() + << " (retryable=" << (e.is_retryable() ? "true" : "false") + << ")\n"; + return 1; + } + + return 0; +} diff --git a/cpp/examples/proto/README.md b/cpp/examples/proto/README.md new file mode 100644 index 00000000..15aa4b2e --- /dev/null +++ b/cpp/examples/proto/README.md @@ -0,0 +1,184 @@ +# Protocol Buffers Examples + +This directory contains examples demonstrating Protocol Buffers-based data +ingestion into Databricks Delta tables using the Zerobus C++ SDK. + +## Table of Contents + +- [Overview](#overview) +- [How the Schema Is Built (Dynamic Proto)](#how-the-schema-is-built-dynamic-proto) +- [Single-Record Example](#single-record-example) + - [Running the Example](#running-the-example) + - [Code Highlights](#code-highlights) +- [Batch Example](#batch-example) + - [Running the Example](#running-the-example-1) + - [Code Highlights](#code-highlights-1) +- [Adapting for Your Custom Table](#adapting-for-your-custom-table) + +## Overview + +Protocol Buffers provide efficient binary encoding. **No `.proto` file and no +`protoc` are required to run these examples** — the descriptor is built at +runtime from Unity Catalog table metadata via `ProtoSchema::from_uc_json()`, +which also encodes each record's JSON into protobuf bytes. + +**Features:** +- Efficient binary encoding +- No protobuf toolchain in your build +- Schema always matches the live table (fetched from Unity Catalog) + +**Available examples:** +- **`single.cpp`** — ingest records one at a time using `ingest_proto_record()` +- **`batch.cpp`** — ingest multiple records at once using `ingest_proto_records()` + +The `orders.proto` file in this directory is a **human-readable reference** for +the table's shape (and a starting point if you prefer the static-proto path). The +examples do not compile or use it. + +## How the Schema Is Built (Dynamic Proto) + +`ProtoSchema::from_uc_json()` takes the JSON body of +`GET /api/2.1/unity-catalog/tables/{full_name}` and produces: + +1. **A descriptor** (`descriptor_bytes()`) — passed as + `TableProperties::descriptor_proto` when creating the proto stream. +2. **A JSON→proto encoder** (`encode_json()`) — turns each record's JSON string + into the protobuf bytes the stream ingests. + +Fetch the metadata JSON once and pass it via the environment: + +```bash +export ZEROBUS_UC_TABLE_JSON="$(curl -s \ + -H "Authorization: Bearer $DATABRICKS_TOKEN" \ + "$WORKSPACE_URL/api/2.1/unity-catalog/tables/")" +``` + +Per-column value shaping rules (DATE/TIMESTAMP as integers, BINARY as base64, +etc.) are documented in the FFI README / `zerobus.h`. + +## Single-Record Example + +### Running the Example + +1. Edit the placeholder constants at the top of `single.cpp` — see + [Prerequisites](../README.md#prerequisites). + +2. Export the secrets and the table metadata, then run: + ```bash + export DATABRICKS_CLIENT_ID="" + export DATABRICKS_CLIENT_SECRET="" + export ZEROBUS_UC_TABLE_JSON="$(curl -s \ + -H "Authorization: Bearer $DATABRICKS_TOKEN" \ + "$WORKSPACE_URL/api/2.1/unity-catalog/tables/")" + ./build/examples/proto_single + ``` + +**Expected output:** +``` +Record 1 queued with offset ID: 0 +Record 2 queued with offset ID: 1 +Record 3 queued with offset ID: 2 +All records acknowledged. +Stream closed successfully. +``` + +### Code Highlights + +Each iteration encodes the record's JSON to protobuf bytes and queues them; the +example ingests all of them and calls `flush()` ONCE at the end rather than +waiting per record: + +```cpp +zerobus::ProtoSchema schema = zerobus::ProtoSchema::from_uc_json(uc_table_json); + +std::vector encoded = schema.encode_json(record_json); +std::int64_t offset = stream.ingest_proto_record(encoded); // queue only +// ... encode + ingest more records ... + +stream.flush(); // the single wait point — confirm all queued records at once +``` + +**Building a Protocol Buffers stream:** +```cpp +zerobus::TableProperties props; +props.table_name = kTableName; +props.descriptor_proto = schema.descriptor_bytes(); + +zerobus::StreamOptions options; +options.record_type = zerobus::RecordType::Proto; + +zerobus::Stream stream = + sdk.create_stream(props, client_id, client_secret, options); +``` + +## Batch Example + +### Running the Example + +1. Edit the placeholder constants at the top of `batch.cpp` — see + [Prerequisites](../README.md#prerequisites). + +2. Export the secrets and the table metadata, then run: + ```bash + export DATABRICKS_CLIENT_ID="" + export DATABRICKS_CLIENT_SECRET="" + export ZEROBUS_UC_TABLE_JSON="$(curl -s \ + -H "Authorization: Bearer $DATABRICKS_TOKEN" \ + "$WORKSPACE_URL/api/2.1/unity-catalog/tables/")" + ./build/examples/proto_batch + ``` + +**Expected output:** +``` +Batch of 3 records queued; last offset ID: 2 +Batch acknowledged through offset ID: 2 +Stream closed successfully. +``` + +### Code Highlights + +Encode each record, collect the bytes into a batch, and hand the whole batch to +`ingest_proto_records()` in one call. It returns the offset of the **last** +record; waiting on that one offset confirms the batch: + +```cpp +const std::vector> batch = { + schema.encode_json(record1), + schema.encode_json(record2), + schema.encode_json(record3), +}; + +const std::int64_t last_offset = stream.ingest_proto_records(batch); +if (last_offset >= 0) { + stream.wait_for_offset(last_offset); // one wait confirms the batch +} +``` + +**Batch semantics:** +- **All-or-nothing** — the entire batch succeeds or fails as a unit. +- **Single acknowledgment** — one offset (the last record's) for the whole batch. +- **Empty batches** — a no-op; `ingest_proto_records()` returns `-1`. + +In a hot path you would queue **many** batches and `flush()` once, rather than +waiting after each batch. + +## Adapting for Your Custom Table + +Because the schema is fetched from Unity Catalog at runtime, adapting to your own +table needs no code changes to the encoding: + +1. **Point `ZEROBUS_UC_TABLE_JSON` at your table** — fetch the metadata for your + own `catalog.schema.table` (see [How the Schema Is + Built](#how-the-schema-is-built-dynamic-proto)). +2. **Update the record shape.** Change the JSON your records produce (the + `make_order_json` helper) so its fields match your table's columns; unknown + keys are ignored by `encode_json()`. +3. **Update the constants** at the top of the source file: `kTableName`, + `kWorkspaceUrl`, and `kServerEndpoint`. + +> **Note on static proto.** These examples use the dynamic path only, which needs +> no protobuf toolchain. If you instead want compile-time typing and no runtime +> Unity Catalog fetch, compile `orders.proto` with `protoc`, build a +> `DescriptorProto` from the generated C++ class, and pass its serialized bytes +> as `TableProperties::descriptor_proto`. That trades drift-safety (the `.proto` +> must be kept in sync with the table by hand) for those benefits. diff --git a/cpp/examples/proto/batch.cpp b/cpp/examples/proto/batch.cpp new file mode 100644 index 00000000..9704c91b --- /dev/null +++ b/cpp/examples/proto/batch.cpp @@ -0,0 +1,152 @@ +// Batch protobuf ingestion with the Zerobus C++ SDK (dynamic schema). +// +// Like proto/single.cpp, this builds the table's descriptor from Unity Catalog +// metadata via ProtoSchema::from_uc_json() (no .proto file, no protoc) and uses +// the same ProtoSchema to encode records. Here records are ingested with the +// BATCH API, ingest_proto_records(), which hands a whole vector of encoded +// records to the SDK in a single FFI crossing. +// +// Batching is the right choice in hot paths: each FFI crossing has a fixed cost +// that batching amortizes, and a batch is acknowledged all-or-nothing as a +// unit. The call returns the offset of the LAST record; because acks are +// monotonic, waiting on that single offset confirms the whole batch. +// +// Configuration: +// * Edit the placeholder constants below to match your workspace and table. +// * Secrets and the Unity Catalog table metadata JSON are read from the +// environment: +// +// export DATABRICKS_CLIENT_ID="" +// export DATABRICKS_CLIENT_SECRET="" +// # JSON body of GET /api/2.1/unity-catalog/tables/{full_name}: +// export ZEROBUS_UC_TABLE_JSON="$(curl -s \ +// -H "Authorization: Bearer $DATABRICKS_TOKEN" \ +// "$WORKSPACE_URL/api/2.1/unity-catalog/tables/")" +// +// ./build/examples/proto_batch +// +// Target table (see ../README.md and orders.proto): +// orders(id INT, customer_name STRING, product_name STRING, quantity INT, +// price DOUBLE, status STRING, created_at TIMESTAMP, updated_at +// TIMESTAMP) + +#include +#include +#include +#include +#include +#include + +#include "zerobus/zerobus.hpp" + +namespace { + +// Change these constants to match your workspace and table. +constexpr const char* kTableName = + ""; // catalog.schema.orders + +// The values below are for AWS. For Azure, replace the +// `.cloud.databricks.com` hosts with `.azuredatabricks.net`. +constexpr const char* kWorkspaceUrl = + "https://.cloud.databricks.com"; +constexpr const char* kServerEndpoint = + "https://.zerobus..cloud.databricks.com"; + +std::string require_env(const char* name) { + const char* value = std::getenv(name); + if (value == nullptr || *value == '\0') { + std::cerr << "error: environment variable " << name << " is not set.\n" + << "See the header of this file for the required variables.\n"; + std::exit(2); + } + return value; +} + +std::int64_t now_micros() { + return std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); +} + +std::string make_order_json(int id, const std::string& customer, + const std::string& product, int quantity, + double price, const std::string& status, + std::int64_t ts) { + return "{\"id\": " + std::to_string(id) + ", \"customer_name\": \"" + + customer + "\", \"product_name\": \"" + product + + "\", \"quantity\": " + std::to_string(quantity) + + ", \"price\": " + std::to_string(price) + ", \"status\": \"" + status + + "\", \"created_at\": " + std::to_string(ts) + + ", \"updated_at\": " + std::to_string(ts) + "}"; +} + +} // namespace + +int main() { + const std::string client_id = require_env("DATABRICKS_CLIENT_ID"); + const std::string client_secret = require_env("DATABRICKS_CLIENT_SECRET"); + const std::string uc_table_json = require_env("ZEROBUS_UC_TABLE_JSON"); + + try { + // 1. Build a protobuf schema for the table from its Unity Catalog metadata. + zerobus::ProtoSchema schema = + zerobus::ProtoSchema::from_uc_json(uc_table_json); + + // 2. Build the SDK. + zerobus::Sdk sdk = zerobus::Sdk::builder() + .endpoint(kServerEndpoint) + .unity_catalog_url(kWorkspaceUrl) + .application_name("proto-batch") + .build(); + + // 3. Open a proto stream, passing the descriptor. + zerobus::TableProperties props; + props.table_name = kTableName; + props.descriptor_proto = schema.descriptor_bytes(); + + zerobus::StreamOptions options; + options.record_type = zerobus::RecordType::Proto; + + zerobus::Stream stream = + sdk.create_stream(props, client_id, client_secret, options); + + const std::int64_t now = now_micros(); + + // 4. Encode each record's JSON to protobuf bytes, collect them into a + // batch, then hand the whole batch over in a single call. + const std::vector> batch = { + schema.encode_json(make_order_json(1, "Alice Smith", "Wireless Mouse", + 2, 25.99, "pending", now)), + schema.encode_json(make_order_json( + 2, "Bob Johnson", "Mechanical Keyboard", 1, 89.99, "shipped", now)), + schema.encode_json(make_order_json(3, "Carol Williams", "USB-C Hub", 3, + 45.00, "delivered", now)), + }; + + const std::int64_t last_offset = stream.ingest_proto_records(batch); + std::cout << "Batch of " << batch.size() + << " records queued; last offset ID: " << last_offset << "\n"; + + // 5. Confirm the batch. Waiting on the last offset is enough — acks are + // monotonic. In a hot path you would queue many batches and flush() once + // instead of waiting after each. + if (last_offset >= 0) { + stream.wait_for_offset(last_offset); + std::cout << "Batch acknowledged through offset ID: " << last_offset + << "\n"; + } + + // 6. flush() drains anything still pending, then close at a controlled + // point. + stream.flush(); + stream.close(); + std::cout << "Stream closed successfully.\n"; + } catch (const zerobus::ZerobusException& e) { + std::cerr << "Zerobus error: " << e.what() + << " (retryable=" << (e.is_retryable() ? "true" : "false") + << ")\n"; + return 1; + } + + return 0; +} diff --git a/cpp/examples/proto/orders.proto b/cpp/examples/proto/orders.proto new file mode 100644 index 00000000..409c0287 --- /dev/null +++ b/cpp/examples/proto/orders.proto @@ -0,0 +1,30 @@ +// Protobuf schema for the example `orders` table: +// orders(id INT, customer_name STRING, product_name STRING, quantity INT, +// price DOUBLE, status STRING, created_at TIMESTAMP, updated_at TIMESTAMP) +// +// NOTE: the C++ examples in this directory (single.cpp / batch.cpp) do NOT use +// this file. They build the descriptor at runtime from Unity Catalog table +// metadata via ProtoSchema::from_uc_json(), so no .proto file and no protoc are +// required to run them. This file is kept as a human-readable reference for the +// table's shape and as a starting point if you prefer the static-proto path +// (compile it with protoc and build the descriptor from the generated class). +// +// Field numbers map to the table's column ordinals (1-based, declaration order). +// proto2 is used so nullable columns can be `optional` (proto3 has no explicit +// presence for scalars). INT -> int32, STRING -> string, DOUBLE -> double, +// TIMESTAMP -> int64 microseconds since the Unix epoch (UTC). + +syntax = "proto2"; + +package zerobus_examples; + +message Order { + optional int32 id = 1; + optional string customer_name = 2; + optional string product_name = 3; + optional int32 quantity = 4; + optional double price = 5; + optional string status = 6; + optional int64 created_at = 7; + optional int64 updated_at = 8; +} diff --git a/cpp/examples/proto/single.cpp b/cpp/examples/proto/single.cpp new file mode 100644 index 00000000..c3e06163 --- /dev/null +++ b/cpp/examples/proto/single.cpp @@ -0,0 +1,155 @@ +// Single-record protobuf ingestion with the Zerobus C++ SDK (dynamic schema). +// +// This example builds a protobuf descriptor for the target table straight from +// its Unity Catalog metadata via ProtoSchema::from_uc_json() — no hand-written +// .proto file and no protoc required. The same ProtoSchema also encodes each +// record's JSON into protobuf bytes for ingestion. +// +// Records are ingested ONE AT A TIME with ingest_proto_record(), then flushed +// ONCE at the end. ingest_proto_record() returns as soon as the record is +// queued; sending and acknowledgement happen on background tasks. Calling +// wait_for_offset()/flush() after every record would force a full server +// round-trip per record and collapse throughput. For high volume, prefer the +// batch API in batch.cpp. +// +// Configuration: +// * Edit the placeholder constants below to match your workspace and table. +// * The two OAuth secrets and the Unity Catalog table metadata JSON are read +// from the environment so no credential is baked into source: +// +// export DATABRICKS_CLIENT_ID="" +// export DATABRICKS_CLIENT_SECRET="" +// # JSON body of GET /api/2.1/unity-catalog/tables/{full_name}: +// export ZEROBUS_UC_TABLE_JSON="$(curl -s \ +// -H "Authorization: Bearer $DATABRICKS_TOKEN" \ +// "$WORKSPACE_URL/api/2.1/unity-catalog/tables/")" +// +// ./build/examples/proto_single +// +// Target table (see ../README.md and orders.proto): +// orders(id INT, customer_name STRING, product_name STRING, quantity INT, +// price DOUBLE, status STRING, created_at TIMESTAMP, updated_at +// TIMESTAMP) + +#include +#include +#include +#include +#include +#include + +#include "zerobus/zerobus.hpp" + +namespace { + +// Change these constants to match your workspace and table. +constexpr const char* kTableName = + ""; // catalog.schema.orders + +// The values below are for AWS. For Azure, replace the +// `.cloud.databricks.com` hosts with `.azuredatabricks.net`. +constexpr const char* kWorkspaceUrl = + "https://.cloud.databricks.com"; +constexpr const char* kServerEndpoint = + "https://.zerobus..cloud.databricks.com"; + +std::string require_env(const char* name) { + const char* value = std::getenv(name); + if (value == nullptr || *value == '\0') { + std::cerr << "error: environment variable " << name << " is not set.\n" + << "See the header of this file for the required variables.\n"; + std::exit(2); + } + return value; +} + +std::int64_t now_micros() { + return std::chrono::duration_cast( + std::chrono::system_clock::now().time_since_epoch()) + .count(); +} + +// Build one order record as a JSON string. ProtoSchema::encode_json shapes it +// into protobuf bytes; per-column value rules (DATE/TIMESTAMP as integers, +// BINARY as base64, etc.) are documented in the FFI README / zerobus.h. +std::string make_order_json(int id, const std::string& customer, + const std::string& product, int quantity, + double price, const std::string& status, + std::int64_t ts) { + return "{\"id\": " + std::to_string(id) + ", \"customer_name\": \"" + + customer + "\", \"product_name\": \"" + product + + "\", \"quantity\": " + std::to_string(quantity) + + ", \"price\": " + std::to_string(price) + ", \"status\": \"" + status + + "\", \"created_at\": " + std::to_string(ts) + + ", \"updated_at\": " + std::to_string(ts) + "}"; +} + +} // namespace + +int main() { + const std::string client_id = require_env("DATABRICKS_CLIENT_ID"); + const std::string client_secret = require_env("DATABRICKS_CLIENT_SECRET"); + const std::string uc_table_json = require_env("ZEROBUS_UC_TABLE_JSON"); + + try { + // 1. Build a protobuf schema for the table from its Unity Catalog metadata. + // This yields both the descriptor (for stream creation) and a + // JSON->proto encoder — no .proto file required. + zerobus::ProtoSchema schema = + zerobus::ProtoSchema::from_uc_json(uc_table_json); + + // 2. Build the SDK. + zerobus::Sdk sdk = zerobus::Sdk::builder() + .endpoint(kServerEndpoint) + .unity_catalog_url(kWorkspaceUrl) + .application_name("proto-single") + .build(); + + // 3. Open a proto stream, passing the descriptor built above. + zerobus::TableProperties props; + props.table_name = kTableName; + props.descriptor_proto = schema.descriptor_bytes(); + + zerobus::StreamOptions options; + options.record_type = zerobus::RecordType::Proto; + + zerobus::Stream stream = + sdk.create_stream(props, client_id, client_secret, options); + + const std::int64_t now = now_micros(); + + // 4. Ingest records one at a time. Each iteration encodes the record's JSON + // to protobuf bytes and queues them — with NO per-record wait. The + // single wait point is the flush() below. + std::vector encoded = schema.encode_json(make_order_json( + 1, "Alice Smith", "Wireless Mouse", 2, 25.99, "pending", now)); + std::int64_t offset = stream.ingest_proto_record(encoded); + std::cout << "Record 1 queued with offset ID: " << offset << "\n"; + + encoded = schema.encode_json(make_order_json( + 2, "Bob Johnson", "Mechanical Keyboard", 1, 89.99, "shipped", now)); + offset = stream.ingest_proto_record(encoded); + std::cout << "Record 2 queued with offset ID: " << offset << "\n"; + + encoded = schema.encode_json(make_order_json( + 3, "Carol Williams", "USB-C Hub", 3, 45.00, "delivered", now)); + offset = stream.ingest_proto_record(encoded); + std::cout << "Record 3 queued with offset ID: " << offset << "\n"; + + // 5. Flush once at the end: block until every queued record is durably + // acknowledged. This is the right place to wait — not after each ingest. + stream.flush(); + std::cout << "All records acknowledged.\n"; + + // 6. Close at a controlled point rather than leaving it to the destructor. + stream.close(); + std::cout << "Stream closed successfully.\n"; + } catch (const zerobus::ZerobusException& e) { + std::cerr << "Zerobus error: " << e.what() + << " (retryable=" << (e.is_retryable() ? "true" : "false") + << ")\n"; + return 1; + } + + return 0; +}