This PyPi package contains the Python APIs for using Paimon.

Pypaimon requires Python 3.6+.

Run from paimon-python/ with pip supporting dependency groups.

Normal build:

python -m pip install --group build
python -m build

Development build (editable installation with development dependencies):

python -m pip install -e . --group dev
python -m build

Both produce a source archive and wheel in dist/.

Install pypaimon[oss] (legacy PyArrow data access) or pypaimon[jindo] (Jindo data access). Both include oss2 for atomic metadata writes. Configure fs.oss.accessKeyId, fs.oss.accessKeySecret and fs.oss.endpoint, plus fs.oss.securityToken when using STS. An endpoint without a scheme uses HTTPS for metadata writes. Credentials supplied only through an underlying filesystem's credential provider must also be supplied through these options.

Metadata requests always use OSS Signature V4 for both AK and STS credentials, independently of the Jindo data-access signer setting. Set fs.oss.region to the bucket's region ID, such as cn-hangzhou. If unset, it is inferred from standard oss-<region>.aliyuncs.com or oss-<region>-internal.aliyuncs.com endpoints. Other endpoints, including acceleration endpoints, require an explicit region.

Atomic metadata PUTs also forward the OSS server-side encryption options, using the same resolution as Java OSSFileIO:

The first three options reject blank values and invalid combinations before any request is sent. If no encryption options are set, no encryption headers are added and OSS applies the bucket's default policy. These settings cover the atomic metadata PUT path; ordinary data writes still use the configured PyArrow/Jindo filesystem and its encryption capabilities.

For oss:// paths, FileIO.get selects OssFileIO, a thin PyArrowFileIO subclass that overrides atomic creation. Filesystem initialization, path handling, and ordinary PyArrow/Jindo file operations are inherited unchanged. Use FileIO.get(path, options) or construct OssFileIO explicitly for OSS atomic writes. REST token refresh and ResolvingFileIO also route atomic writes through this implementation.

When the bucket is confirmed unversioned, OssFileIO.try_to_write_atomic uses a single OSS PUT with x-oss-forbid-overwrite=true. Exactly one writer can create a given object; FileAlreadyExists returns False so snapshot commits can retry. Other SDK errors are raised as OSError, retaining their cause for diagnostics. A lost PUT response is not retried by the SDK; the snapshot commit loop checks the commit user and identifier before retrying.

The vendor SDK is an optional backend dependency, imported only for OSS atomic writes. The common FileIO API and Paimon table format remain independent of it. This implementation uses OSS-specific conditional creation; it does not provide the same atomic-write capability for every object store.

Conditional creation requires a bucket that has never enabled versioning. The first atomic write on each OssFileIO instance checks GetBucketVersioning and caches the result, including the query-denied fallback. Concurrent first writes may repeat the check and warning. Query errors other than 403 AccessDenied are not cached. If versioning is Enabled/Suspended, the state is unrecognized, or the query returns 403 AccessDenied, the operation logs a warning when caching the fallback and uses the inherited PyArrow/Jindo temporary-file-and-rename path. This preserves legacy writes without making version-query permission mandatory, but the fallback does not guarantee safe concurrent commits. It also retains the existing backend's encryption behavior rather than applying the conditional PUT's OSS SSE headers. Invalid credentials, expired tokens, missing buckets, and other query failures still propagate as errors.

Grant oss:GetBucketVersioning and keep versioning disabled to use conditional creation. Keep bucket versioning and version-query permissions unchanged for the instance's lifetime; recreate the FileIO after changing them. A configuration change is not guaranteed to produce an error and can invalidate the conditional-write guarantee.

All concurrent writers must use conditional creation. Older Python clients or other clients that overwrite snapshot objects can still overwrite a successful commit. This change does not add conditional writes for other object stores.

For a batch update of selected columns in a data-evolution table with row tracking, pass an Arrow table containing _ROW_ID and the columns to update. Create the updater and committer from the same builder so they share a commit user:

