Skip to content

Feat/dotnet arrow streams #541

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nalyd2 wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Feat/dotnet arrow streams #541

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
2655e9b
[.NET] Add Arrow Flight streams and ProtoSchema support
nalyd2 Jul 20, 2026
6305225
[.NET] Add StreamBuilder, Arrow async ops, and integration tests
nalyd2 Jul 20, 2026
8d44769
[.NET] Add ProtoSchema example and update README
nalyd2 Jul 20, 2026
8035166
[.NET] Add MockFlightServer for Arrow integration tests
nalyd2 Jul 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 21 additions & 22 deletions dotnet/Directory.Packages.props
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,24 +1,23 @@
<Project>
<!-- Central Package Management (CPM) - Microsoft Learn -->
<!-- https://learn.microsoft.com/en-us/nuget/consume-packages/central-package-management -->

<PropertyGroup>
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
</PropertyGroup>

<ItemGroup>
<!-- SDK Libraries -->
<PackageVersion Include="Microsoft.SourceLink.GitHub" Version="10.0.103" />

<!-- Testing -->
<PackageVersion Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
<PackageVersion Include="NUnit" Version="4.4.0" />
<PackageVersion Include="NUnit3TestAdapter" Version="5.2.0" />
<PackageVersion Include="NSubstitute" Version="5.3.0" />

<!-- gRPC and Protobuf -->
<PackageVersion Include="Grpc.AspNetCore" Version="2.76.0" />
<PackageVersion Include="Grpc.Tools" Version="2.78.0" />
<PackageVersion Include="Google.Protobuf" Version="3.33.5" />
</ItemGroup>
<!-- Central Package Management (CPM) - Microsoft Learn -->
<!-- https://learn.microsoft.com/en-us/nuget/consume-packages/central-package-management -->
<PropertyGroup>
<ManagePackageVersionsCentrally>true</ManagePackageVersionsCentrally>
</PropertyGroup>
<ItemGroup>
<!-- SDK Libraries -->
<PackageVersion Include="Apache.Arrow" Version="19.0.0" />
<PackageVersion Include="Microsoft.SourceLink.GitHub" Version="10.0.103" />
<!-- Testing -->
<PackageVersion Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
<PackageVersion Include="NUnit" Version="4.4.0" />
<PackageVersion Include="NUnit3TestAdapter" Version="5.2.0" />
<PackageVersion Include="NSubstitute" Version="5.3.0" />
<!-- gRPC and Protobuf -->
<PackageVersion Include="Grpc.AspNetCore" Version="2.76.0" />
<PackageVersion Include="Apache.Arrow.Flight" Version="19.0.0" />
<PackageVersion Include="Apache.Arrow.Flight.AspNetCore" Version="19.0.0" />
<PackageVersion Include="Grpc.Tools" Version="2.78.0" />
<PackageVersion Include="Google.Protobuf" Version="3.33.5" />
</ItemGroup>
</Project>
4 changes: 4 additions & 0 deletions dotnet/NEXT_CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,10 @@

### New Features and Improvements

