---
sidebar_position: 2
---
# API Reference

Complete API reference for the Fluss Rust client.

## `Config`

| Field                                 | Type            | Default            | Description                                                                          |
|---------------------------------------|-----------------|--------------------|--------------------------------------------------------------------------------------|
| `bootstrap_servers`                   | `String`        | `"127.0.0.1:9123"` | Coordinator server address                                                           |
| `writer_request_max_size`             | `i32`           | `10485760` (10 MB) | Maximum request size in bytes                                                        |
| `writer_acks`                         | `String`        | `"all"`            | Acknowledgment setting (`"all"` waits for all replicas)                              |
| `writer_retries`                      | `i32`           | `i32::MAX`         | Number of retries on failure                                                         |
| `writer_batch_size`                   | `i32`           | `2097152` (2 MB)   | Batch size for writes in bytes. Upper bound when dynamic sizing is on; fixed batch size when off. |
| `writer_dynamic_batch_size_enabled`   | `bool`          | `true`             | Enable per-table dynamic batch sizing: target grows 10% above 80% fill, shrinks 5% below 50%, clamped to `[writer_dynamic_batch_size_min, writer_batch_size]` |
| `writer_dynamic_batch_size_min`       | `i32`           | `262144` (256 KB)  | Lower bound for the dynamic batch size estimator (ignored when `writer_dynamic_batch_size_enabled` is `false`) |
| `writer_batch_timeout_ms`             | `i64`           | `100`              | Maximum time in ms to wait for a writer batch to fill up before sending              |
| `writer_bucket_no_key_assigner`       | `NoKeyAssigner` | `sticky`           | Bucket assignment strategy for tables without bucket keys: `sticky` or `round_robin` |
| `scanner_remote_log_prefetch_num`     | `usize`         | `4`                | Number of remote log segments to prefetch                                            |
| `remote_file_download_thread_num`     | `usize`         | `3`                | Number of threads for remote log downloads                                           |
| `scanner_remote_log_read_concurrency` | `usize`         | `4`                | Streaming read concurrency within a remote log file                                  |
| `scanner_log_max_poll_records`        | `usize`         | `500`              | Maximum number of records returned in a single poll()                                |
| `scanner_log_fetch_max_bytes`         | `i32`           | `16777216` (16 MB) | Maximum bytes per fetch response for LogScanner                                      |
| `scanner_log_fetch_min_bytes`         | `i32`           | `1`                | Minimum bytes the server must accumulate before returning a fetch response           |
| `scanner_log_fetch_wait_max_time_ms`  | `i32`           | `500`              | Maximum time (ms) the server may wait to satisfy min-bytes                           |
| `scanner_log_fetch_max_bytes_for_bucket`| `i32`         | `1048576` (1 MB)   | Maximum bytes per fetch response per bucket for LogScanner                           |
| `connect_timeout_ms`                  | `u64`           | `120000`           | TCP connect timeout in milliseconds                                                  |
| `security_protocol`                   | `String`        | `"PLAINTEXT"`      | `PLAINTEXT` (default) or `sasl` for SASL auth                                        |
| `security_sasl_mechanism`             | `String`        | `"PLAIN"`          | SASL mechanism (only `PLAIN` is supported)                                           |
| `security_sasl_username`              | `String`        | (empty)            | SASL username (required when protocol is `sasl`)                                     |
| `security_sasl_password`              | `String`        | (empty)            | SASL password (required when protocol is `sasl`)                                     |

## `FlussConnection`

| Method                                                                        | Description                                    |
|-------------------------------------------------------------------------------|------------------------------------------------|
| `async fn new(config: Config) -> Result<Self>`                                | Create a new connection to a Fluss cluster     |
| `fn get_admin(&self) -> Result<Arc<FlussAdmin>>`                              | Get the admin interface for cluster management |
| `async fn get_table(&self, table_path: &TablePath) -> Result<FlussTable<'_>>` | Get a table for read/write operations          |
| `fn config(&self) -> &Config`                                                 | Get a reference to the connection config       |

## `FlussAdmin`

### Database Operations

| Method                                                                                                                       | Description                |
|------------------------------------------------------------------------------------------------------------------------------|----------------------------|
| `async fn create_database(&self, name: &str, descriptor: Option<&DatabaseDescriptor>, ignore_if_exists: bool) -> Result<()>` | Create a database          |
| `async fn drop_database(&self, name: &str, ignore_if_not_exists: bool, cascade: bool) -> Result<()>`                         | Drop a database            |
| `async fn list_databases(&self) -> Result<Vec<String>>`                                                                      | List all databases         |
| `async fn database_exists(&self, name: &str) -> Result<bool>`                                                                | Check if a database exists |
| `async fn get_database_info(&self, name: &str) -> Result<DatabaseInfo>`                                                      | Get database metadata      |

### Table Operations

