docs/src/getting-started.md

<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

  http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.
-->

# Getting Started

## Installation

Add `paimon` to your `Cargo.toml`:

```toml
[dependencies]
paimon = "0.3.0"
tokio = { version = "1", features = ["full"] }
```

By default, the `storage-fs` (local filesystem), `storage-memory` (in-memory) and `storage-oss` (Alibaba Cloud OSS) backends are enabled. To use additional storage backends, enable the corresponding feature flags:

```toml
[dependencies]
paimon = { version = "0.3.0", features = ["storage-s3"] }
```

Available storage features:

| Feature          | Backend          |
|------------------|------------------|
| `storage-fs`     | Local filesystem |
| `storage-memory` | In-memory        |
| `storage-s3`     | Amazon S3        |
| `storage-oss`    | Alibaba Cloud OSS|
| `storage-jindo`  | Alibaba Cloud OSS through JindoSDK |
| `storage-oss-cpp` | Alibaba Cloud OSS through C++ SDK v2 (read-only) |
| `storage-cos`    | Tencent Cloud COS |
| `storage-azdls`  | Azure Data Lake Storage Gen2 |
| `storage-obs`    | Huawei Cloud OBS |
| `storage-gcs`    | Google Cloud Storage |
| `storage-hdfs`   | HDFS             |
| `storage-all`    | All of the above |

## Reusing a storage backend

On the development branch, Rust embedders can implement `paimon::io::FileIOProvider`
to reuse an OpenDAL operator managed by their application. The provider receives
the original URI and returns a shared operator and its relative object path.
It can route different schemes or buckets to different operators.

```rust
use std::sync::Arc;
use opendal::Operator;
use paimon::io::{FileIOBuilder, FileIOProvider};
use paimon::{Error, Result};

#[derive(Debug)]
struct SharedStorage {
    // An application-managed operator rooted at this bucket's root.
    operator: Operator,
}

#[async_trait::async_trait]
impl FileIOProvider for SharedStorage {
    async fn create(&self, uri: &str) -> Result<(Operator, String)> {
        let key = uri.strip_prefix("s3://my-bucket/").ok_or_else(|| {
            Error::ConfigInvalid {
                message: "URI is outside the configured bucket".to_string(),
            }
        })?;
        Ok((self.operator.clone(), key.to_string()))
    }
}

// `operator` is supplied by the embedding application.
let file_io = FileIOBuilder::new("s3")
    .with_provider(Arc::new(SharedStorage { operator }))
    .build()?;
let bytes = file_io.new_input("s3://my-bucket/table/schema/schema-0")?
    .read().await?;
```

The application needs `async-trait` and a compatible `opendal-core` dependency
(named `opendal` above). Injecting a provider does not require Paimon's built-in
feature for that storage service. Supply the resulting `FileIO` to `Table::new`
or other APIs accepting a `FileIO`; option-based catalog constructors continue
to construct their own storage backends.

Provider configuration bypasses built-in storage construction. Provider errors
are returned without trying properties, environment credentials, or another
backend. `new_input` and `new_output` remain synchronous: they retain the URI and
provider, and resolution errors are returned by the subsequent async operation.

For directory listings, the returned relative path must be an unchanged suffix
of the URI, starting at a path-component boundary. An operator rooted at
`/tenant/`, for example, can resolve `s3://my-bucket/tenant/table/` to `table/`.
Both listing APIs return full URIs that can be passed back to `FileIO`. Empty
relative paths and `/` represent the operator root. Preserve literal percent
escapes, Unicode, `?`, and `#` in object keys; do not decode or normalize them as
URL components. Object paths that OpenDAL would change by trimming whitespace
or collapsing slashes are rejected instead of accessing a different object.

Providers manage operator reuse and credential refresh. Resolution is repeated
for each async file operation, but an already-open reader or writer keeps its
operator: that backend must refresh credentials internally. Custom services
must report distinct OpenDAL storage identities (`scheme`, `name`, `root`) for
different storage namespaces so cached data cannot overlap. Rename through a
provider requires both paths to resolve to the same shared service instance;
cross-backend rename is rejected.