builder = table.new_batch_write_builder()
updater = builder.new_update().new_update_by_row_id()
messages = updater.update_columns(updates, ["name"])
commit = builder.new_commit()
try:
    commit.commit(messages)
finally:
    commit.close()

For stream updates, use table.new_stream_write_builder() and pass the stream commit identifier to new_update().new_update_by_row_id(commit_identifier).

For row-tracking tables with a Parquet OffsetIndex, PyPaimon can read a contiguous _ROW_ID range, including standard VARIANT columns, without decoding the full row group. This is enabled by default and can be disabled with the table option:

table = table.copy({"parquet.filter.columnindex.enabled": "false"})

Unsupported reads use the normal path. Reading fewer bytes may require more object-store requests.

PyPaimon can write Arrow batches through the optional pypaimon-rust runtime. Enable it on a table independently of native commit:

native_table = table.copy({"write.native.enabled": "true",
                           "commit.native.enabled": "true"})
builder = native_table.new_batch_write_builder()
writer, commit = builder.new_write(), builder.new_commit()
try:
    writer.write_arrow(data)
    commit.commit(writer.prepare_commit())
finally:
    writer.close()
    commit.close()

The native writer returns ordinary PyPaimon commit messages, so the Python committer also works when commit.native.enabled is false. Batch overwrite and reusable stream writers retain the builder's commit user and identifier. Native write supports Parquet append, primary-key and data-evolution tables, including top-level scalar BLOB Arrow columns, on the same filesystem/JDBC publication route as native commit. ARRAY/MAP BLOB, video and optional data-evolution row sidecars select the Python writer. Writer methods requiring Python's specialized path select the Python writer before native data is written. If the runtime or table route is unavailable, write uses Python. Once Rust starts writing a batch, errors propagate without retrying that batch through Python.

Batch and stream merge_into also use Rust core for eligible data-evolution Parquet tables when write.native.enabled=true. Existing WhenMatched and WhenNotMatched clauses accept Arrow/pandas input, Paimon table sources, or self-merge on _ROW_ID. Core pins the target snapshot, matches keys, selects the first satisfied clause and prepares updates, deletion vectors and inserts. SQL conditions, including subqueries, execute in the Rust DataFusion adapter; Python transports the normalized clauses and returned commit messages. Conditional MERGE requires Python 3.10+ and the existing pypaimon[sql] extra, which installs both Python DataFusion and pypaimon-rust. Shared clause validation checks that dependency before native execution. NULL keys do not match. Multiple source rows matching a target are rejected before action conditions, except for a sole unconditional DELETE, following Paimon Spark MERGE. For table sources, core independently pins the selected source branch/snapshot and reads only keys and referenced columns. Append and primary-key sources do not require data evolution. Source scan options select the same point in time on Python and native paths. Numeric incremental-between-timestamp source windows are resolved and read by core. Non-REST Blob views requiring catalog resolution and packed-video inserts use the Python path selected before native execution starts. Native failures propagate without Python retry or deleting files from earlier prepared actions.

Python and native writers use Java's escaped partition directories for data, changelog, Blob and row sidecar files, including external locations. Both planners, file-range metadata, commit callbacks and explicit aborts use the same path rules. Row sidecar reads resolve that path even when the primary file is unavailable.

MAP columns configured with fields.<name>.map.storage-layout=shared-shredding also use native Parquet writes and data-evolution updates, including predicate updates and upserts. Rust applies Java's plain, sequential and lru column placement policies (lru by default), and adapts the physical column count between completed files. Files remain readable by the Python reader, including literal MAP-key projections. Shared-shredding requires none, lz4 or zstd compression.

Primary-key dynamic buckets (bucket=-1, with partition fields included in the primary key) also support native writes. HASH indexes are restored across writer restarts; dynamic-bucket.max-buckets bounds bucket growth and reuses existing buckets after that limit. Dynamic buckets use the trimmed primary key, so bucket-key must not be configured. As in Java, each bucket must have one writer owner: a HASH ADD replaces the complete previous index and does not carry a concurrent-writer baseline. Partitions with existing data but missing HASH indexes must be rewritten before incremental writes.

