The Python integration is a binding built on top of Apache Paimon Rust, allowing you to access Paimon tables from Python programs. It uses PyArrow for zero-copy data transfer via the Arrow C Data Interface.
pip install pypaimon-rust pyarrow
The pre-built native library is embedded in the package and automatically loaded at runtime — no manual build step is needed. PyArrow is a required peer dependency and must be installed separately.
Use PaimonCatalog with a dictionary of options to create a catalog. The catalog type is determined by the metastore option (default: filesystem).
from pypaimon_rust.datafusion import PaimonCatalog # Local filesystem catalog = PaimonCatalog({"warehouse": "/path/to/warehouse"}) # List databases and tables print(catalog.list_databases()) print(catalog.list_tables("default")) # Get a table handle table = catalog.get_table("default.my_table")
catalog = PaimonCatalog({ "warehouse": "oss://bucket/warehouse", "fs.oss.accessKeyId": "your-access-key-id", "fs.oss.accessKeySecret": "your-access-key-secret", "fs.oss.endpoint": "oss-cn-hangzhou.aliyuncs.com", })
catalog = PaimonCatalog({ "metastore": "rest", "uri": "http://localhost:8080", "warehouse": "my_warehouse", })
SQLContext supports registering multiple Paimon catalogs and executing SQL queries with DataFusion.
from pypaimon_rust.datafusion import SQLContext ctx = SQLContext() ctx.register_catalog("paimon", {"warehouse": "/path/to/warehouse"}) # DDL and DML ctx.sql("CREATE SCHEMA paimon.my_db") ctx.sql("CREATE TABLE paimon.my_db.t (id INT, name STRING)") ctx.sql("INSERT INTO paimon.my_db.t VALUES (1, 'alice'), (2, 'bob')") # Query returns a list of PyArrow RecordBatches batches = ctx.sql("SELECT * FROM paimon.my_db.t") for batch in batches: print(batch)
SQLContext can use a bounded DataFusion memory pool and a dedicated temporary directory. This allows spill-capable operators, such as sorts and aggregations, to move intermediate data to disk when execution memory is constrained.
ctx = SQLContext( memory_pool_type="fair", memory_pool_bytes=512 * 1024 * 1024, temp_directory="/tmp/paimon-spill", max_temp_directory_size_bytes=4 * 1024 * 1024 * 1024, )
memory_pool_type accepts "fair" or "greedy" and requires memory_pool_bytes. If only memory_pool_bytes is provided, the greedy pool is used. All arguments are optional, and SQLContext() continues to use DataFusion's default runtime environment.
Memory limits apply to allocations tracked by DataFusion's memory pool. They do not account for every allocation made by the host application or by external libraries.
Paimon Python uses a scan-then-read pattern: first scan the table to produce splits, then read data from those splits as PyArrow RecordBatches.
import pyarrow as pa from pypaimon_rust.datafusion import PaimonCatalog catalog = PaimonCatalog({"warehouse": "/path/to/warehouse"}) table = catalog.get_table("default.my_table") # Create a read builder rb = table.new_read_builder() # Step 1: Scan — produces a Plan containing Splits scan = rb.new_scan() plan = scan.plan() splits = plan.splits() snapshot_id = plan.snapshot_id() # Step 2: Read — consumes splits and returns PyArrow RecordBatches read = rb.new_read() batches = read.read(splits) for batch in batches: print(batch)
plan.snapshot_id() identifies the snapshot used for planning, including when filters or a zero limit produce no splits. It returns None when no snapshot exists or the scan is for a format table without Paimon snapshots.
Use an explicit snapshot range to plan a single incremental batch:
plan = rb.new_incremental_scan(2, 5).plan() batches = rb.new_read().read(plan.splits())
The range is (start_snapshot_id, end_snapshot_id]. The default delta mode uses APPEND delta manifests. Use changelog to read physical changelog manifests, or auto to follow the table's changelog-producer option (delta when it is none, otherwise changelog):
plan = rb.new_incremental_scan(2, 5, "changelog").plan() batches = rb.with_include_row_kind(True).new_read().read(plan.splits())
Explicit diff mode is rejected because a diff contains before/after split pairs and cannot be represented by the ordinary Plan.splits() contract. Files use Java batch split packing, with streaming read semantics: repeated primary keys and physical retracts remain separate rows. Readers do not merge these events into the window's final table state. The end snapshot must exist and supplies snapshot metadata, including for empty results. Snapshot deletion vectors and automatic global indexes are not applied to historical events. Builder filters, projections, and limits still apply.
split.is_streaming() identifies this read contract. split.serialize() exports both batch and streaming splits to Java binary encoding, preserving the streaming flag.
For continuous reads, new_stream_scan() returns a stateful scan that starts from scan.mode. See Streaming Reading for startup modes, checkpoints, and row kinds:
rb = table.new_read_builder({"scan.mode": "from-snapshot", "scan.snapshot-id": "5"}) scan = rb.new_stream_scan() read = rb.with_include_row_kind(True).new_read() plan = scan.plan() if plan is not None: batches = read.read(plan.splits()) # Process batches, then acknowledge the progress. scan.notify_checkpoint_complete(scan.checkpoint())
new_scan() and new_incremental_scan() raise when a time-travel selector resolves to no snapshot, while new_stream_scan() resolves its start at the first plan, so snapshot 5 may also be the next commit or an expired one.
For append and Data Evolution tables, select half-open row positions or one balanced shard on a scan:
plan = rb.new_scan().with_row_position_slice(10, 20).plan() plan = rb.new_scan().with_row_position_shard(1, 4).plan()
For ordinary append tables, positions follow the final stats-pruned split and file order. Row-tracked tables encode selected stable row IDs; tables without row tracking encode positions local to each filtered output split. Unselected files are removed, so the reader does not open them. For Data Evolution, positions count candidate rows before explicit/global-index range pruning, group statistics, projection, and deletion-vector filtering; column updates sharing row IDs count once. Explicit with_row_ranges and index-selected ranges intersect the positions after assignment, even when they exclude earlier files. Slices require start < end; shards require a positive count and 0 <= index < count. Slice and shard selection are mutually exclusive and may also be applied to new_incremental_scan results, where positions count the combined APPEND-delta batch. The selection is encoded in the returned splits and survives serialization and direct native reads.
Deletion-vector reads accept both Java and Python Avro array-item schemas. Historical Python bucket-local references are resolved from the table's index directory when the canonical bucket path is absent. Explicit external paths and existing canonical paths retain priority.
Alternatively, read via SQL using SQLContext:
from pypaimon_rust.datafusion import SQLContext ctx = SQLContext() ctx.register_catalog("paimon", {"warehouse": "/path/to/warehouse"}) batches = ctx.sql("SELECT id, name FROM paimon.default.my_table") for batch in batches: print(batch)
Paimon Python uses a write-then-commit pattern: write PyArrow RecordBatches to a writer, prepare commit messages, then commit.
import pyarrow as pa from pypaimon_rust.datafusion import PaimonCatalog catalog = PaimonCatalog({"warehouse": "/path/to/warehouse"}) table = catalog.get_table("default.my_table") # Build a batch matching the table schema batch = pa.record_batch( [pa.array([1, 2, 3], pa.int32()), pa.array(["a", "b", "c"], pa.string())], names=["id", "name"], ) # Create a write builder (shared commit_user for writer and committer) wb = table.new_batch_write_builder() # Write batches write = wb.new_write() write.write_arrow(batch) # Prepare commit messages messages = write.prepare_commit() # Commit wb.new_commit().commit(messages)
Alternatively, write via SQL using SQLContext:
from pypaimon_rust.datafusion import SQLContext ctx = SQLContext() ctx.register_catalog("paimon", {"warehouse": "/path/to/warehouse"}) ctx.sql("INSERT INTO paimon.default.my_table VALUES (1, 'alice'), (2, 'bob')")
!!! warning “Schema Validation” The input batch schema is strictly validated against the table schema: field count, order, names, and types must match exactly. A ValueError is raised on mismatch.
!!! note “Write Builder Consistency” The writer and committer must come from the same BatchWriteBuilder — they share a commit_user for duplicate-commit detection. Passing messages from one builder‘s writer to another builder’s committer will raise a ValueError.
Use with_projection to select specific columns. Only the requested columns are read, reducing I/O.
rb = table.new_read_builder() rb.with_projection(["id", "name"]) # Continue with scan-then-read as above...
Use with_limit to set a hint for the number of rows returned. A limit of 0 returns zero rows.
rb = table.new_read_builder() rb.with_limit(100)
!!! warning with_limit is a scan-planning hint, not an exact row cap. When all rows fall within a single split, the entire split is returned regardless of the limit value. Callers should apply application-level limiting if an exact upper bound is required.
Use with_row_ranges to restrict a Data Evolution scan to inclusive row ID ranges. Each range is a (from, to) tuple; from must not exceed to.
rb = table.new_read_builder() rb.with_row_ranges([(0, 99), (200, 299)])
!!! warning An empty list selects zero rows, not all rows. Format tables are not supported and raise NotImplementedError.
Use with_case_sensitive to control whether column-name matching in projections and predicates is case-sensitive. Defaults to True (exact match). Set to False for case-insensitive matching (ASCII case-folding).
rb = table.new_read_builder() rb.with_case_sensitive(False)
!!! note with_case_sensitive must be called before with_filter to affect predicate construction. The predicate is built using the case-sensitivity setting at the time with_filter is invoked; changing it afterward has no effect on an already-constructed predicate.
Filter push-down prunes data at two levels:
!!! note Filter push-down at the scan-planning level is best-effort: it may conservatively include files that do not contain matching rows. The read-side applies exact residual filtering and will not return rows that fail the predicate.
Predicates use a lightweight dictionary format. Each leaf node specifies a method, field, and literals:
# Comparison predicates {"method": "equal", "field": "id", "literals": [1]} # id = 1 {"method": "notEqual", "field": "name", "literals": ["bob"]} # name != "bob" {"method": "lessThan", "field": "id", "literals": [3]} # id < 3 {"method": "lessOrEqual", "field": "id", "literals": [2]} # id <= 2 {"method": "greaterThan", "field": "id", "literals": [1]} # id > 1 {"method": "greaterOrEqual", "field": "id", "literals": [2]} # id >= 2 # Null checks {"method": "isNull", "field": "name"} # name IS NULL {"method": "isNotNull", "field": "name"} # name IS NOT NULL # IN / NOT IN {"method": "in", "field": "id", "literals": [1, 2, 3]} # id IN (1, 2, 3) {"method": "notIn", "field": "name", "literals": ["x", "y"]} # name NOT IN ("x", "y") # String predicates {"method": "startsWith", "field": "name", "literals": ["al"]} # name LIKE 'al%' {"method": "endsWith", "field": "name", "literals": ["ce"]} # name LIKE '%ce' {"method": "contains", "field": "name", "literals": ["ic"]} # name LIKE '%ic%' {"method": "like", "field": "name", "literals": ["a%b%c"]} # name LIKE 'a%b%c'
Pass a predicate dict to with_filter on the ReadBuilder:
rb = table.new_read_builder() rb.with_filter({"method": "equal", "field": "id", "literals": [1]}) # Continue with scan-then-read...
Combine predicates with "and" / "or" methods using a "children" list:
# id >= 1 AND id <= 3 rb.with_filter({ "method": "and", "children": [ {"method": "greaterOrEqual", "field": "id", "literals": [1]}, {"method": "lessOrEqual", "field": "id", "literals": [3]}, ] }) # (id = 1 OR id = 2) AND name = "alice" rb.with_filter({ "method": "and", "children": [ { "method": "or", "children": [ {"method": "equal", "field": "id", "literals": [1]}, {"method": "equal", "field": "id", "literals": [2]}, ] }, {"method": "equal", "field": "name", "literals": ["alice"]}, ] })
Literal values are automatically converted from Python types based on the column's declared schema type:
| Python Type | Paimon Type |
|---|---|
bool | Bool |
int | TinyInt / SmallInt / Int / BigInt |
int / float | Float / Double |
str | String |
bytes / bytearray | Binary / VarBinary |
datetime.date | Date |
datetime.time (naive) | Time |
datetime.datetime (naive) | Timestamp |
datetime.datetime (aware) | LocalZonedTimestamp |
decimal.Decimal / int | Decimal |
For temporal types, ensure the Python object matches the column type exactly:
import datetime from decimal import Decimal # Date {"method": "equal", "field": "dt", "literals": [datetime.date(2024, 1, 1)]} # Decimal at scale 2 — int or Decimal accepted {"method": "equal", "field": "amount", "literals": [Decimal("123.45")]} # Naive datetime for TIMESTAMP (no timezone) {"method": "equal", "field": "ts", "literals": [datetime.datetime(2024, 1, 1, 12, 0, 0)]} # Aware datetime for TIMESTAMP WITH LOCAL TIME ZONE from zoneinfo import ZoneInfo {"method": "equal", "field": "ts_ltz", "literals": [ datetime.datetime(2024, 1, 1, 12, 0, 0, tzinfo=ZoneInfo("Asia/Shanghai")) ]}
Use scan options on new_read_builder to read a table at a specific point in time:
# By snapshot ID rb = table.new_read_builder({"scan.snapshot-id": "1"}) # By timestamp (epoch millis) rb = table.new_read_builder({"scan.timestamp-millis": "1700000000000"}) # By timestamp string (process local time zone; fractional seconds are supported) rb = table.new_read_builder({"scan.timestamp": "2024-01-01 12:00:00.123"}) # By version rb = table.new_read_builder({"scan.version": "3"}) # By tag name rb = table.new_read_builder({"scan.tag-name": "release-1.0"})
!!! warning Only one time-travel selector may be set. Providing multiple selectors will raise a ValueError. A selector that resolves to no snapshot raises a ValueError from new_scan() and new_incremental_scan().
scan.timestamp selects the latest snapshot committed at or before the given local time. It accepts YYYY-MM-DD, YYYY-MM-DD HH:MM:SS[.fraction], or YYYY-MM-DDTHH:MM[:SS[.fraction]], with up to nine fractional digits truncated to milliseconds. It can be combined with scan.mode=from-timestamp. Invalid timestamps and times before the earliest available snapshot raise an error.
Inspect snapshots, tags, and partition statistics on a table. A branch-qualified identifier selects that branch's schema and snapshots:
branch_table = catalog.get_table("default.my_table$branch_blue") assert branch_table.branch() == "blue"
table.branch() returns "main" for an ordinary table. latest_snapshot(), list_snapshots(), list_tags(), list_partitions(), and partition_stats() inspect the selected branch, including returning empty metadata for an empty branch.
# Latest snapshot snap = table.latest_snapshot() if snap: print(f"Snapshot {snap.id()} at {snap.commit_time_ms()}") print(f" commit kind: {snap.commit_kind()}") print(f" total records: {snap.total_record_count()}") # All snapshots (newest first) for snap in table.list_snapshots(): print(snap.id(), snap.commit_kind()) # Tags for tag in table.list_tags(): print(tag.name(), tag.snapshot_id()) # Partition stats for stat in table.partition_stats(): print(stat.partition(), stat.record_count(), stat.total_size_bytes())
table.new_compact_builder() plans, executes and commits table compaction in one call, like the CALL sys.compact SQL procedure. Nothing is committed when planning yields no task. When a task fails, the files written by earlier tasks are deleted on a best-effort basis; a failed deletion is reported as a RuntimeWarning before the task's error is raised.
metrics = table.new_compact_builder().execute() print(metrics["tasks"], metrics["files_removed"], metrics["files_added"]) # Restrict compaction to partitions matching a predicate (same predicate # format as read filters), and set the commit user. It defaults to a generated # UUID; the snapshot records it as `<commit_user>-compact-<suffix>`. metrics = ( table.new_compact_builder() .with_partition_filter({"method": "equal", "field": "dt", "literals": ["2026-11-01"]}) .with_commit_user("my-compactor") .execute() )
The returned metrics are a dictionary with the task count and the write statistics of the compaction commit: tasks, files_removed, files_added, rows_written, bytes_written, changelog_files_added, changelog_rows_written and changelog_bytes_written.
Compacting a primary-key table that uses the lookup changelog producer, the first-row merge engine or deletion vectors needs local-cache.enabled=true in the catalog options; without it execute() raises ValueError once there are tasks to run. See Table Compaction for the supported table modes.
Register Python scalar UDFs into a SQLContext:
from pypaimon_rust.datafusion import SQLContext, udf import pyarrow as pa ctx = SQLContext() ctx.register_catalog("paimon", {"warehouse": "/tmp/paimon-warehouse"}) batch = pa.record_batch([[1, None, 3]], names=["id"]) ctx.register_batch("my_temp", batch) def plus_ten(values): # values is a PyArrow Array (not a tuple) return pa.array( [None if value is None else value + 10 for value in values.to_pylist()], type=pa.int64(), ) ctx.register_udf(udf(plus_ten, [pa.int64()], pa.int64(), "volatile", "plus_ten")) batches = ctx.sql( "SELECT plus_ten(id) AS id FROM paimon.default.my_temp ORDER BY id" ) for batch in batches: print(batch)
import pyarrow as pa from pypaimon_rust.datafusion import PaimonCatalog, SQLContext # 1. Write data via SQL ctx = SQLContext() ctx.register_catalog("paimon", {"warehouse": "/tmp/paimon-warehouse"}) ctx.sql("CREATE SCHEMA paimon.wdb") ctx.sql("CREATE TABLE paimon.wdb.t (id INT, name STRING)") # 2. Write using the programmatic API catalog = PaimonCatalog({"warehouse": "/tmp/paimon-warehouse"}) table = catalog.get_table("wdb.t") batch = pa.record_batch( [pa.array([1, 2, 3], pa.int32()), pa.array(["alice", "bob", "carol"], pa.string())], names=["id", "name"], ) wb = table.new_batch_write_builder() write = wb.new_write() write.write_arrow(batch) wb.new_commit().commit(write.prepare_commit()) # 3. Read with projection and filter rb = table.new_read_builder() rb.with_projection(["id", "name"]) rb.with_filter({"method": "greaterThan", "field": "id", "literals": [0]}) scan = rb.new_scan() plan = scan.plan() splits = plan.splits() read = rb.new_read() for batch in read.read(splits): tbl = pa.Table.from_batches([batch]).sort_by("id") print(tbl) # 4. Read via SQL for batch in ctx.sql("SELECT id, name FROM paimon.wdb.t ORDER BY id"): print(batch)