`with_fs_operator` remains available for filesystem paths. It cannot be combined
with `with_provider` on the same builder. Without a provider, property-based
storage configuration behaves as before.

## Parquet Dictionary Encoding

Dictionary encoding is enabled by default. Disable it for all columns with
`parquet.enable.dictionary=false`, or override one column with
`parquet.enable.dictionary#event_time=false` (Java-compatible syntax).
Paths use dot-separated physical Parquet leaves, not logical MAP keys.
These options affect new writes only.

## Parquet Row Groups and Pages

The Parquet writer accepts Java's sizing keys as plain integers:
`parquet.block.size` (row group bytes), `parquet.page.size`,
`parquet.dictionary.page.size`, `parquet.page.row.count.limit` and
`parquet.page.size.row.check.min`. As in Java, `file.block-size` (for example
`32 mb`) overrides `parquet.block.size`. Unset keys keep the parquet-rs
defaults: no row group byte limit with at most 1,048,576 rows, 1 MiB data and
dictionary pages, and 20,000 rows per page. parquet-rs checks page limits between
write batches of `parquet.page.size.row.check.min` rows (1,024 by default, and
never more than the effective page row limit), so a page can exceed its row limit
by up to one batch. With a row group byte limit, the writer passes rows in slices
of at most a quarter of the limit and closes the row group once its encoded size
reaches the limit, so large or skewed batches stay close to it. These options
affect new writes only.

## Map and Multiset Arrow Values

`MAP<K, V>` and `MULTISET<T>` both use Arrow Map with a non-null entry struct
and non-null keys. A multiset's values are non-null `Int32` occurrence counts.
Column and ordinary map value nullability follow the existing schema rules.
NULL keys are rejected by Arrow's runtime validation.
Row, Avro, and Parquet retain their Java-compatible file encodings.

## Parquet Page Pruning

Parquet Page Index pruning is enabled by default. Set
`parquet.filter.columnindex.enabled=false` to avoid fine-grained range requests;
this may reduce request count while increasing bytes read. The option does not
change Page Index generation.

## Mosaic File Format

Mosaic data file reads and writes are available for ordinary Paimon tables. Set
`file.format=mosaic` when creating an append or primary-key table. Mosaic supports
scalar, array, and map columns; `ROW`, `MULTISET`, `VECTOR`, `VARIANT`, and `BLOB`
columns are unsupported, matching Java Paimon. Mosaic writes use Zstd compression.

The writer accepts `mosaic.num-buckets`, `file.block-size`, and a comma-separated
`mosaic.stats-columns` list. Statistics are included in manifest entries for the
selected columns and can prune files during scans. An empty list disables writer
statistics. For example:

```text
file.format = mosaic
file.compression = zstd
mosaic.num-buckets = 4
file.block-size = 128mb
mosaic.stats-columns = id,event_time
```

## FileIndexes for Append Writes

Ordinary append writes can generate Bitmap, Bloom Filter, and Range Bitmap indexes for supported
top-level columns using table options:

```text
file-index.bitmap.columns = category
file-index.bloom-filter.columns = id
file-index.bloom-filter.id.items = 100000
file-index.bloom-filter.id.fpp = 0.01
file-index.range-bitmap.columns = score
file-index.range-bitmap.score.chunk-size = 16kb
file-index.in-manifest-threshold = 500 B
```

Column lists are comma-separated. Bitmap supports `version` (currently `2` only)
and `index-block-size` per column. Bloom Filter supports `items` and `fpp`.
Range Bitmap writes Java V1 payloads and supports `chunk-size` per column
(0 through 2147483647 bytes). The default is `16kb`, except for Boolean,
TinyInt, and SmallInt, which default to `0b`. A chunk's first key is stored
in its header, so `0b` is valid and creates one dictionary chunk per distinct key.
Supported Range Bitmap types are Boolean, TinyInt, SmallInt, Int, BigInt,
Float, Double, Decimal (precision <= 18), Date, Time, Timestamp and
LocalZonedTimestamp (precision <= 6), and Char/VarChar/String.
Range Bitmap enables range predicate pruning through the existing file index
reader; it does not add TopN support.
For index types supported for writing, invalid columns, unsupported data types,
and invalid index options fail when creating the writer. Index types without a
writer (such as `bsi`) and all their options are ignored, so
these table properties do not prevent append writes.