Cross-partition primary keys (bucket=-1, with some partition columns outside of the primary key) also support native writes. Rust rebuilds the global index from live rows on writer startup and preserves the full primary key. Deduplicate moves a key by deleting its old location; first-row keeps its first partition; partial-update and aggregation apply changes in the existing partition, matching Java. Use one writer owner for the global key space. Index TTL is not supported, and sequence.field and bucket-key are invalid for this mode. The Python writer does not implement cross-partition routing, so this mode requires the native runtime.

Ordinary postpone writes (bucket=-2) support native Parquet deduplicate writes without BLOB columns. Pending files retain input order, row kinds and duplicates, and roll at a batch boundary after reaching target-file-row-num. Normal scans expose only real buckets after deferred bucket assignment. Other postpone merge engines, BLOB writes use the Python writer.

The fixed-bucket postpone builder uses native Parquet writes for deduplicate, first-row, partial-update and aggregation when supplied with a shared bucket plan or postpone.default-bucket-num. The default count is used exactly, without rounding or applying the inference limit. Append reuses each existing partition's bucket count; overwrite uses the new count for overwritten partitions. Pending files may coexist with real buckets and remain available for later assignment.

from pypaimon.write.postpone_bucket import PostponeBucketPlan

# A driver can serialize this small plan and give it to every worker.
plan = PostponeBucketPlan({("2026-10-01",): 3, ("2026-10-02",): 5})
builder = table.new_postpone_fixed_bucket_write_builder().with_bucket_plan(plan)
writer = builder.new_write()
try:
    writer.write_arrow(arrow_table)
    messages = writer.prepare_commit()
finally:
    writer.close()
builder.new_commit().commit(messages)

All rows for one partition/bucket must have one writer owner. Python and native committers reject overlapping owners or writes that became stale after their baseline snapshot. Different real buckets and pending files can be appended concurrently. Overwrite detects concurrent changes to any overwritten partition. A shared plan must cover every input partition; a configured default does not fill missing entries in an explicit plan. Ray workers use native routing and writing with the driver's shared plan.

Local automatic size estimation still uses the Python planner. Fixed-bucket native writes currently exclude BLOB and deletion-vector tables. They do not perform staging, compaction or automatic rescaling of existing partitions.

Native writes honor data-file.path-directory and the configured data-file.external-paths strategy. Existing files keep their recorded locations when the write destinations change. Python and native readers and committers can exchange these files, including external data files and their index sidecars.

BLOB Arrow values may contain payload bytes or serialized descriptors. Fields listed in blob-descriptor-field remain inline and require descriptors. HTTP(S) references use decoded response streams, including gzip and deflate. Blob files roll by payload size, independently of the normal Parquet files. Python Blob row objects can provide custom streams or URI readers; write_row on a BLOB table selects the Python writer before any native data is written. Switching to row writes after native Arrow writes is rejected. Use write.native.enabled=false when mixing Arrow batches and Python Blob objects in one writer.

A successful prepare_commit() transfers file ownership to the caller. Writer abort() and close() preserve those prepared files and may clean only outputs still owned by the writer. Explicit commit.abort(messages) deletes newly written data, changelog, sidecars and added indexes, matching Java's Paimon Table committer. Call it only when those messages are known to be uncommitted and will never be submitted. Never call it after a commit with an unknown outcome: publication may have succeeded before its response failed. Internal failure paths preserve prepared files instead of calling commit abort.

Both native options are disabled by default.

PyPaimon can submit append and batch overwrite commits through the optional pypaimon-rust runtime. Enable it independently of native planning and reading:

native_table = table.copy({"commit.native.enabled": "true"})
builder = native_table.new_batch_write_builder()
writer, commit = builder.new_write(), builder.new_commit()
try:
    writer.write_arrow(data)
    commit.commit(writer.prepare_commit())
