Add paimon to your Cargo.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:
[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 |
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.
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.
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<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 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 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
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.
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.
Paimon supports multiple catalog types. The CatalogFactory provides a unified way to create catalogs based on configuration options.
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.
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 |
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.
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.
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.
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?;
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?;
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:#?}"); }
git clone https://github.com/apache/paimon-rust.git cd paimon-rust cargo build
# Unit tests cargo test # Integration tests (automatically extract the committed golden warehouse) cargo test -p paimon-integration-tests