| Method                                                                                                                     | Description                                                                 |
|----------------------------------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------|
| `async fn create_table(&self, table_path: &TablePath, descriptor: &TableDescriptor, ignore_if_exists: bool) -> Result<()>` | Create a table                                                              |
| `async fn drop_table(&self, table_path: &TablePath, ignore_if_not_exists: bool) -> Result<()>`                             | Drop a table                                                                |
| `async fn get_table_info(&self, table_path: &TablePath) -> Result<TableInfo>`                                              | Get table metadata                                                          |
| `async fn get_table_schema(&self, table_path: &TablePath, schema_id: Option<i32>) -> Result<SchemaInfo>`                   | Get a table's schema by id, or the latest schema when `schema_id` is `None` |
| `async fn list_tables(&self, database_name: &str) -> Result<Vec<String>>`                                                  | List tables in a database                                                   |
| `async fn table_exists(&self, table_path: &TablePath) -> Result<bool>`                                                     | Check if a table exists                                                     |

### Partition Operations

| Method                                                                                                                               | Description                     |
|--------------------------------------------------------------------------------------------------------------------------------------|---------------------------------|
| `async fn list_partition_infos(&self, table_path: &TablePath) -> Result<Vec<PartitionInfo>>`                                         | List all partitions             |
| `async fn list_partition_infos_with_spec(&self, table_path: &TablePath, spec: Option<&PartitionSpec>) -> Result<Vec<PartitionInfo>>` | List partitions matching a spec |
| `async fn create_partition(&self, table_path: &TablePath, spec: &PartitionSpec, ignore_if_exists: bool) -> Result<()>`               | Create a partition              |
| `async fn drop_partition(&self, table_path: &TablePath, spec: &PartitionSpec, ignore_if_not_exists: bool) -> Result<()>`             | Drop a partition                |

### Offset Operations

| Method                                                                                                                                                           |  Description                          |
|------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------------------------|
| `async fn list_offsets(&self, table_path: &TablePath, bucket_ids: &[i32], offset_spec: OffsetSpec) -> Result<HashMap<i32, i64>>`                                 | Get offsets for buckets               |
| `async fn list_partition_offsets(&self, table_path: &TablePath, partition_name: &str, bucket_ids: &[i32], offset_spec: OffsetSpec) -> Result<HashMap<i32, i64>>` | Get offsets for a partition's buckets |

### Lake Operations

| Method                                                                                     |  Description                 |
|--------------------------------------------------------------------------------------------|------------------------------|
| `async fn get_latest_lake_snapshot(&self, table_path: &TablePath) -> Result<LakeSnapshot>` | Get the latest lake snapshot |

### Cluster Operations

| Method                                                        | Description                                         |
|---------------------------------------------------------------|-----------------------------------------------------|
| `async fn get_server_nodes(&self) -> Result<Vec<ServerNode>>` | Get all alive server nodes (coordinator + tablets)  |

## `ServerNode`

| Method                            | Description                                          |
|-----------------------------------|------------------------------------------------------|
| `fn id(&self) -> i32`            | Server node ID                                       |
| `fn host(&self) -> &str`         | Hostname of the server                               |
| `fn port(&self) -> u32`          | Port number                                          |
| `fn server_type(&self) -> &ServerType` | Server type (`CoordinatorServer`, `TabletServer`, or `Unknown`) |
| `fn uid(&self) -> &str`          | Unique identifier (e.g. `"cs-0"`, `"ts-1"`)         |

`ServerType::Unknown` is used when the client has not yet determined the endpoint
type, such as bootstrap endpoints.

## `FlussTable<'a>`

| Method                                        | Description                             |
|-----------------------------------------------|-----------------------------------------|
| `fn get_table_info(&self) -> &TableInfo`      | Get table metadata                      |
| `fn new_append(&self) -> Result<TableAppend>` | Create an append builder for log tables |
| `fn new_scan(&self) -> TableScan<'_>`         | Create a scan builder                   |
| `fn new_lookup(&self) -> Result<TableLookup>` | Create a lookup builder for PK tables   |
| `fn new_upsert(&self) -> Result<TableUpsert>` | Create an upsert builder for PK tables  |
| `fn has_primary_key(&self) -> bool`           | Check if the table has a primary key    |
| `fn table_path(&self) -> &TablePath`          | Get the table path                      |

## `TableAppend`

| Method                                            | Description             |
|---------------------------------------------------|-------------------------|
| `fn create_writer(&self) -> Result<AppendWriter>` | Create an append writer |

## `AppendWriter`

| Method                                                                          | Description                                       |
|---------------------------------------------------------------------------------|---------------------------------------------------|
| `fn append(&self, row: &impl InternalRow) -> Result<WriteResultFuture>`         | Append a row; returns a future for acknowledgment |
| `fn append_arrow_batch(&self, batch: RecordBatch) -> Result<WriteResultFuture>` | Append an Arrow RecordBatch                       |
| `async fn flush(&self) -> Result<()>`                                           | Flush all pending writes to the server            |

The writer pipeline emits `fluss.client.writer.*` metrics (send latency, batch
queue time, records/bytes sent, retries, per-batch size, and buffer-pool
gauges). Unlike scanner metrics, these are **unlabeled** (global per process),
matching Java's single `WriterMetricGroup` per client (Java scopes only by
`client_id`, which the Rust `metrics` facade has no concept of yet). The same
series are shared by `AppendWriter` (log tables) and `UpsertWriter` (PK tables).

## `TableScan<'a>`