finally:
    writer.close()
    commit.close()

When only native commit is enabled, the Python writer produces files. Its commit messages cross the Java v14 wire format into CommitMessage.deserialize() and are committed by Rust. Batch and stream append commits retain the Python builder's commit user, identifier, empty-commit option, and batch one-shot lifecycle. Explicit abort also supports native cleanup of uncommitted files.

For batch overwrite, configure the Python builder as usual:

builder = native_table.new_batch_write_builder().overwrite({"pt": "2026-09-22"})

Overwrite preserves the Python commit user and follows the table's dynamic-partition-overwrite option. Dynamic overwrite replaces only partitions present in the messages and does nothing for empty input. Static overwrite replaces partitions matching the spec, including for empty input; an empty spec matches the whole table. Empty overwrite of an unpartitioned table truncates it. Static and unpartitioned overwrite record an OVERWRITE snapshot even when no files match, following Java; this also applies when the operation uses Python.

Overwrite is configured only through BatchWriteBuilder, following Java's batch/stream API split. StreamWriteBuilder does not expose overwrite.

Native commits require a runtime containing the batch identity bridge in paimon-rust #916, built on #915. If the optional runtime is not installed, commits use Python. The current native route supports main-branch tables using filesystem/JDBC catalogs or FileStoreTable.from_path() with standard FileIO. Truncate, REST/catalog-managed publication, custom FileIO/environments, commit callbacks, and snapshot properties use Python. Data-evolution updates that need Python's row-id conflict rewriting also retain the Python path. Compact increments remain unsupported by both committers.

Fallback is limited to runtime availability, table construction and message conversion before a native mutation starts. A native commit error propagates; the adapter neither retries it through Python nor aborts files, since the snapshot may already have been published. The option is disabled by default.

PyPaimon can plan splits with the optional pypaimon-rust package. Planning and reading are independently selectable, so native plans can still use the Python reader:

native_table = table.copy({"scan.native-plan.enabled": "true"})
builder = native_table.new_read_builder()
plan = builder.new_scan().plan()
rows = builder.new_read().to_arrow(plan.splits())
explanation = builder.explain()
print(explanation.native_planned)

The adapter checks the installed binding's capabilities and falls back to the Python planner for unsupported scans. New bindings preserve plan.snapshot_id even when pruning removes every split. Native explain output includes snapshot and split metadata; native pruning counters are not exposed.

To run both split planning and data-file reading in Rust, enable the independent native-read option:

native_table = table.copy({"read.native.enabled": "true"})
builder = native_table.new_read_builder().with_projection(["id", "name"])
plan = builder.new_scan().plan()
rows = builder.new_read().to_arrow(plan.splits())

Native reads return PyArrow batches through the Arrow C Data interface. Native planner handles are used directly when available; splits refined by Python index or shuffle logic are serialized through the stable split contract and reconstructed by Rust. Nested projection and row-kind output are supported. Query authorization retains the Python reader. For both materialized to_arrow() and streaming to_arrow_batch_reader() reads, the effective split parallelism (the method argument, read.parallelism, or the automatic default) runs independent Rust readers. Splits stay in input order but contiguous groups are balanced by physical file bytes to reduce worker skew. An unfiltered row limit also caps reader fan-out to avoid speculative work. Streaming buffers at most one batch per reader and closing the batch reader interrupts native reads that are still in flight. A missing reader capability, unsupported route, or native-reader construction failure falls back to Python; I/O and data errors raised after streaming starts surface to the caller.