When supported indexes are configured, each data file gets its own index.
The complete serialized index is embedded in the manifest when its size
is at most `file-index.in-manifest-threshold` (default `500 B`); larger indexes
are stored beside the data file as a `.index` sidecar. Indexed write failures
return an error. Failed preparation discards outputs from that invocation before
returning any messages. Files handed off by an earlier successful preparation
remain available, including after a later preparation or Paimon Table commit
exception. Explicit commit `abort` deletes newly written files only when the caller
knows those messages are uncommitted and will never be submitted. Never call it
when the Paimon Table commit outcome is unknown: publication can succeed even when
its response fails.

Snapshot publication never replaces an existing snapshot. The built-in local
filesystem stages a complete file before publishing it without replacement;
memory and supported object stores use conditional creation. A publication
conflict rebases the commit on the latest snapshot and retries. Backends without
atomic publication of a complete file must use a catalog-provided `SnapshotCommit`
with concurrency control; filesystem commits on those backends return an
unsupported I/O error. This includes COS conditional writes, which do not reliably
prevent overwrites on buckets that have ever enabled versioning, and position
writers such as ADLS and custom filesystem operators without conditional rename.
OSS conditional publication assumes the bucket uses its default versioning
configuration and has never enabled versioning.
Application storage providers can implement `FileIOProvider::write_atomic_if_not_exists`
to supply their own atomic publication primitive. FUSE routes use local staged
publication and require the mount to support publishing without replacement.
Storage failures never fall back to an unconditional snapshot write.
All concurrent writers, including writers from other engines, must use compatible
concurrency control.

Format Tables follow Java's two-phase publication: validation failures discard
staging only; append publication or partition-registration failures roll back new
targets; overwrite failures retain replacement targets after old data is removed.
Successfully registered append targets also survive a later abort. Cleanup attempts
every file, and a cleanup failure does not replace the original commit error.

`file-index.read.enabled` controls only reading, independently of index creation.
Existing files are not backfilled. Index generation is not supported for
primary-key writes, data-evolution writes, or dedicated Blob/Vector paths;
ordinary writer creation rejects supported index configuration on these paths.
COW and partial DataEvolution rewrites do not generate indexes. Nested indexes
and additional index types are not supported.

## Scalar Global Index Build Memory

BTree, Bitmap, and Multivalue global index builds stream projected Arrow batches
and sort serialized keys within `write-buffer-size` (default `256mb`), matching
Java's sort buffer configuration. This does not change index files or shard boundaries.
Set it as a table property or pass it through the index builder's `with_options`.

When keys exceed the sort buffer, the build writes temporary runs to the system
temporary directory and merges them pairwise. Temporary files are removed on
success, failure, or when cancelled work finishes. Allow local disk space for
serialized keys and merge output; a merge can temporarily retain both its inputs
and output. A single key larger than the sort budget returns `ResourceExhausted`.

The budget covers retained serialized sort keys and their row slots, not total
process RSS. The active Arrow batch, reader prefetch, format-writer state, two
merge keys and fixed I/O buffers have additional memory costs. FM indexes keep
their existing streaming writer and do not use this sort setting.

## Catalog Management

Paimon supports multiple catalog types. The `CatalogFactory` provides a unified way to create catalogs based on configuration options.

### Create a Catalog

The `CatalogFactory` automatically determines the catalog type based on the `metastore` option:

```rust
use paimon::{CatalogFactory, CatalogOptions, Options};

// Local filesystem (no credentials needed)
let mut options = Options::new();
options.set(CatalogOptions::WAREHOUSE, "/path/to/warehouse");
let catalog = CatalogFactory::create(options).await?;

// Amazon S3
let mut options = Options::new();
options.set(CatalogOptions::WAREHOUSE, "s3://bucket/warehouse");
options.set("s3.access-key-id", "your-access-key-id");
options.set("s3.secret-access-key", "your-secret-access-key");
options.set("s3.region", "us-east-1");
// Optional: configure retries for temporary S3 failures.
options.set("s3.retry.count", "10");
options.set("s3.retry.interval.millisecond", "500");
let catalog = CatalogFactory::create(options).await?;

// Alibaba Cloud OSS
let mut options = Options::new();
options.set(CatalogOptions::WAREHOUSE, "oss://bucket/warehouse");
options.set("fs.oss.accessKeyId", "your-access-key-id");
options.set("fs.oss.accessKeySecret", "your-access-key-secret");
options.set("fs.oss.endpoint", "oss-cn-hangzhou.aliyuncs.com");
// Optional: configure retries for temporary OSS failures.
options.set("fs.oss.retry.count", "10");
options.set("fs.oss.retry.interval.millisecond", "500");
let catalog = CatalogFactory::create(options).await?;

// Tencent Cloud COS
let mut options = Options::new();
options.set(CatalogOptions::WAREHOUSE, "cosn://bucket/warehouse");
options.set("fs.cosn.userinfo.secretId", "your-secret-id");
options.set("fs.cosn.userinfo.secretKey", "your-secret-key");
options.set("fs.cosn.endpoint", "https://cos.ap-shanghai.myqcloud.com");
let catalog = CatalogFactory::create(options).await?;

// Azure Data Lake Storage Gen2
let mut options = Options::new();
options.set(CatalogOptions::WAREHOUSE, "abfs://filesystem@account.dfs.core.windows.net/warehouse");
options.set("azure.account-key", "your-account-key");
let catalog = CatalogFactory::create(options).await?;

// If you use the short form "abfs://filesystem/warehouse", set the endpoint explicitly:
// options.set("azure.endpoint", "https://account.dfs.core.windows.net");

// Huawei Cloud OBS
let mut options = Options::new();
options.set(CatalogOptions::WAREHOUSE, "obs://bucket/warehouse");
options.set("fs.obs.access.key", "your-access-key-id");
options.set("fs.obs.secret.key", "your-secret-access-key");
options.set("fs.obs.endpoint", "https://obs.cn-north-4.myhuaweicloud.com");
let catalog = CatalogFactory::create(options).await?;

// Google Cloud Storage
let mut options = Options::new();
options.set(CatalogOptions::WAREHOUSE, "gs://bucket/warehouse");
options.set("gcs.credential-path", "/path/to/service-account.json");
let catalog = CatalogFactory::create(options).await?;

// REST catalog
let mut options = Options::new();
options.set(CatalogOptions::METASTORE, "rest");
options.set(CatalogOptions::URI, "http://localhost:8080");
options.set(CatalogOptions::WAREHOUSE, "my_warehouse");
let catalog = CatalogFactory::create(options).await?;
```

For a DLF REST catalog on ECS, RAM-role credentials can be rotated automatically:

```rust
let mut options = Options::new();
options.set(CatalogOptions::METASTORE, "rest");
options.set(CatalogOptions::URI, "https://your-dlf-endpoint");
options.set(CatalogOptions::WAREHOUSE, "your_catalog");
options.set(CatalogOptions::TOKEN_PROVIDER, "dlf");
options.set(CatalogOptions::DLF_REGION, "cn-hangzhou");
options.set(CatalogOptions::DLF_TOKEN_LOADER, "ecs");
options.set(CatalogOptions::DLF_TOKEN_ECS_ROLE_NAME, "your-ram-role");
options.set(CatalogOptions::DATA_TOKEN_ENABLED, "true");
let catalog = CatalogFactory::create(options).await?;
```

`dlf.token-loader=ecs` refreshes the credentials used to authenticate DLF
catalog requests. The role name is optional; when omitted, it is read from the
ECS metadata service. `data-token.enabled=true` separately enables temporary
credentials returned by the REST server for table data access. A loaded table
refreshes those credentials before expiration, so callers do not need to load
the table again. Static `dlf.access-key-id`, `dlf.access-key-secret`, and
`dlf.security-token` values are not rotated.

With `storage-fs`, internal REST tables can access an existing FUSE mount:

```rust
options.set("fuse.enabled", "true");
options.set("fuse.root", "/mnt/warehouse");
options.set("fuse.mode", "pvfs");
options.set("fuse.validation-mode", "strict");
```