| Method                                                                      | Description                             |
|-----------------------------------------------------------------------------|-----------------------------------------|
| `fn project(self, indices: &[usize]) -> Result<Self>`                       | Project columns by index                |
| `fn project_by_name(self, names: &[&str]) -> Result<Self>`                  | Project columns by name                 |
| `fn limit(self, n: i32) -> Result<Self>`                                    | Set a row limit (enables `create_bucket_batch_scanner`; rejected by log scanners) |
| `fn create_log_scanner(self) -> Result<LogScanner>`                         | Create a record-based log scanner; on a primary-key table, subscribes to its CDC changelog (per-record `ChangeType`) |
| `fn create_record_batch_log_scanner(self) -> Result<RecordBatchLogScanner>` | Create an Arrow batch-based log scanner (log tables only — no per-record change types) |
| `fn create_bucket_batch_scanner(self, bucket: TableBucket) -> Result<LimitBatchScanner>` | Bounded scan of one bucket (requires `limit`; runs on first `next_batch`) |

## `LogScanner`

Single-consumer: do not call `poll` concurrently on the same scanner (e.g. from `tokio::join!` or two tasks sharing an `Arc`). Mirrors Java's `LogScannerImpl.acquire()` guard. Debug builds surface overlapping calls via a `debug_assert!`; release builds skip the check for performance and produce skewed poll-timing metrics (`fluss.client.scanner.time_between_poll_ms`, `fluss.client.scanner.poll_idle_ratio`) if the contract is violated.