Native planning and reading require Rust's resolved-schema bindings: Table.from_resolved_schema() for filesystem, JDBC and path-based tables, and Table.copy_with_resolved_schema() for REST tables. The adapter passes the Python table's resolved schema and complete effective options without an option whitelist. Stale table objects, historical schemas, and copy() overrides or option removals no longer require catalog reloading or Python planning. Tables opened with FileStoreTable.from_path(path, file_io_options=None) use the same path with standard local, PyArrow or resolving FileIO. Storage options configure FileIO; use copy() for table read options. JDBC planning uses the resolved table location and storage properties without opening another database connection. REST tables use Table.copy_with_resolved_schema() to preserve the same schema and option semantics, including branches whose schemas are catalog-managed. Matching REST tables retain the native environment across scans and read-option copies, preserving FileIO caches. Worker deserialization creates a fresh environment. The native table retains REST credentials, token refresh and catalog snapshot resolution. Database and table names containing dots are passed as separate identifier components. REST snapshot results (including empty results) take precedence over filesystem snapshots. REST errors, including HTTP 501, are propagated as in Java. Custom catalog/FileIO contexts still fall back when they cannot be reproduced.

Explicit row ranges on data-evolution tables require ReadBuilder.with_row_ranges(). Watermark time travel requires Rust 0.4 or newer. Branch reads require the branch-aware binding exposing Table.branch(), and the resolved branch is checked before planning. Deletion-vector scans require pypaimon-rust>=0.4.0, which includes schema-aware decoding of index manifests. Index paths follow Java: an explicit external path takes precedence; otherwise index-file-in-data-file-dir selects the bucket or table index directory. Bucket paths use the partition field types and partition.legacy-name to match Java formatting, including timestamp precision and different JVM float spellings. Python bucket-index writes use Java partition paths. Explicit paths identify external files and floating partition directories whose spelling depends on the JDK version. Missing bucket indexes fail instead of searching Python layouts. Older releases and prereleases before 0.4.0 use the Python planner for deletion vectors. When using an unreleased 0.4.0 development wheel, rebuild it with these fixes; package version checks cannot distinguish local builds with identical versions.

Append scans support with_shard() and with_slice() with Rust 0.4 or newer, which plans positional selection directly into native-readable splits; primary-key scans support bucket-based with_shard(). Both append and data-evolution position selection require the binding's TableScan.with_row_position_slice() and with_row_position_shard(). Ordinary append positions follow the stats-pruned split/file order, while Data Evolution assigns positions before group pruning. Selection occurs before reader filtering and deletion vectors, so surviving row counts can differ between shards. Limits are applied after shard/slice selection.

Timestamp incremental scans require ReadBuilder.new_incremental_scan() and stream-aware splits exposing Split.is_streaming(). Python resolves (start_timestamp, end_timestamp] to snapshot IDs; Rust packs the selected APPEND deltas into one plan. Equal timestamp bounds produce an empty result, as in Java. AUTO selects physical changelog files when a changelog producer is configured; each split carries the selected ending snapshot ID. Numeric bounds use signed 64-bit milliseconds and Java's integer syntax. Continuous streaming uses the same native path for initial and delta frames. When changelog-producer is enabled, follow-up frames request Rust's explicit changelog mode and read the physical changelog manifests. OVERWRITE changelog frames retain per-snapshot Python planning because Java streaming reads them while Java and Rust range-based incremental scans skip OVERWRITE; the resulting splits can still use native reads. Like Java, readers retain physical change events, including repeated primary keys and retracts across commits. They do not merge the window into a final table state or apply endpoint deletion vectors or global indexes. Other commit kinds are excluded; the ending snapshot still supplies plan metadata. Rebuild development wheels from Rust main to obtain this contract.

scan.version supports tags, snapshot IDs and watermark-<value>, resolving tags first and using the historical schema. Ordinary postpone-bucket batch scans can use native planning and exclude pending files in negative buckets.

Dynamic and cross-partition primary-key buckets support native planning, including bucket sharding. Cross-partition key migration is maintained by the writer's index. Batch first-row scans follow Java and exclude un-compacted level-0 files; they can use native planning. With deletion vectors, batch scans exclude level 0 unless deletion-vectors.merge-on-read=true, in which case overlapping key ranges stay together when they include L0 and require reader-side merging. Fully materialized DV files across levels use raw splits, including first-row clustering tables. First-row L0 runs can use native planning, including plans with materialized files in separate raw splits. Plans that require merging clustered materialized files still fall back to Python. Readers preserve physical row positions until deletion vectors are applied, then evaluate residual predicates after merging. Write scans and incremental scans retain level 0.

