Getting Started¶
Installation¶
Add paimon to your Cargo.toml:
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:
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.
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:
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:
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:
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:
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:
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:
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.
For DLF authentication (token.provider=dlf), DLF_TOKEN_PATH provides a
fallback path when no token loader, token path, or inline AK/STS credentials are
configured. The same loader and refresh behavior apply. Unset or empty
environment values are ignored.
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.
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:
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¶
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¶
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.
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:#?}");
}