[Rust] Add support to fetch schema from UC for dynamic proto

rust/NEXT_CHANGELOG.md

@@ -6,6 +6,29 @@
 
 ### New Features and Improvements
 
+- Dynamic protobuf streams can resolve their schema from Unity Catalog, so a
+  runtime descriptor no longer has to be assembled by hand. Fetch the descriptor
+  from the live table metadata and pass it to the existing `.dynamic_proto(...)`
+  selector:
+
+  ```rust
+  let descriptor = sdk
+      .fetch_message_descriptor("catalog.schema.table", client_id, client_secret)
+      .await?;
+
+  let stream = sdk
+      .stream_builder()
+      .table("catalog.schema.table")
+      .oauth(client_id, client_secret)
+      .dynamic_proto(descriptor)
+      .build()
+      .await?;
+  ```
+
+  `ZerobusSdk::fetch_message_descriptor()` uses the SDK's configured
+  `unity_catalog_url`; the underlying `uc_schema` module takes the endpoint
+  directly. The fetch needs OAuth credentials able to read the table's metadata.
+
 ### Bug Fixes
 
 - Arrow Flight acknowledgment deadlines are pending-relative: no timer runs
@@ -37,3 +60,7 @@
 ### Deprecations
 
 ### API Changes
+
+- Added `ZerobusSdk::fetch_message_descriptor()`, the `uc_schema` module
+  (`fetch_message_descriptor`, `fetch_table_schema`), and the
+  `ZerobusError::SchemaFetchError` variant. All additive.

rust/README.md