Append and data-evolution chunk shuffle use Rust file and deletion-vector planning. Python retains live-row chunk sizing, seeded shuffle order and balanced worker assignment, so the same seed selects the same chunks with either planner. Projection does not remove aligned column files before chunk construction. Chunk shuffle supports partition predicates, deletion vectors and timestamp incremental scans; its existing restrictions on limits, slices, row ranges and global-index results still apply.

Scored global-index results on data-evolution append tables use native row-range planning; Python attaches scores to the selected ranges and reads the data. Primary-key sorted indexes refine native batch splits through Python's existing index reader, preserving merge-required splits and the selected snapshot.

Query authorization, first-row plans mixing L0 with merge-required materialized files, and precomputed primary-key global-index results still use the Python planner. Continuous streaming retains Python polling and resume handling, while its initial, delta and physical-changelog frames can use native planning and reads. Write planning retains its Python entrypoint. Native planning remains optional and is disabled by default.

FileIO merges nearby BLOB ranges before reading. Set file-io.read-coalesce.max-gap and file-io.read-coalesce.max-block in the catalog or connection options to tune the 1 MiB and 8 MiB defaults:

import pypaimon.multimodal as pmm

connection = pmm.connect(options={
    "warehouse": "/tmp/warehouse",
    "file-io.read-coalesce.max-gap": "64 kb",
    "file-io.read-coalesce.max-block": "16 mb",
})

max-block constrains coalescing, but does not split an individual BLOB range. A single read can therefore exceed this value.

Install the optional dependency, then import a local directory, FileIO URI, or Hugging Face repository:

pip install 'pypaimon[lerobot]'

import pypaimon.multimodal as pmm

connection = pmm.connect(options={"warehouse": "/tmp/warehouse"})
connection.load_from_lerobot(
    "robot_data",
    "/data/lerobot_dataset",
)

The source dataset must be non-empty. Its schema comes from meta/info.json. Each frame becomes one row; media uses BLOB columns. The import creates frame, Episode, task, info, and optional stats/subtask tables. Info and stats use key STRING, value STRING rows, with each value JSON-encoded to preserve nested metadata. Decode values with json.loads.

Before training, pause writes and create a shared tag:

connection.create_lerobot_tag("robot_data", "train-2026-09-07")
frames = connection.get_table("robot_data").scan(
    tag_name="train-2026-09-07").to_arrow()

Read every metadata component with the same tag. Use the tag only after creation succeeds; cross-table tagging is not atomic. Alternatively, pass tag_name to load_from_lerobot to tag the imported snapshots immediately.

HDF5 loading requires Python 3.8 or newer. Install the optional dependency and create the target multimodal table before loading local or remote HDF5 files as one or more Arrow batches:

pip install 'pypaimon[hdf5,vortex]'

import pyarrow as pa
import pypaimon.multimodal as pmm

EMBEDDING_VECTOR_TYPE = pa.list_(pa.float32(), 3)
IMAGE_BLOB_TYPE = pa.large_binary()

schema = pa.schema([
    pa.field("episode_id", pa.string(), nullable=False),
    pa.field("frame_index", pa.int32(), nullable=False),
    # Arrow fixed-size lists map to Paimon VECTOR columns.
    pa.field("embedding", EMBEDDING_VECTOR_TYPE, nullable=False),
    # Arrow binary and large-binary values map to Paimon BLOB columns.
    pa.field("image", IMAGE_BLOB_TYPE),
])


def transform(h5, source):
    episode_id = source.stem
    for begin in range(0, len(h5["embedding"]), 128):
        end = min(begin + 128, len(h5["embedding"]))
        yield pa.RecordBatch.from_pydict({
            "episode_id": [episode_id] * (end - begin),
            "frame_index": list(range(begin, end)),
            "embedding": h5["embedding"][begin:end].tolist(),
            "image": [bytes(value) for value in h5["image"][begin:end]],
        }, schema=schema)

