paimon-python/README.md

![Paimon](https://github.com/apache/paimon/blob/master/docs/static/paimon-simple.png)

[![License](https://img.shields.io/badge/license-Apache%202-4EB1BA.svg)](https://www.apache.org/licenses/LICENSE-2.0.html)

# PyPaimon

This PyPi package contains the Python APIs for using Paimon.

# Version

Pypaimon requires Python 3.6+.

# Build

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

Normal build:

```shell
python -m pip install --group build
python -m build
```

Development build (editable installation with development dependencies):

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

Both produce a source archive and wheel in `dist/`.

# OSS metadata commits

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`:

| Option | Behavior |
| --- | --- |
| `fs.oss.server-side-encryption` | `AES256`, `KMS` or `SM4` (case-insensitive) |
| `fs.oss.server-side-encryption-key-id` | KMS key ID; implies `KMS` if the method is unset |
| `fs.oss.server-side-data-encryption` | `SM4` with `KMS`; implies `KMS` if the method is unset |
| `fs.oss.server-side-encryption-algorithm` | Legacy method fallback, used only if all three options above are unset |

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.

# Row ID column updates

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:

```python
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)`.

# Parquet page-index reads

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:

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

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

# Native write and commit

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

```python
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 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.
Non-self table sources 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.

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.

```python
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.

# Native commit

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

```python
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:

```python
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](https://github.com/apache/paimon-rust/pull/916), built on
[#915](https://github.com/apache/paimon-rust/pull/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.

# Native scan planning

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:

```python
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:

```python
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. 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.

# Coalesced BLOB reads

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:

```python
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.

# Load LeRobot Dataset v3

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

```commandline
pip install 'pypaimon[lerobot]'
```

```python
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:

```python
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 to multimodal tables

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:

```commandline
pip install 'pypaimon[hdf5,vortex]'
```

```python
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 to multimodal tables

ROSBag loading requires Python 3.10 or newer:

```commandline
pip install 'pypaimon[rosbag]'
```

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

```python
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:

```commandline
pip install 'pypaimon[ray,rosbag]'
```

# HDFS without a local Hadoop install

`pypaimon` supports HDFS through a pure-protocol client based on
[`hdfs-native`](https://github.com/Kimahriman/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:

```commandline
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:

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

## Sourcing the cluster wiring

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:

   ```python
   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.

## Kerberos

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.

## Fallback behaviour

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.


# Vector index range reads

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.


# Native vector index training

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.


# Vector fallback scoring and refinement

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.