The default `pvfs` layout uses `<root>/<database>/<table>`; `raw` uses the URI
path beneath the root, excluding the scheme and authority. Original table URIs
remain in metadata and REST requests. The default `strict` validation checks the
default database's mount once per catalog; `warn` falls back to ordinary FileIO
when that mount is missing, and `none` skips validation. Mounted tables do not
request cloud data credentials. External tables keep their ordinary FileIO.

To read credentials managed by another process, use a local token file:

```rust
options.set(CatalogOptions::DLF_TOKEN_LOADER, "local_file");
options.set(CatalogOptions::DLF_TOKEN_PATH, "/path/to/token.json");
```

The JSON uses `AccessKeyId`, `AccessKeySecret`, optional `SecurityToken`, and
optional `Expiration`. Expiring credentials are reloaded before expiration.

With `storage-jindo`, set `fs.oss.impl` to `jindo` to use an installed JindoSDK
for OSS scan planning. Set `fs.jindo.library.path` unless the library is
available through `JINDOSDK_HOME` or `JINDOSDK_LIBRARY_PATH`. The Python binding
also discovers it from an installed `pyjindosdk` package. This initial
integration supports object stat, reads, and listings, but not writes, deletes,
or copies. Jindo reads are limited to 8 concurrent requests per OSS operator by
default. Set `fs.jindo.max.concurrent.reads` to a positive integer to tune this
limit for the available network and JindoSDK connection capacity. JindoSDK state
initialized before `fork` cannot be reused in the child; use `spawn` or initialize
Jindo only after worker processes start.