connection = pmm.connect(options={"warehouse": "/tmp/warehouse"})
frames = connection.create_table(
    "frames",
    schema=schema,
)
result = connection.load_from_hdf5(
    "frames", "/data/episodes", transform=transform)
print(result.file_count, result.batch_count, result.row_count, result.snapshot_id)

load_from_hdf5 accepts one .h5/.hdf5 file, an iterable of paths, or directories that are searched recursively. Paths are resolved, duplicate files within the call are removed, and the remaining files are processed in sorted order. Every yielded batch must have exactly the target columns and be safely convertible to the table schema; missing or extra columns, nulls for non-nullable fields, incompatible types, and invalid fixed-size vector lengths fail the call.

Remote hdfs://, viewfs://, oss://, s3://, and gs:// sources use PyPaimon's FileIO abstraction. Pass source-only credentials and endpoints via source_options={"fs.oss.endpoint": "...", ...}; target warehouse FileIO settings are deliberately not reused. h5py reads the seekable FileIO stream directly without a local temporary download. Legacy OSS with PyArrow before 16 supports explicit files but requires Jindo or a newer PyArrow for recursive directory discovery. In transforms, source.local_path returns a decoded Path for local sources (including spaces and Unicode) and None for remote sources.

An empty path iterable or an existing directory without HDF5 files returns zero counts and snapshot_id=None without creating a writer or snapshot. Nonexistent paths, unsupported file suffixes, and discovered files whose transform produces no rows remain errors.

All files in one call use one writer and one commit, so success creates one snapshot. The API is append-only: it does not add provenance columns, keep a source ledger, skip files, or detect drift. Repeating the same call appends the rows again. It is not retry-safe because an exception from the commit can have an unknown result; inspect table state before deciding whether to retry.

ROSBag loading requires Python 3.10 or newer:

pip install 'pypaimon[rosbag]'

Create the target table, then map ROS messages with a user transform:

import pyarrow as pa


schema = pa.schema([
    pa.field("source", pa.string(), nullable=False),
    pa.field("timestamp", pa.int64(), nullable=False),
    pa.field("value", pa.string(), nullable=False),
])
connection.create_table("messages", schema=schema)


def transform(reader, source):
    rows = []
    for connection, timestamp, rawdata in reader.messages():
        message = reader.deserialize(rawdata, connection.msgtype)
        rows.append({
            "source": source.name,
            "timestamp": timestamp,
            "value": message.data,
        })
    return pa.Table.from_pylist(rows)

result = connection.load_from_rosbag(
    "messages",
    "s3://robot-data/recordings",
    transform=transform,
    source_options={"fs.s3.endpoint": "https://s3.example.com"},
)

ROS1 .bag, ROS2 SQLite3/MCAP directories, and standalone ROS2 .mcap files are supported. OSS, S3, HDFS, ViewFS, and GCS URI sources use FileIO and are copied in bounded chunks to a local temporary directory because rosbags requires local paths. Standalone .db3 files are rejected by default; allow_storage_fragment=True imports the one SQLite fragment without claiming that the complete recording is present.

Every source is scanned to EOF before its transform runs. Transform output is strictly checked against the target Arrow schema and stored in a temporary Arrow IPC file. Paimon writers are created only after every source passes, so source, transform, and schema errors do not create Paimon data files. This front-loaded validation reads each recording twice and requires temporary disk space. A successful call commits all sources in one snapshot.

Ray uses the same validation contract. Install both extras and call pypaimon.ray.load_from_rosbag; transformed output is fully materialized in Ray before write_paimon starts:

pip install 'pypaimon[ray,rosbag]'

pypaimon supports HDFS through a pure-protocol client based on hdfs-native (Rust + PyO3). Use it when you want HDFS access without installing Hadoop, a JDK, libhdfs, or wrestling with CLASSPATH / LD_LIBRARY_PATH.