@@ -573,6 +573,43 @@ stream.flush().await?; // wait once for all pending acknowledgments
 
 On the wire this is identical to `.compiled_proto(...)`; the difference is that records are built dynamically rather than from a generated struct. See the [`dynamic_proto`](https://docs.rs/databricks-zerobus-ingest-sdk/latest/databricks_zerobus_ingest_sdk/dynamic_proto/) module and the `proto_dynamic_single` example for details.
 
+##### Fetching the Schema from Unity Catalog
+
+Rather than assembling the columns yourself, let the SDK read the table's schema from Unity Catalog. `fetch_message_descriptor` resolves the descriptor from the live table metadata; pass it to `.dynamic_proto(...)` as usual:
+
+```rust
+// Fetch the descriptor once from Unity Catalog (uses the SDK's unity_catalog_url).
+let descriptor = sdk
+    .fetch_message_descriptor("catalog.schema.orders", client_id, client_secret)
+    .await?;
+
+// Inspect it if the columns are unknown to the program...
+for field in descriptor.fields() {
+    println!("{} ({:?})", field.name(), field.kind());
+}
+
+// ...then plug it into the ordinary dynamic-proto selector. Cloning a descriptor
+// is cheap (Arc-backed), so one fetch can serve many streams.
+let mut stream = sdk
+    .stream_builder().table("catalog.schema.orders")
+    .oauth(client_id, client_secret)
+    .dynamic_proto(descriptor)
+    .build()
+    .await?;
+
+// Records are built exactly as above — `new_record()` uses the fetched schema.
+for i in 0..100_000i64 {
+    let mut record = stream.new_record()?;
+    record.set("id", i)?.set("customer_name", "Alice Smith")?;
+    let _offset = stream.ingest_record_offset(ProtoBytes(record.encode()?)).await?; // queue only
+}
+stream.flush().await?; // wait once for all pending acknowledgments
+```
+
+The fetch needs OAuth credentials able to read the table's metadata (they are presented to the Unity Catalog REST API) and `unity_catalog_url` on the SDK builder. For direct control over the endpoint — outside an `SDK`, or against a different workspace — call `uc_schema::fetch_message_descriptor(unity_catalog_url, table, client_id, client_secret)`.
+
+The fetched schema is a snapshot. If the table changes afterwards, the server rejects stream creation with `ZerobusError::InvalidSchema`; re-fetch and rebuild to pick the change up. A failed fetch surfaces as `ZerobusError::SchemaFetchError`. Note that `DATE` and `TIMESTAMP` columns map to integers (days and microseconds since the Unix epoch) — see the [`schema`](https://docs.rs/databricks-zerobus-ingest-sdk/latest/databricks_zerobus_ingest_sdk/schema/) module for the full type mapping, and [`uc_schema`](https://docs.rs/databricks-zerobus-ingest-sdk/latest/databricks_zerobus_ingest_sdk/uc_schema/) for the fetch API.
+
 Setters can be called in any order. The builder validates at `build()` time that both authentication and format have been configured.
 
 ### 5. Ingest Data
@@ -1000,6 +1037,7 @@ The `examples/` directory contains four working examples covering different seri
 | `proto/compiled/batch.rs` | Protocol Buffers | Batch | `cargo run -p rust-examples-proto --example proto_compiled_batch` |
 | `proto/dynamic/single.rs` | Protocol Buffers (runtime schema) | Single-record | `cargo run -p rust-examples-proto --example proto_dynamic_single` |
 | `proto/dynamic/batch.rs` | Protocol Buffers (runtime schema) | Batch | `cargo run -p rust-examples-proto --example proto_dynamic_batch` |
+| `proto/dynamic/from_uc.rs` | Protocol Buffers (schema fetched from Unity Catalog) | Single-record | `cargo run -p rust-examples-proto --example proto_dynamic_from_uc` |
 
 
 Check [`examples/README.md`](https://github.com/databricks/zerobus-sdk/blob/main/rust/examples/README.md) for setup instructions and detailed comparisons.
@@ -1273,6 +1311,7 @@ cargo run -p rust-examples-proto --example proto_compiled_batch
 # Build and run Protocol Buffers dynamic-schema examples
 cargo run -p rust-examples-proto --example proto_dynamic_single
 cargo run -p rust-examples-proto --example proto_dynamic_batch
+cargo run -p rust-examples-proto --example proto_dynamic_from_uc
 ```
 
 ## Community and Contributing

rust/examples/README.md

@@ -42,6 +42,7 @@ The SDK supports two serialization formats and two ingestion methods:
 | [Proto Compiled Batch](proto/README.md#compiled-batch-example) | Protocol Buffers | Batch | `cargo run -p rust-examples-proto --example proto_compiled_batch` |
 | [Proto Dynamic](proto/README.md#dynamic-schema-example) | Protocol Buffers | Single-record (runtime schema) | `cargo run -p rust-examples-proto --example proto_dynamic_single` |
 | [Proto Dynamic Batch](proto/README.md#dynamic-batch) | Protocol Buffers | Batch (runtime schema) | `cargo run -p rust-examples-proto --example proto_dynamic_batch` |
+| [Proto Dynamic from UC](proto/README.md#dynamic-schema-from-unity-catalog) | Protocol Buffers | Single-record (schema fetched from Unity Catalog) | `cargo run -p rust-examples-proto --example proto_dynamic_from_uc` |
 | [Arrow](arrow/README.md) | Arrow Flight (Beta) | `RecordBatch` | `cargo run -p example_arrow` |
 
 ## Prerequisites

rust/examples/proto/Cargo.toml

@@ -16,6 +16,10 @@ path = "compiled/single.rs"
 name = "proto_dynamic_batch"
 path = "dynamic/batch.rs"
 
+[[example]]
+name = "proto_dynamic_from_uc"
+path = "dynamic/from_uc.rs"
+
 [[example]]
 name = "proto_dynamic_single"
 path = "dynamic/single.rs"

rust/examples/proto/README.md

@@ -17,6 +17,7 @@ This directory contains examples demonstrating Protocol Buffers-based data inges
   - [Running the Example](#running-the-example-2)
   - [Code Highlights](#code-highlights-2)
   - [Dynamic Batch](#dynamic-batch)
+  - [Dynamic Schema from Unity Catalog](#dynamic-schema-from-unity-catalog)
 - [Adapting for Your Custom Table](#adapting-for-your-custom-table)
   - [Generate Schema Files](#generate-schema-files)
   - [Update main.rs](#update-mainrs)
@@ -43,6 +44,7 @@ The examples are grouped by how the protobuf schema is obtained:
   are built field-by-field with `DynamicRecord`.
   - **`dynamic/single.rs`** - Build the descriptor in code and ingest dynamic records one at a time
   - **`dynamic/batch.rs`** - Ingest multiple dynamic records at once using `ingest_records_offset()`
+  - **`dynamic/from_uc.rs`** - Fetch the schema from Unity Catalog with `fetch_message_descriptor` and feed it to `.dynamic_proto(...)`, so no columns are hardcoded
 
 ## Three Ways to Pass Data
 
@@ -280,6 +282,57 @@ if let Some(offset) = stream.ingest_records_offset(batch).await? {
 stream.flush().await?;
 ```
 
+### Dynamic Schema from Unity Catalog
+
+`dynamic/from_uc.rs` goes one step further: instead of assembling the columns in code,
+`fetch_message_descriptor` reads the table's schema from Unity Catalog and the resolved
+descriptor is handed to the ordinary `.dynamic_proto(...)` selector. Nothing about the
+schema is hardcoded, so the same program works against any table the credentials can read:
+
+```bash
+cargo run -p rust-examples-proto --example proto_dynamic_from_uc
+```
+
+```rust
+// `unity_catalog_url` is required — it is where the schema is fetched from.
+let sdk = ZerobusSdk::builder()
+    .endpoint(SERVER_ENDPOINT)
+    .unity_catalog_url(DATABRICKS_WORKSPACE_URL)
+    .build()?;
+
+// Fetch the descriptor from the live table metadata.
+let descriptor = sdk
+    .fetch_message_descriptor(TABLE_NAME, DATABRICKS_CLIENT_ID, DATABRICKS_CLIENT_SECRET)
+    .await?;
+
+// The fetched schema can be inspected when the columns are genuinely unknown.
+for field in descriptor.fields() {
+    println!("  {} ({:?})", field.name(), field.kind());
+}
+
+// Plug it into the same builder used for a hand-built descriptor.
+let mut stream = sdk
+    .stream_builder()
+    .table(TABLE_NAME)
+    .oauth(DATABRICKS_CLIENT_ID, DATABRICKS_CLIENT_SECRET)
+    .dynamic_proto(descriptor)
+    .build()
+    .await?;
+
+// Records are built exactly as in the examples above.
+for i in 0..1_000i64 {
+    let mut record = stream.new_record()?;
+    record.set("id", i)?.set("customer_name", "Alice Smith")?;
+    stream.ingest_record_offset(ProtoBytes(record.encode()?)).await?; // queue only
+}
+stream.flush().await?; // wait once for all pending acks
+```
+
+This needs `.oauth(...)` credentials (they are presented to the Unity Catalog REST
+API) and `unity_catalog_url` on the SDK builder. `sdk.fetch_message_descriptor(...)`
+uses that configured URL; for direct control over the endpoint, call
+`uc_schema::fetch_message_descriptor(unity_catalog_url, table, client_id, client_secret)`.
+
 ## Adapting for Your Custom Table
 
 To use your own table, you need to generate schema files and update the example code.

rust/examples/proto/dynamic/from_uc.rs

@@ -0,0 +1,68 @@
+//! Dynamic protobuf ingestion with the schema fetched from Unity Catalog.
+//!
+//! Unlike `dynamic/single.rs`, which builds the descriptor in code, this fetches
+//! it with `fetch_message_descriptor` and feeds it to the usual `.dynamic_proto(...)`.
+//!
+//! Throughput: ingest in a loop, then `flush()` once — never wait per record.
+
+use std::error::Error;
+
+use databricks_zerobus_ingest_sdk::{ProtoBytes, ZerobusSdk};
+
+// Change constants to match your data.
+const TABLE_NAME: &str = "<your_table_name>";
+const DATABRICKS_CLIENT_ID: &str = "<your_databricks_client_id>";
+const DATABRICKS_CLIENT_SECRET: &str = "<your_databricks_client_secret>";
+
+// For AWS (for Azure, use *.azuredatabricks.net):
+const DATABRICKS_WORKSPACE_URL: &str = "https://<your-workspace>.cloud.databricks.com";
+const SERVER_ENDPOINT: &str = "https://<your-shard-id>.zerobus.<region>.cloud.databricks.com";
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    // `unity_catalog_url` is required: it is where the schema is fetched from.
+    let sdk = ZerobusSdk::builder()
+        .endpoint(SERVER_ENDPOINT)
+        .unity_catalog_url(DATABRICKS_WORKSPACE_URL)
+        .build()?;
+
+    // Descriptor from live table metadata — no columns or `.proto` needed up front.
+    let descriptor = sdk
+        .fetch_message_descriptor(TABLE_NAME, DATABRICKS_CLIENT_ID, DATABRICKS_CLIENT_SECRET)
+        .await?;
+    println!(
+        "Fetched schema '{}' with {} fields",
+        descriptor.name(),
+        descriptor.fields().count()
+    );
+
+    let mut stream = sdk
+        .stream_builder()
+        .table(TABLE_NAME)
+        .oauth(DATABRICKS_CLIENT_ID, DATABRICKS_CLIENT_SECRET)
+        .dynamic_proto(descriptor)
+        .build()
+        .await?;
+
+    // (customer_name, quantity, price) — adjust the field names to your table.
+    let orders = [("Alice Smith", 2i32, 25.99f64), ("Bob Johnson", 1, 89.99)];
+    for (i, (customer_name, quantity, price)) in orders.iter().enumerate() {
+        // set()'s value must match the column's proto type (BIGINT -> i64, INT -> i32).
+        let mut record = stream.new_record()?;
+        record
+            .set("id", i as i64)?
+            .set("customer_name", *customer_name)?
+            .set("quantity", *quantity)?
+            .set("price", *price)?;
+        // Queue without waiting for the ack.
+        stream
+            .ingest_record_offset(ProtoBytes(record.encode()?))
+            .await?;
+    }
+
+    stream.flush().await?; // wait once for all pending acks
+    stream.close().await?;
+    println!("Done");
+
+    Ok(())
+}

rust/sdk/src/dynamic_proto.rs

@@ -9,8 +9,10 @@
 //!
 //! Obtain the [`MessageDescriptor`] from [`message_descriptor`], which resolves a
 //! [`prost_types::DescriptorProto`] (built with
-//! [`crate::schema::descriptor_from_uc_columns`] or fetched from Unity Catalog),
-//! or from your own [`prost_reflect::DescriptorPool`].
+//! [`crate::schema::descriptor_from_uc_columns`]), or from your own
+//! [`prost_reflect::DescriptorPool`]. To read the schema straight from Unity
+//! Catalog instead, see [`crate::uc_schema`] (or
+//! [`ZerobusSdk::fetch_message_descriptor`](crate::ZerobusSdk::fetch_message_descriptor)).
 //!
 //! Ingest in a loop, then `flush()` once — never wait per record.
 //!

rust/sdk/src/errors.rs

@@ -131,6 +131,10 @@ pub enum ZerobusError {
     /// Returned when OAuth token fetching fails due to network or server errors.
     #[error("Token fetch failed: {0}")]
     TokenFetchError(String),
+    /// Returned when resolving a table's schema from Unity Catalog failed (see
+    /// [`crate::uc_schema`]).
+    #[error("Failed to fetch table schema from Unity Catalog: {0}.")]
+    SchemaFetchError(String),
 }
 
 /// List of gRPC status codes that indicate unretriable errors.
@@ -210,6 +214,8 @@ impl ZerobusError {
             ZerobusError::InvalidStateError(_) => false,
             ZerobusError::ConnectionTimeout(_) => true,
             ZerobusError::TokenFetchError(_) => true,
+            // A schema fetch is a one-shot setup step, not part of the recovery loop.
+            ZerobusError::SchemaFetchError(_) => false,
         }
     }
 

rust/sdk/src/lib.rs

@@ -57,6 +57,7 @@ mod stream_configuration;
 pub mod stream_options;
 mod tls_config;
 mod token_cache;
+pub mod uc_schema;
 
 pub use builder::{StreamBuilder, ZerobusSdkBuilder};
 pub use callbacks::AckCallback;

rust/sdk/src/sdk.rs

@@ -117,6 +117,49 @@ impl ZerobusSdk {
         StreamBuilder::new(self)
     }
 
+    /// Fetch `table_name`'s schema from Unity Catalog and resolve it to a
+    /// [`MessageDescriptor`](crate::MessageDescriptor), using this SDK's configured
+    /// `unity_catalog_url`. Pass the result to
+    /// [`dynamic_proto`](StreamBuilder::dynamic_proto). For direct control over the
+    /// endpoint, use [`uc_schema::fetch_message_descriptor`](crate::uc_schema::fetch_message_descriptor).
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// # use databricks_zerobus_ingest_sdk::ZerobusSdk;
+    /// # async fn example(sdk: &ZerobusSdk) -> Result<(), Box<dyn std::error::Error>> {
+    /// let descriptor = sdk
+    ///     .fetch_message_descriptor("catalog.schema.table", "client-id", "client-secret")
+    ///     .await?;
+    /// let stream = sdk
+    ///     .stream_builder()
+    ///     .table("catalog.schema.table")
+    ///     .oauth("client-id", "client-secret")
+    ///     .dynamic_proto(descriptor)
+    ///     .build()
+    ///     .await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// # Errors
+    ///
+    /// See [`uc_schema::fetch_message_descriptor`](crate::uc_schema::fetch_message_descriptor).
+    pub async fn fetch_message_descriptor(
+        &self,
+        table_name: &str,
+        client_id: &str,
+        client_secret: &str,
+    ) -> ZerobusResult<crate::MessageDescriptor> {
+        crate::uc_schema::fetch_message_descriptor(
+            &self.unity_catalog_url,
+            table_name,
+            client_id,
+            client_secret,
+        )
+        .await
+    }
+
     /// Creates a new SDK instance with explicit configuration.
     ///
     /// This is used internally by the builder pattern. `sdk_identifier` is the