All `fluss.client.scanner.*` metrics carry `database` and `table` labels (matching Java's per-`TablePath` `ScannerMetricGroup`), so multi-table consumers get one time series per scanned table.

| Method                                                                                                    | Description                                              |
|-----------------------------------------------------------------------------------------------------------|----------------------------------------------------------|
| `async fn subscribe(&self, bucket_id: i32, start_offset: i64) -> Result<()>`                              | Subscribe to a bucket                                    |
| `async fn subscribe_buckets(&self, bucket_offsets: &HashMap<i32, i64>) -> Result<()>`                     | Subscribe to multiple buckets                            |
| `async fn subscribe_partition(&self, partition_id: i64, bucket_id: i32, start_offset: i64) -> Result<()>` | Subscribe to a partition bucket                          |
| `async fn subscribe_partition_buckets(&self, offsets: &HashMap<(i64, i32), i64>) -> Result<()>`           | Subscribe to multiple partition-bucket pairs             |
| `async fn unsubscribe(&self, bucket_id: i32) -> Result<()>`                                               | Unsubscribe from a bucket (non-partitioned tables)       |
| `async fn unsubscribe_partition(&self, partition_id: i64, bucket_id: i32) -> Result<()>`                  | Unsubscribe from a partition bucket (partitioned tables) |
| `async fn poll(&self, timeout: Duration) -> Result<ScanRecords>`                                          | Poll for records                                         |

## `RecordBatchLogScanner`

Single-consumer: overlapping `poll` calls on handles that share state, or `poll` concurrent with `RecordBatchLogReader::next_batch`, are not supported — use one active polling/consumption call at a time per underlying scanner state. Mirrors Java's `LogScannerImpl.acquire()` guard. Debug builds surface overlapping calls via a `debug_assert!`; release builds skip the check for performance and produce skewed poll-timing metrics (`fluss.client.scanner.time_between_poll_ms`, `fluss.client.scanner.poll_idle_ratio`) if the contract is violated.

| Method                                                                                                    | Description                                              |
|-----------------------------------------------------------------------------------------------------------|----------------------------------------------------------|
| `async fn subscribe(&self, bucket_id: i32, start_offset: i64) -> Result<()>`                              | Subscribe to a bucket                                    |
| `async fn subscribe_buckets(&self, bucket_offsets: &HashMap<i32, i64>) -> Result<()>`                     | Subscribe to multiple buckets                            |
| `async fn subscribe_partition(&self, partition_id: i64, bucket_id: i32, start_offset: i64) -> Result<()>` | Subscribe to a partition bucket                          |
| `async fn subscribe_partition_buckets(&self, offsets: &HashMap<(i64, i32), i64>) -> Result<()>`           | Subscribe to multiple partition-bucket pairs             |
| `async fn unsubscribe(&self, bucket_id: i32) -> Result<()>`                                               | Unsubscribe from a bucket (non-partitioned tables)       |
| `async fn unsubscribe_partition(&self, partition_id: i64, bucket_id: i32) -> Result<()>`                  | Unsubscribe from a partition bucket (partitioned tables) |
| `async fn poll(&self, timeout: Duration) -> Result<Vec<ScanBatch>>`                                       | Poll for Arrow record batches                            |
| `fn is_partitioned(&self) -> bool`                                                                        | Check if the table is partitioned                        |
| `fn get_subscribed_buckets(&self) -> Vec<(TableBucket, i64)>`                                             | Get all current subscriptions as (bucket, offset) pairs  |
| `fn schema(&self) -> SchemaRef`                                                                           | Get the Arrow schema for batches produced by this scanner|
| `fn table_path(&self) -> &TablePath`                                                                      | Get the table path                                       |
| `fn table_id(&self) -> TableId`                                                                           | Get the table ID                                         |

## `RecordBatchLogReader`

Bounded log reader that consumes data up to specified stopping offsets, then terminates.
Unlike `RecordBatchLogScanner` which polls indefinitely, this reader stops automatically.

| Method                                                                                                      | Description                                              |
|-------------------------------------------------------------------------------------------------------------|----------------------------------------------------------|
| `async fn new_until_latest(scanner: RecordBatchLogScanner, admin: &FlussAdmin) -> Result<Self>`              | Read until the latest offsets at time of creation         |
| `fn new_until_offsets(scanner: RecordBatchLogScanner, stopping_offsets: HashMap<TableBucket, i64>) -> Result<Self>` | Read until custom stopping offsets per bucket             |
| `async fn next_batch(&mut self) -> Result<Option<ScanBatch>>`                                                | Get the next batch with bucket/offset metadata, or `None` when all buckets caught up |
| `async fn collect_all_batches(&mut self) -> Result<Vec<ScanBatch>>`                                          | Drain all batches (with metadata) until stopping offsets are satisfied |
| `fn schema(&self) -> SchemaRef`                                                                              | Arrow schema for produced batches                        |
| `fn to_record_batch_reader(self, handle: tokio::runtime::Handle) -> SyncRecordBatchLogReader`                | Sync adapter implementing `arrow::RecordBatchReader` (see below) |

## `SyncRecordBatchLogReader`

Synchronous adapter for `RecordBatchLogReader`. Created via
`RecordBatchLogReader::to_record_batch_reader(handle)`.

Implements both [`Iterator`] and [`arrow::record_batch::RecordBatchReader`], so it
plugs into the wider Arrow ecosystem — FFI, PyArrow's
`pa.RecordBatchReader.from_batches`, the C++ Arrow `RecordBatchReader` interface,
DataFusion sources, etc.

Each `next()` call drives the underlying async reader via
`tokio::runtime::Handle::block_on`. **Do not call from inside a Tokio worker
thread that belongs to the same runtime** — nested `block_on` panics. Prefer
`RecordBatchLogReader::next_batch` in async Rust code; use this adapter only at
sync/FFI boundaries.

Bucket and offset metadata carried by `ScanBatch` is **dropped** here, because
the Arrow trait contract yields plain `RecordBatch`. If you need offsets or
bucket identity per batch, use `next_batch` instead.

| Method                                                          | Description                                      |
|-----------------------------------------------------------------|--------------------------------------------------|
| `fn next(&mut self) -> Option<Result<RecordBatch, ArrowError>>` | Iterator: next batch, or `None` when caught up   |
| `fn schema(&self) -> SchemaRef`                                 | Arrow schema for produced batches                |

## `LimitBatchScanner`

One-shot bounded scanner from `TableScan::limit(n).create_bucket_batch_scanner(bucket)`.
Poll it with `next_batch` until it returns `None` (mirrors `RecordBatchLogReader`).
Supports both log and primary-key tables (the latter returns the current,
server-deduplicated state); yields a single batch of at most `n` rows.

| Method                                                        | Description                          |
|---------------------------------------------------------------|--------------------------------------|
| `async fn next_batch(&mut self) -> Result<Option<ScanBatch>>` | Rows on the first call, `None` after |
| `async fn collect_all_batches(&mut self) -> Result<Vec<ScanBatch>>` | Drain into all batches         |
| `fn bucket(&self) -> &TableBucket`                            | The scanned bucket                   |

## `ScanRecord`

| Method                                 | Description                            |
|----------------------------------------|----------------------------------------|
| `fn row(&self) -> &dyn InternalRow`    | Get the row data                       |
| `fn offset(&self) -> i64`              | Record offset in the log               |
| `fn timestamp(&self) -> i64`           | Record timestamp                       |
| `fn change_type(&self) -> &ChangeType` | Change type (AppendOnly, Insert, etc.) |

## `ChangeType`

The kind of change a `ScanRecord` represents. Log (append-only) tables always
produce `AppendOnly`; scanning a primary-key table's changelog produces the CDC
variants below.

| Variant        | Symbol | Meaning                                     |
|----------------|--------|---------------------------------------------|
| `AppendOnly`   | `+A`   | Append-only record (log tables)             |
| `Insert`       | `+I`   | A new primary key was inserted              |
| `UpdateBefore` | `-U`   | Row image *before* an update                |
| `UpdateAfter`  | `+U`   | Row image *after* an update                 |
| `Delete`       | `-D`   | A primary key was deleted (carries old row) |

With the default `FULL` changelog mode (`table.changelog.image`), updating an
existing key emits `-U` then `+U`; the `WAL` mode emits only `+U`.

## `ScanRecords`

| Method                                                                   | Description                       |
|--------------------------------------------------------------------------|-----------------------------------|
| `fn count(&self) -> usize`                                               | Number of records                 |
| `fn is_empty(&self) -> bool`                                             | Whether the result set is empty   |
| `fn records(&self, bucket: &TableBucket) -> &[ScanRecord]`               | Get records for a specific bucket |
| `fn records_by_buckets(&self) -> &HashMap<TableBucket, Vec<ScanRecord>>` | Get all records grouped by bucket |

`ScanRecords` also implements `IntoIterator`, so you can iterate over all records directly:

```rust
for record in records {
    println!("offset={}", record.offset());
}
```

## `ScanBatch`

| Method                             | Description                    |
|------------------------------------|--------------------------------|
| `fn bucket(&self) -> &TableBucket` | Bucket this batch belongs to   |
| `fn batch(&self) -> &RecordBatch`  | Arrow RecordBatch data         |
| `fn base_offset(&self) -> i64`     | First record offset            |
| `fn last_offset(&self) -> i64`     | Last record offset             |
| `fn num_records(&self) -> usize`   | Number of records in the batch |

## `TableUpsert`

| Method                                                                                | Description                                       |
|---------------------------------------------------------------------------------------|---------------------------------------------------|
| `fn create_writer(&self) -> Result<UpsertWriter>`                                     | Create an upsert writer                           |
| `fn partial_update(&self, column_indices: Option<Vec<usize>>) -> Result<TableUpsert>` | Create a partial update builder by column indices |
| `fn partial_update_with_column_names(&self, names: &[&str]) -> Result<TableUpsert>`   | Create a partial update builder by column names   |

## `UpsertWriter`

| Method                                                                  | Description                           |
|-------------------------------------------------------------------------|---------------------------------------|
| `fn upsert(&self, row: &impl InternalRow) -> Result<WriteResultFuture>` | Upsert a row (insert or update by PK) |
| `fn delete(&self, row: &impl InternalRow) -> Result<WriteResultFuture>` | Delete a row by primary key           |
| `async fn flush(&self) -> Result<()>`                                   | Flush all pending operations          |

## `TableLookup`

| Method                                          |  Description                        |
|-------------------------------------------------|-------------------------------------|
| `fn create_lookuper(&self) -> Result<Lookuper>` | Create a lookuper for point lookups |

## `Lookuper`

| Method                                                                       |  Description                |
|------------------------------------------------------------------------------|-----------------------------|
| `async fn lookup(&mut self, key: &impl InternalRow) -> Result<LookupResult>` | Lookup a row by primary key |

## `LookupResult`

| Method                                                         |  Description                     |
|----------------------------------------------------------------|----------------------------------|
| `fn get_single_row(&self) -> Result<Option<impl InternalRow>>` | Get a single row from the result |
| `fn get_rows(&self) -> Result<Vec<impl InternalRow>>`          | Get all rows from the result     |
| `fn to_record_batch(&self) -> Result<RecordBatch>`             | Convert all rows to an Arrow `RecordBatch` for DataFusion or other Arrow-based tools    |

## `WriteResultFuture`

| Description                                                                                                                                   |
|-----------------------------------------------------------------------------------------------------------------------------------------------|
| Implements `Future<Output = Result<(), Error>>`. Await to wait for server acknowledgment. Returned by `append()`, `upsert()`, and `delete()`. |

Usage:

```rust
// Fire-and-forget (batched)
writer.append(&row)?;
writer.flush().await?;

// Per-record acknowledgment
writer.append(&row)?.await?;
```

## `Schema`

| Method                                         |  Description                             |
|------------------------------------------------|------------------------------------------|
| `fn builder() -> SchemaBuilder`                | Create a schema builder                  |
| `fn columns(&self) -> &[Column]`               | Get all columns                          |
| `fn primary_key(&self) -> Option<&PrimaryKey>` | Get primary key (None if no primary key) |
| `fn column_names(&self) -> Vec<&str>`          | Get all column names                     |
| `fn primary_key_indexes(&self) -> Vec<usize>`  | Get primary key column indices           |

## `SchemaBuilder`

| Method                                               |  Description            |
|------------------------------------------------------|-------------------------|
| `fn column(name: &str, data_type: DataType) -> Self` | Add a column            |
| `fn primary_key(keys: Vec<&str>) -> Self`            | Set primary key columns |
| `fn build() -> Result<Schema>`                       | Build the schema        |

## `SchemaInfo`

A schema together with its server-assigned version id. Returned by [`FlussAdmin::get_table_schema`](#flussadmin).

| Method                                           | Description                              |
|--------------------------------------------------|------------------------------------------|
| `fn new(schema: Schema, schema_id: i32) -> Self` | Construct from a schema and id           |
| `fn schema(&self) -> &Schema`                    | Borrow the schema                        |
| `fn schema_id(&self) -> i32`                     | Get the server-assigned schema id        |
| `fn into_parts(self) -> (Schema, i32)`           | Consume and return `(schema, schema_id)` |

## `TableDescriptor`

| Method                                                    | Description                          |
|-----------------------------------------------------------|--------------------------------------|
| `fn builder() -> TableDescriptorBuilder`                  | Create a table descriptor builder    |
| `fn schema(&self) -> &Schema`                             | Get the table schema                 |
| `fn partition_keys(&self) -> &[String]`                   | Get partition key column names       |
| `fn has_primary_key(&self) -> bool`                       | Check if the table has a primary key |
| `fn properties(&self) -> &HashMap<String, String>`        | Get all table properties             |
| `fn custom_properties(&self) -> &HashMap<String, String>` | Get custom properties                |
| `fn comment(&self) -> Option<&str>`                       | Get table comment                    |

## `TableDescriptorBuilder`

| Method                                                                                    | Description                                 |
|-------------------------------------------------------------------------------------------|---------------------------------------------|
| `fn schema(schema: Schema) -> Self`                                                       | Set the schema                              |
| `fn log_format(format: LogFormat) -> Self`                                                | Set log format (e.g., `LogFormat::ARROW`)   |
| `fn kv_format(format: KvFormat) -> Self`                                                  | Set KV format (e.g., `KvFormat::COMPACTED`) |
| `fn property(key: &str, value: &str) -> Self`                                             | Set a table property                        |
| `fn custom_property(key: impl Into<String>, value: impl Into<String>) -> Self`            | Set a single custom property                |
| `fn custom_properties(properties: HashMap<impl Into<String>, impl Into<String>>) -> Self` | Set custom properties                       |
| `fn partitioned_by(keys: Vec<&str>) -> Self`                                              | Set partition columns                       |
| `fn distributed_by(bucket_count: Option<i32>, bucket_keys: Vec<String>) -> Self`          | Set bucket distribution                     |
| `fn comment(comment: &str) -> Self`                                                       | Set table comment                           |
| `fn build() -> Result<TableDescriptor>`                                                   | Build the table descriptor                  |

## `TablePath`

| Method                                                |  Description        |
|-------------------------------------------------------|---------------------|
| `TablePath::new(database: &str, table: &str) -> Self` | Create a table path |
| `fn database(&self) -> &str`                          | Get database name   |
| `fn table(&self) -> &str`                             | Get table name      |

## `TableInfo`

| Field / Method       | Description                                         |
|----------------------|-----------------------------------------------------|
| `.table_path`        | `TablePath` -- Table path                           |
| `.table_id`          | `i64` -- Table ID                                   |
| `.schema_id`         | `i32` -- Schema ID                                  |
| `.schema`            | `Schema` -- Table schema                            |
| `.primary_keys`      | `Vec<String>` -- Primary key column names           |
| `.partition_keys`    | `Vec<String>` -- Partition key column names         |
| `.num_buckets`       | `i32` -- Number of buckets                          |
| `.properties`        | `HashMap<String, String>` -- All table properties   |
| `.custom_properties` | `HashMap<String, String>` -- Custom properties only |
| `.comment`           | `Option<String>` -- Table comment                   |
| `.created_time`      | `i64` -- Creation timestamp                         |
| `.modified_time`     | `i64` -- Last modification timestamp                |

## `TableBucket`

| Method                                                                                              | Description                                |
|-----------------------------------------------------------------------------------------------------|--------------------------------------------|
| `TableBucket::new(table_id: i64, bucket_id: i32) -> Self`                                           | Create a non-partitioned bucket            |
| `TableBucket::new_with_partition(table_id: i64, partition_id: Option<i64>, bucket_id: i32) -> Self` | Create a partitioned bucket                |
| `fn table_id(&self) -> i64`                                                                         | Get table ID                               |
| `fn partition_id(&self) -> Option<i64>`                                                             | Get partition ID (None if non-partitioned) |
| `fn bucket_id(&self) -> i32`                                                                        | Get bucket ID                              |

## `PartitionSpec`

| Method                                                      | Description                                           |
|-------------------------------------------------------------|-------------------------------------------------------|
| `PartitionSpec::new(spec_map: HashMap<&str, &str>) -> Self` | Create from a map of partition column names to values |
| `fn get_spec_map(&self) -> &HashMap<String, String>`        | Get the partition spec map                            |

## `PartitionInfo`

| Method                                   |  Description       |
|------------------------------------------|--------------------|
| `fn get_partition_id(&self) -> i64`      | Get partition ID   |
| `fn get_partition_name(&self) -> String` | Get partition name |

## `DatabaseDescriptor`

| Method                                                    | Description                          |
|-----------------------------------------------------------|--------------------------------------|
| `fn builder() -> DatabaseDescriptorBuilder`               | Create a database descriptor builder |
| `fn comment(&self) -> Option<&str>`                       | Get database comment                 |
| `fn custom_properties(&self) -> &HashMap<String, String>` | Get custom properties                |

## `DatabaseDescriptorBuilder`

| Method                                                                                    | Description                   |
|-------------------------------------------------------------------------------------------|-------------------------------|
| `fn comment(comment: impl Into<String>) -> Self`                                          | Set database comment          |
| `fn custom_properties(properties: HashMap<impl Into<String>, impl Into<String>>) -> Self` | Set custom properties         |
| `fn custom_property(key: impl Into<String>, value: impl Into<String>) -> Self`            | Set a single custom property  |
| `fn build() -> DatabaseDescriptor`                                                        | Build the database descriptor |

## `DatabaseInfo`

| Method                                                 | Description                     |
|--------------------------------------------------------|---------------------------------|
| `fn database_name(&self) -> &str`                      | Get database name               |
| `fn created_time(&self) -> i64`                        | Get creation timestamp          |
| `fn modified_time(&self) -> i64`                       | Get last modification timestamp |
| `fn database_descriptor(&self) -> &DatabaseDescriptor` | Get the database descriptor     |

## `LakeSnapshot`

| Field                   | Description                                       |
|-------------------------|---------------------------------------------------|
| `.snapshot_id`          | `i64` -- Snapshot ID                              |
| `.table_buckets_offset` | `HashMap<TableBucket, i64>` -- All bucket offsets |

## `GenericRow<'a>`

| Method                                                             | Description                                      |
|--------------------------------------------------------------------|--------------------------------------------------|
| `GenericRow::new(field_count: usize) -> Self`                      | Create a new row with the given number of fields |
| `fn set_field(&mut self, pos: usize, value: impl Into<Datum<'a>>)` | Set a field value by position                    |
| `GenericRow::from_data(data: Vec<impl Into<Datum<'a>>>) -> Self`   | Create a row from existing field data            |

Implements the `InternalRow` trait (see below).

## `InternalRow` trait

| Method                                                                                 | Description                             |
|----------------------------------------------------------------------------------------|-----------------------------------------|
| `fn is_null_at(&self, idx: usize) -> Result<bool>`                                     | Check if a field is null                |
| `fn get_boolean(&self, idx: usize) -> Result<bool>`                                    | Get boolean value                       |
| `fn get_byte(&self, idx: usize) -> Result<i8>`                                         | Get tinyint value                       |
| `fn get_short(&self, idx: usize) -> Result<i16>`                                       | Get smallint value                      |
| `fn get_int(&self, idx: usize) -> Result<i32>`                                         | Get int value                           |
| `fn get_long(&self, idx: usize) -> Result<i64>`                                        | Get bigint value                        |
| `fn get_float(&self, idx: usize) -> Result<f32>`                                       | Get float value                         |
| `fn get_double(&self, idx: usize) -> Result<f64>`                                      | Get double value                        |
| `fn get_string(&self, idx: usize) -> Result<&str>`                                     | Get string value                        |
| `fn get_decimal(&self, idx: usize, precision: usize, scale: usize) -> Result<Decimal>` | Get decimal value                       |
| `fn get_date(&self, idx: usize) -> Result<Date>`                                       | Get date value                          |
| `fn get_time(&self, idx: usize) -> Result<Time>`                                       | Get time value                          |
| `fn get_timestamp_ntz(&self, idx: usize, precision: u32) -> Result<TimestampNtz>`      | Get timestamp value                     |
| `fn get_timestamp_ltz(&self, idx: usize, precision: u32) -> Result<TimestampLtz>`      | Get timestamp with local timezone value |
| `fn get_bytes(&self, idx: usize) -> Result<&[u8]>`                                     | Get bytes value                         |
| `fn get_binary(&self, idx: usize, length: usize) -> Result<&[u8]>`                     | Get fixed-length binary value           |
| `fn get_char(&self, idx: usize, length: usize) -> Result<&str>`                        | Get fixed-length char value             |
| `fn get_array(&self, idx: usize) -> Result<ArrayView<'_>>`                             | Get array value (zero-copy view)        |
| `fn get_map(&self, idx: usize) -> Result<MapView<'_>>`                                 | Get map value (zero-copy view)          |
| `fn get_row(&self, idx: usize) -> Result<RowView<'_>>`                                 | Get nested row value (zero-copy view)   |

The nested getters return borrowed views — `ArrayView`, `MapView`, `RowView`. Read elements directly through the `DataGetters` accessors, or call `try_into_binary()` (`try_into_generic()` for rows) for the owned binary form.

## `ArrayView<'a>`

Returned by `get_array()`. Implements `InternalArray` + `DataGetters`, so elements are read in place via the typed getters (`get_int()`, `get_string()`, nested `get_array()` / `get_map()`, …).

| Method | Description |
|--------|-------------|
| `fn try_into_binary(self) -> Result<FlussArray>` | Materialize the owned binary `FlussArray`, re-encoding a columnar slice if needed |

## `MapView<'a>`

Returned by `get_map()`. Implements `InternalMap`.

| Method | Description |
|--------|-------------|
| `fn try_into_binary(self) -> Result<FlussMap>` | Materialize the owned binary `FlussMap`, re-encoding a columnar slice if needed |

## `RowView<'a>`

Returned by `get_row()`. Implements `InternalRow`, so nested fields are read in place via the typed getters.

| Method | Description |
|--------|-------------|
| `fn try_into_generic(self, row_type: &RowType) -> Result<GenericRow<'static>>` | Materialize the owned `GenericRow`, walking a columnar cursor field-by-field if needed |

## `FlussArray`

`FlussArray` is the owned binary row representation for `ARRAY` values. You obtain it from `ArrayView::try_into_binary()`.

| Method | Description |
|--------|-------------|
| `fn size(&self) -> usize` | Number of elements in the array |
| `fn is_null_at(&self, pos: usize) -> bool` | Check whether an element is null |
| `fn as_bytes(&self) -> &[u8]` | Get encoded bytes of the array |

Element getters mirror `InternalRow` typed getters and return `Result<T>`. For example, use `get_int()`, `get_long()`, and `get_double()` for primitive elements, and `get_string()`, `get_binary()`, `get_decimal()`, `get_timestamp_ntz()`, `get_timestamp_ltz()`, and `get_array()` for variable-length or nested elements.

## `FlussMap`

`FlussMap` is the owned binary row representation for `MAP` values. You obtain it from `MapView::try_into_binary()`.

| Method | Description |
|--------|-------------|
| `fn size(&self) -> usize` | Number of entries in the map |
| `fn as_bytes(&self) -> &[u8]` | Get encoded bytes of the map |
| `fn key_type(&self) -> &DataType` | Schema-declared type of keys |
| `fn value_type(&self) -> &DataType` | Schema-declared type of values |
| `fn entries(&self) -> Entries<'_>` | Iterator yielding `Result<(Datum, Datum)>` pairs |
| `fn get(&self, key: &Datum) -> Result<Option<Datum>>` | Linear-scan lookup by key (`O(n)`) |
| `fn key_array(&self) -> &FlussArray` | Parallel keys array (zero-copy view) |
| `fn value_array(&self) -> &FlussArray` | Parallel values array (zero-copy view) |

Most user code should prefer `entries()` (iteration) and `get()` (lookup). The `key_array()` / `value_array()` views are for serdes and Arrow-adapter code that needs zero-copy access to the underlying parallel-array layout.

## `FlussMapWriter`

`FlussMapWriter` builds a `FlussMap` for write paths.

| Method | Description |
|--------|-------------|
| `fn new(capacity: usize, key_type: &DataType, value_type: &DataType) -> Self` | Create a writer sized for `capacity` entries |
| `fn write_entry(&mut self, key: Datum, value: Datum) -> Result<()>` | Append a single entry; rejects null keys and type mismatches |
| `fn extend<I, K, V>(&mut self, entries: I) -> Result<()>` | Append every pair from `entries: IntoIterator<Item = (K, V)>` |
| `fn complete(self) -> Result<FlussMap>` | Finalize the writer and produce the `FlussMap` |

## `ChangeType`

| Value                      | Short String  | Description                      |
|----------------------------|---------------|----------------------------------|
| `ChangeType::AppendOnly`   | `+A`          | Append-only record               |
| `ChangeType::Insert`       | `+I`          | Inserted row                     |
| `ChangeType::UpdateBefore` | `-U`          | Previous value of an updated row |
| `ChangeType::UpdateAfter`  | `+U`          | New value of an updated row      |
| `ChangeType::Delete`       | `-D`          | Deleted row                      |

| Method                           | Description                         |
|----------------------------------|-------------------------------------|
| `fn short_string(&self) -> &str` | Get the short string representation |

## `OffsetSpec`

| Variant                      | Description                                     |
|------------------------------|-------------------------------------------------|
| `OffsetSpec::Earliest`       | Start from the earliest available offset        |
| `OffsetSpec::Latest`         | Start from the latest offset (only new records) |
| `OffsetSpec::Timestamp(i64)` | Start from a specific timestamp in milliseconds |

## Constants

| Constant                         | Value  | Description                                             |
|----------------------------------|--------|---------------------------------------------------------|
| `fluss::client::EARLIEST_OFFSET` | `-2`   | Start reading from the earliest available offset        |

To start reading from the latest offset (only new records), resolve the current offset via `list_offsets` before subscribing:

```rust
use fluss::rpc::message::OffsetSpec;

let offsets = admin.list_offsets(&table_path, &[0], OffsetSpec::Latest).await?;
let latest = offsets[&0];
log_scanner.subscribe(0, latest).await?;
```

## `DataTypes` factory

| Method                                           | Returns    | Description                        |
|--------------------------------------------------|------------|------------------------------------|
| `DataTypes::boolean()`                           | `DataType` | Boolean type                       |
| `DataTypes::tinyint()`                           | `DataType` | 8-bit signed integer               |
| `DataTypes::smallint()`                          | `DataType` | 16-bit signed integer              |
| `DataTypes::int()`                               | `DataType` | 32-bit signed integer              |
| `DataTypes::bigint()`                            | `DataType` | 64-bit signed integer              |
| `DataTypes::float()`                             | `DataType` | 32-bit floating point              |
| `DataTypes::double()`                            | `DataType` | 64-bit floating point              |
| `DataTypes::string()`                            | `DataType` | Variable-length string             |
| `DataTypes::bytes()`                             | `DataType` | Variable-length byte array         |
| `DataTypes::date()`                              | `DataType` | Date (days since epoch)            |
| `DataTypes::time()`                              | `DataType` | Time (milliseconds since midnight) |
| `DataTypes::timestamp()`                         | `DataType` | Timestamp without timezone         |
| `DataTypes::timestamp_ltz()`                     | `DataType` | Timestamp with local timezone      |
| `DataTypes::decimal(precision: u32, scale: u32)` | `DataType` | Fixed-point decimal                |
| `DataTypes::char(length: u32)`                   | `DataType` | Fixed-length string                |
| `DataTypes::binary(length: usize)`               | `DataType` | Fixed-length byte array            |
| `DataTypes::array(element: DataType)`            | `DataType` | Array of elements                  |
| `DataTypes::map(key: DataType, value: DataType)` | `DataType` | Map of key-value pairs             |
| `DataTypes::row(fields: Vec<DataField>)`         | `DataType` | Nested row type                    |

## `DataField`

| Method                                                                                                   | Description         |
|----------------------------------------------------------------------------------------------------------|---------------------|
| `DataField::new(name: impl Into<String>, data_type: DataType, description: Option<String>) -> DataField` | Create a data field |
| `fn name(&self) -> &str`                                                                                 | Get the field name  |