Install with the optional extra:

pip install 'pypaimon[hdfs]'

The native backend requires Python 3.10+ (and is unavailable on Windows). On older interpreters the extra is skipped, so pypaimon still installs — keep using the legacy pyarrow (libhdfs/JVM) backend there via hdfs.client.impl=pyarrow.

For hdfs:// and viewfs:// URIs this backend is now the default. Switch back to the legacy libhdfs (JNI) path with:

catalog = CatalogFactory.create({
    "warehouse": "hdfs://ns1/warehouse",
    "hdfs.client.impl": "pyarrow",   # default: "native"
})

The client still needs to know about NameNode addresses, HA failover groups, and viewfs mount tables. Three options:

1. Local xml — set HADOOP_CONF_DIR (or the hdfs.conf-dir option) to a directory containing core-site.xml / hdfs-site.xml. Only the xml is required; no Hadoop binaries or JDK.

2. Catalog options (REST-friendly) — pass the original Hadoop key/values directly in catalog options. Keys with prefixes dfs., fs., hadoop., ipc., io. are forwarded as-is. A REST catalog can deliver these in its response, giving a fully zero-file client experience:

   CatalogFactory.create({
       "warehouse": "viewfs://cluster/warehouse",
       "dfs.nameservices": "ns1",
       "dfs.ha.namenodes.ns1": "nn1,nn2",
       "dfs.namenode.rpc-address.ns1.nn1": "host-1:8020",
       "dfs.namenode.rpc-address.ns1.nn2": "host-2:8020",
       "fs.viewfs.mounttable.cluster.link./prod": "hdfs://ns1/prod",
   })

3. Namespaced overrides — use hdfs.config.<key> to forward any other Hadoop key not covered by the prefix whitelist.

The three sources can be combined; catalog options take precedence over xml.

A secured cluster still needs the GSSAPI system library (libgssapi-krb5-2 on Debian/Ubuntu, krb5 via Homebrew on macOS, krb5-libs on RHEL) plus a krb5.conf. Provide credentials by either:

• Running kinit yourself and pointing KRB5CCNAME at the cache, or
• Setting security.kerberos.login.principal and security.kerberos.login.keytab in catalog options — pypaimon will run kinit for you.

If the native backend fails to initialise (e.g. wheel missing on an unsupported platform such as Windows), pypaimon automatically falls back to the pyarrow (libhdfs/JVM) path and logs a warning. Disable the fallback with hdfs.client.fallback-to-pyarrow=false if you want hard failures instead.

Native vector indexes (ivf-flat, ivf-pq, ivf-sq, ivf-rq, and diskann) read multiple file ranges concurrently when the input stream supports thread-safe positional reads. Set the table option vindex.read.parallelism to a positive integer to control the maximum number of concurrent reads per index reader, including reads from concurrent native query callbacks.

The default is 4 for remote index paths and 1 for local paths (including file://). Setting it to 1 disables range-level concurrency. Streams that only support seek and read remain serialized. Workers are created lazily and released when the index reader closes; separate readers have separate budgets. This option controls index I/O, not shard search or native compute threads.

The native vector index writer submits training vectors in bounded batches. <index-type>.train.sample-ratio (or its field-level override) still selects the same evenly spaced non-null vectors in the same order. Native training receives the final corpus size for automatic IVF sizing. This bounds Python training buffers; native training and index construction have their own memory requirements.

Raw vector fallback and refinement score regular FLOAT vectors in bounded blocks using NumPy. List, large-list and fixed-size-list Arrow arrays are supported, including slices and multiple chunks. Null or unsupported blocks use the scalar path. Candidate filters are applied before scoring.

L2 and cosine retain scalar accumulation order. Inner product retains Python sum semantics, including its behavior on newer Python versions. Existing Top-K tie-breaking rules are preserved. The same scoring path is used for raw and refined primary-key vector results.