With `storage-oss-cpp`, set `fs.oss.impl=cpp` and
`fs.oss.cpp.library.path` to the installed bridge library. This experimental
backend supports stat, reads and listings, with 8 concurrent requests per
operator by default. OpenDAL remains the default backend.
See the [build instructions and options](https://github.com/apache/paimon-rust/tree/main/integrations/oss-cpp).

Supported metastore types:

| Metastore Type | Description                      |
|----------------|----------------------------------|
| `filesystem`   | Local or remote filesystem (default) |
| `rest`         | REST catalog server              |

### Local Disk Cache

Catalogs can cache immutable Paimon metadata and index file blocks on local
disk. The cache works with both filesystem and REST catalogs:

```rust
let mut options = Options::new();
options.set(CatalogOptions::WAREHOUSE, "s3://bucket/warehouse");
options.set(CatalogOptions::LOCAL_CACHE_ENABLED, "true");
options.set(CatalogOptions::LOCAL_CACHE_DIR, "/var/cache/paimon/worker-1");
options.set(CatalogOptions::LOCAL_CACHE_MAX_SIZE, "20 GiB");
options.set(CatalogOptions::LOCAL_CACHE_BLOCK_SIZE, "1 MiB");
options.set(
    CatalogOptions::LOCAL_CACHE_WHITELIST,
    "meta,global-index",
);
let catalog = CatalogFactory::create(options).await?;
```

| Option | Default | Description |
|--------|---------|-------------|
| `local-cache.enabled` | `false` | Enable catalog-scoped local block caching. |
| `local-cache.dir` | none | Optional base directory. When set, Paimon uses a persistent disk cache in a private versioned child directory; otherwise it uses memory. |
| `local-cache.shared` | `false` | Share disk blocks across processes on one node; requires `local-cache.dir` and `local-cache.max-size`. |
| `local-cache.max-size` | unlimited | Maximum cache size. Memory caches count payload bytes; disk caches count encoded bytes. Values accept byte units such as `512 MiB` or `20 GiB`. |
| `local-cache.high-priority-pool-ratio` | `0.25` | Fraction of a finite cache budget protected for `meta` and `global-index` blocks. Use `0` for a single LRU. |
| `local-cache.block-size` | `1 MiB` | Block size used for cached range reads. |
| `local-cache.whitelist` | `meta,global-index` | Comma-separated eligible types: `meta`, `global-index`, `bucket-index`, `data`, `structured-data`, and `file-index`, or `*` for all of them. |

Use `local-cache.whitelist=meta,global-index,structured-data` to cache data
blocks without dedicated BLOB, video, or vector-store files. `data` and `*`
retain their broader behavior.

The protected and regular LRUs share one total budget; either can use idle
capacity. Catalogs or processes sharing a disk cache must use the same ratio.

Each catalog owns its in-memory cache for the catalog's lifetime. Disk caches
are reused after process restarts. Cache keys include a catalog-configuration
fingerprint and the canonical storage object path, so catalogs can safely use
the same base directory without reading one another's entries. Runtime cache
read, write, validation, and eviction failures are fail-open: the original
storage remains the source of truth. Paimon mutable markers and temporary files
always bypass the cache; other eligible files rely on Paimon's immutable-file
convention. Cache managers using the same canonical directory in one process
share LRU, size accounting, and invalidation state; if their configured limits
differ, the smallest `local-cache.max-size` is used. Restart recovery runs on a
blocking worker, reads only block headers and file metadata, and validates
payload CRC lazily on the first hit. Without shared mode, use a separate
`local-cache.dir` per process.

To share disk blocks, set `local-cache.enabled=true`, `local-cache.shared=true`,
`local-cache.dir`, and `local-cache.max-size` consistently in same-UID processes
on one node. NFS, cross-node sharing, and catalog reuse after `fork` are
unsupported. After failed invalidation or rename, verify storage state and use
a fresh cache directory before resuming sharing. Allow disk headroom for up to
eight staged 1-MiB writes per cache instance beyond `local-cache.max-size`.

### Parsed Parquet Metadata Cache

Set `file-format.metadata-cache.shared=true` to reuse parsed Parquet metadata
across read contexts in one process (default `false`). Only equal budgets share
a pool; this is not a process-wide memory limit. Shared hits perform an uncached
one-byte storage-access check; readers without that check reload metadata.
ROW metadata and BLOB indexes remain context-local. This is independent of
`local-cache.shared`, which shares disk blocks across processes.

### BLOB Index Cache

Decoded `.blob` indexes are cached per catalog. Set
`cache.blob-index.max-size` to control the budget (default `64 MiB`; `0`
disables caching). BLOB payloads are not cached.

### Manage Databases

```rust
use paimon::Catalog; // import the trait
use std::collections::HashMap;

// Create a database
catalog.create_database("my_db", false, HashMap::new()).await?;

// List databases
let databases = catalog.list_databases().await?;

// Drop a database (cascade = true to drop all tables inside)
catalog.drop_database("my_db", false, true).await?;
```

### Manage Tables

```rust
use paimon::catalog::Identifier;
use paimon::spec::{DataType, IntType, VarCharType, Schema};

// Define a schema
let schema = Schema::builder()
    .column("id", DataType::Int(IntType::new()))
    .column("name", DataType::VarChar(VarCharType::string_type()))
    .build()?;

// Create a table
let identifier = Identifier::new("my_db", "my_table");
catalog.create_table(&identifier, schema, false).await?;

// List tables in a database
let tables = catalog.list_tables("my_db").await?;

// Get a table handle
let table = catalog.get_table(&identifier).await?;
```

## Reading a Table

Paimon Rust uses a scan-then-read pattern: first scan the table to produce splits, then read data from those splits as Arrow `RecordBatch` streams.

```rust
use futures::StreamExt;

// Get a table from the catalog
let table = catalog.get_table(&Identifier::new("my_db", "my_table")).await?;

// Create a read builder
let read_builder = table.new_read_builder();

// Step 1: Scan — produces a Plan containing DataSplits
let plan = {
    let scan = read_builder.new_scan();
    scan.plan().await?
};

// Step 2: Read — consumes splits and returns Arrow RecordBatches
let reader = read_builder.new_read()?;
let mut stream = reader.to_arrow(plan.splits())?;

while let Some(batch) = stream.next().await {
    let batch = batch?;
    println!("RecordBatch: {batch:#?}");
}
```

## Building from Source

```bash
git clone https://github.com/apache/paimon-rust.git
cd paimon-rust
cargo build
```

## Running Tests

```bash
# Unit tests
cargo test

# Integration tests (automatically extract the committed golden warehouse)
cargo test -p paimon-integration-tests
```