- **Arrow Flight streams**: Added `ZerobusArrowStream` class for ingesting Apache Arrow IPC-encoded RecordBatches into Unity Catalog Delta tables. Includes `IngestBatch`, `WaitForOffset`, `Flush`, `Close`, and `GetUnackedBatches` APIs, with OAuth and `IHeadersProvider` authentication, and async variants (`IngestBatchAsync`, `WaitForOffsetAsync`, `FlushAsync`, `CloseAsync`, `GetUnackedBatchesAsync`). Configure via `ArrowStreamConfigurationOptions` record with IPC compression support (None, LZ4_FRAME, ZSTD). (#532, contributed by @nalyd2)
- **ProtoSchema**: Added `ProtoSchema` class for generating protobuf schemas from Unity Catalog table metadata. `ProtoSchema.FromUnityCatalogJson(ucJson)` creates a schema from the UC API response, `GetDescriptorBytes()` returns the compiled descriptor for stream creation, and `EncodeJson(json)` converts JSON records to protobuf bytes matching the table schema. (#532, contributed by @nalyd2)
- **Fluent StreamBuilder**: Added `StreamBuilder` fluent API as an alternative to factory methods on `ZerobusSdk`. Chain `.Table()`, `.OAuth()`, `.MaxInflightRequests()`, `.Recovery()` etc., then terminate with `.Json()`, `.CompiledProto(bytes)`, or `.Arrow(schema)` sub-builders. Both sync `Build()` and async `BuildAsync()` methods available. (#532, contributed by @nalyd2)

### Deprecations

### Bug Fixes
Expand Down
149 changes: 144 additions & 5 deletions dotnet/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@ High-performance .NET SDK for streaming data ingestion into Databricks Delta tab

## Quick Start

### Fluent StreamBuilder (recommended)

```csharp
using Databricks.Zerobus;

Expand All @@ -18,13 +20,31 @@ using var sdk = ZerobusSdk.CreateBuilder()
.UnityCatalogUrl("https://your-workspace.databricks.com")
.Build();

// 2. Create stream (fluent builder).
using var stream = sdk.StreamBuilder()
.Table("catalog.schema.table")
.OAuth(clientId, clientSecret)
.MaxInflightRequests(50_000)
.Json()
.Build();

// 3. Ingest records.
long offset = stream.IngestRecord("""{"id": 1, "message": "Hello"}""");

// 4. Wait for acknowledgment.
stream.WaitForOffset(offset);
```

### Factory methods (also supported)

```csharp
// 2. Configure stream options.
var options = StreamConfigurationOptions.Default with
{
MaxInflightRequests = 50_000,
};

// 3. Create stream.
// 3. Create stream (factory method).
using var stream = sdk.CreateJsonStream(
"catalog.schema.table",
clientId,
Expand Down Expand Up @@ -156,6 +176,83 @@ stream. The input `stream` is disposed during recreation and must not be used af
A later `Dispose()` on the original wrapper (for example at the end of a `using` scope)
is a no-op.

#### `CreateArrowStream`

Creates an Arrow Flight ingestion stream with OAuth 2.0 credentials. **(Beta)**

```csharp
using var stream = sdk.CreateArrowStream(
"catalog.schema.table",
schemaIpcBytes,
clientId,
clientSecret,
options); // ArrowStreamConfigurationOptions, optional

stream.IngestBatch(recordBatchIpcBytes);
stream.Flush();
```

Also available: `CreateArrowStreamAsync`, `CreateArrowStreamWithHeadersProvider`, `CreateArrowStreamWithHeadersProviderAsync`.

### `ProtoSchema`

Generates protobuf schemas from Unity Catalog table metadata. Useful for obtaining
descriptor bytes for `CreateProtoStream` without compiling `.proto` files manually.

```csharp
using var schema = ProtoSchema.FromUnityCatalogJson(ucTableJson);
byte[] descriptor = schema.GetDescriptorBytes();
byte[] protoBytes = schema.EncodeJson("""{"id": 1, "name": "test"}""");
```

### `StreamBuilder` (fluent API)

Alternative to factory methods — chain configuration then select the stream type.

```csharp
// JSON
using var stream = sdk.StreamBuilder()
.Table("catalog.schema.table")
.OAuth(clientId, clientSecret)
.MaxInflightRequests(50_000)
.Json()
.Build();

// Protobuf
using var stream = sdk.StreamBuilder()
.Table("catalog.schema.table")
.OAuth(clientId, clientSecret)
.CompiledProto(descriptorBytes)
.Build();

// Arrow Flight
using var stream = sdk.StreamBuilder()
.Table("catalog.schema.table")
.OAuth(clientId, clientSecret)
.Arrow(schemaIpcBytes)
.IpcCompression(IPCCompressionType.Lz4Frame)
.Build();
```

All sub-builders support `Build()` (sync) and `BuildAsync()` (async).

### `ZerobusArrowStream`

Arrow Flight ingestion stream for columnar data. Thread-safe, implements `IDisposable` and `IAsyncDisposable`.

```csharp
long offset = stream.IngestBatch(ipcBytes);
stream.WaitForOffset(offset);
stream.Flush();

// Async variants available
await stream.IngestBatchAsync(ipcBytes);
await stream.FlushAsync();

// Retrieve unacknowledged batches after close/failure
ArrowBatchInfo[] unacked = stream.GetUnackedBatches();
```

### `JsonZerobusStream` and `ProtoZerobusStream`

Typed stream wrappers expose only the matching ingest overloads while preserving the
Expand Down Expand Up @@ -297,6 +394,31 @@ var options = StreamConfigurationOptions.Default with
Typed factories set `RecordType` automatically. You only need to set it manually when using
the untyped `CreateStream` or `CreateStreamWithHeadersProvider` APIs.

### `ArrowStreamConfigurationOptions`

Use C# record `with` expressions to configure Arrow Flight streams:

```csharp
var options = ArrowStreamConfigurationOptions.Default with
{
MaxInflightBatches = 5_000,
IpcCompression = IPCCompressionType.Zstd,
};
```

| Property | Default | Description |
| ---------------------------- | ------- | -------------------------------------- |
| `MaxInflightBatches` | 1,000 | Backpressure control for batches |
| `Recovery` | `true` | Auto-recovery on failures |
| `RecoveryTimeoutMs` | 15,000 | Timeout per recovery attempt |
| `RecoveryBackoffMs` | 2,000 | Delay between retries |
| `RecoveryRetries` | 4 | Max recovery attempts |
| `ServerLackOfAckTimeoutMs` | 60,000 | Server ack timeout |
| `FlushTimeoutMs` | 300,000 | Flush timeout (5 min) |
| `ConnectionTimeoutMs` | 30,000 | Connection timeout |
| `IpcCompression` | `None` | IPC compression (`None`, `Lz4Frame`, `Zstd`) |
| `StreamPausedMaxWaitTimeMs` | -1 | Graceful close wait (-1 = full server duration) |

### Error Handling

Errors throw `ZerobusException` with an `IsRetryable` property:
Expand Down Expand Up @@ -382,6 +504,10 @@ The integration tests cover:
| `IngestMultipleRecords` | Multiple sequential records with ack |
| `IngestBatchRecords` | Batch ingest of 5 records |
| `IngestRecordsAfterClose` | Batch ingest after close throws |
| `ArrowStream_CreateAndDispose_DoesNotLeak` | Arrow stream creation and disposal |
| `ArrowStream_Close_AndDispose_NoError` | Arrow stream close then dispose |
| `ArrowStream_MultipleBatches_FlushThenClose` | Arrow batch ingest, flush, and close |
| `StreamBuilder_Arrow_BuildsWithSchema` | Fluent Arrow stream via StreamBuilder |

Each test gets its own mock gRPC server on a unique port, so all tests run in parallel.

Expand All @@ -402,12 +528,18 @@ dotnet/
├── src/
│ └── Zerobus/ # Main SDK library
│ ├── Zerobus.csproj
│ ├── ZerobusSdk.cs # SDK entry point (IDisposable)
│ ├── ZerobusStream.cs # Stream for record ingestion (IDisposable)
│ ├── ZerobusSdk.cs # SDK entry point + Arrow stream factories
│ ├── ZerobusStream.cs # Streaming for JSON/Proto records
│ ├── ZerobusArrowStream.cs # Arrow Flight ingestion stream (Beta)
│ ├── ZerobusException.cs # Error type with IsRetryable
│ ├── IHeadersProvider.cs # Custom auth interface
│ ├── RecordType.cs # Proto / Json / Unspecified enum
│ ├── StreamConfigurationOptions.cs # Config record with defaults
│ ├── StreamConfigurationOptions.cs # Config record for JSON/Proto streams
│ ├── ArrowStreamConfigurationOptions.cs # Config record for Arrow Flight streams
│ ├── IPCCompressionType.cs # Arrow IPC compression enum
│ ├── ArrowBatchInfo.cs # Unacked Arrow batch record
│ ├── ProtoSchema.cs # Protobuf schema from Unity Catalog
│ ├── StreamBuilder.cs # Fluent builder API
│ ├── TableProperties.cs # Table name + optional descriptor
│ ├── Properties/
│ │ └── AssemblyInfo.cs
Expand All @@ -417,9 +549,15 @@ dotnet/
│ └── HeadersProviderBridge.cs # Managed→native callback bridge
├── tests/
│ ├── Zerobus.Tests/ # Unit tests (NUnit)
│ │ ├── StreamBuilderTests.cs # Fluent builder validation
│ │ ├── ArrowStreamConfigurationOptionsTests.cs
│ │ └── ProtoSchemaTests.cs
│ └── Zerobus.IntegrationTests/ # Integration tests (NUnit + gRPC mock)
│ ├── Zerobus.IntegrationTests.csproj
│ ├── IntegrationTests.cs # 11 integration tests
│ ├── StreamCreationIntegrationTests.cs # Stream creation scenarios
│ ├── IngestionIntegrationTests.cs # Ingestion scenarios
│ ├── ArrowIntegrationTests.cs # Arrow Flight integration tests
│ ├── LifecycleRecoveryIntegrationTests.cs
│ ├── MockZerobusServer.cs # Mock gRPC server
│ ├── TestHelpers.cs # Fixtures, response builders, interceptor
│ └── Protos/
Expand All @@ -428,6 +566,7 @@ dotnet/
├── JsonSingle/ # Single JSON record ingestion
├── JsonBatch/ # Batch JSON record ingestion
├── ProtoSingle/ # Single protobuf record ingestion
└── ProtoSchema/ # ProtoSchema: UC → descriptor + JSON→proto
```

## Architecture
Expand Down
Loading