hudi 0.6.0

The native Rust implementation for Apache Hudi, with C++ & Python API bindings.

hudi-rs is a native Rust implementation of Apache Hudi with Python and C++ bindings. It creates, writes, and reads Hudi tables from Rust and Python in a single process, with no JVM.

Source	Downloads	Installation Command
PyPi.org		pip install hudi
Crates.io		cargo add hudi

Features

Quick start

Create a table, upsert and delete rows, then read what changed after the first commit.

Python

import tempfile

import pyarrow as pa
from hudi import HudiQueryType, HudiReadOptions, HudiTable

def docs(ids, version):
    return pa.table({
        "doc_id": ids,
        "title": [f"{d} v{version}" for d in ids],
        "version": pa.array([version] * len(ids), pa.int64()),
    })

table = HudiTable.create(f"{tempfile.mkdtemp()}/docs", record_key="doc_id", ordering_fields="version")
first = table.append(docs(["doc1", "doc2", "doc3"], version=1))

result = table.upsert(docs(["doc2", "doc4"], version=2))
print(result.num_updates, result.num_inserts)  # 1 1

table.delete("doc_id = 'doc3'")

# Commits that completed after the first one.
changed = table.read(
    HudiReadOptions()
    .with_query_type(HudiQueryType.Incremental)
    .with_start_timestamp(first.completion_time)
)
print(sorted(pa.Table.from_batches(changed).column("doc_id").to_pylist()))  # ['doc2', 'doc4']

snapshot = pa.Table.from_batches(table.read())
print(sorted(snapshot.column("doc_id").to_pylist()))  # ['doc1', 'doc2', 'doc4']

Rust

cargo add hudi
cargo add arrow@59
cargo add tokio@1 --features macros,rt-multi-thread

use std::sync::Arc;

use arrow::array::{Int64Array, RecordBatch, StringArray};
use arrow::datatypes::{DataType, Field, Schema};
use hudi::error::Result;
use hudi::table::{QueryType, ReadOptions, Table};

fn docs(ids: &[&str], version: i64) -> RecordBatch {
    let schema = Arc::new(Schema::new(vec![
        Field::new("doc_id", DataType::Utf8, false),
        Field::new("title", DataType::Utf8, false),
        Field::new("version", DataType::Int64, false),
    ]));
    let titles: Vec<String> = ids.iter().map(|d| format!("{d} v{version}")).collect();
    RecordBatch::try_new(
        schema,
        vec![
            Arc::new(StringArray::from(ids.to_vec())),
            Arc::new(StringArray::from(titles)),
            Arc::new(Int64Array::from(vec![version; ids.len()])),
        ],
    )
    .unwrap()
}

#[tokio::main]
async fn main() -> Result<()> {
    let base_uri = std::env::temp_dir().join(format!("docs_{}", std::process::id()));
    let mut table = Table::create(base_uri.to_str().unwrap())
        .with_table_name("docs")
        .with_record_key_fields(["doc_id"])
        .with_ordering_fields(["version"])
        .create()
        .await?;
    let first = table.append([docs(&["doc1", "doc2", "doc3"], 1)]).await?;

    let result = table.upsert([docs(&["doc2", "doc4"], 2)]).await?;
    println!("{} {}", result.num_updates, result.num_inserts); // 1 1

    table.delete("doc_id = 'doc3'").await?;

    // Commits that completed after the first one.
    let options = ReadOptions::new()
        .with_query_type(QueryType::Incremental)
        .with_start_timestamp(first.completion_time.unwrap());
    let changed = table.read(&options).await?;
    let snapshot = table.read(&ReadOptions::new()).await?;
    let num_rows = |batches: &[RecordBatch]| batches.iter().map(|b| b.num_rows()).sum::<usize>();
    println!("{} {}", num_rows(&changed), num_rows(&snapshot)); // 2 3
    Ok(())
}

See Writing Tables for the full write API, and crates/core/examples/table_writes.rs (cargo run -p hudi-core --example table_writes) for a runnable Rust tour of it. The Python/Rust quick start guide on the Hudi website goes further, with DuckDB, Polars, and DataFusion queries. API references are listed under Documentation.

Requirements

• Python 3.10 or later. Wheels are published to PyPI.
• Rust 1.94.1 or later.
• The hudi crate carries two features: datafusion (off by default, see Apache DataFusion) and spill-rocksdb (on by default), the merge map's on-disk tier, which a merge-on-read merge spills to when a file group's log records exceed hoodie.memory.merge.max.size. RocksDB is built from source with bindgen, so it needs libclang and a C++ toolchain; default-features = false drops it, and a merge that would have spilled then fails instead.

Documentation

• Quick start: the Python and Rust quick start guide on the Hudi website, and the shorter quick start above.
• Rust API reference: docs.rs/hudi (the hudi crate re-exports hudi-core and, with the datafusion feature, hudi-datafusion).
• Python API reference: the package ships type stubs with docstrings (python/hudi/_internal.pyi), which IDEs and type checkers read.
• Runnable Rust example: crates/core/examples/table_writes.rs.
• Reader semantics (read options, filters, guarantees): docs/reader-spec.md.
• Writer design: docs/writer-design.md.
• Packages: PyPI and crates.io. Source: github.com/apache/hudi-rs. Contributing: CONTRIBUTING.md.
• Hudi concepts (timeline, table types, query types): hudi.apache.org/docs/overview.

Reading Tables

For the full reader API reference (ReadOptions, filter expressions, behavioral guarantees), see docs/reader-spec.md.

Snapshot Query

Snapshot query reads the latest version of the data from the table. The table API also accepts column filters that drive partition + file pruning and row-level filtering.

Python

from hudi import HudiReadOptions, HudiTableBuilder
import pyarrow as pa

hudi_table = HudiTableBuilder.from_base_uri("/tmp/trips_table").build()
batches = hudi_table.read(
    HudiReadOptions(filters=[("city", "=", "san_francisco")])
)

# convert to PyArrow table
arrow_table = pa.Table.from_batches(batches)
result = arrow_table.select(["rider", "city", "ts", "fare"])
print(result)

Rust

use hudi::error::Result;
use hudi::table::ReadOptions;
use hudi::table::builder::TableBuilder as HudiTableBuilder;
use arrow::compute::concat_batches;

#[tokio::main]
async fn main() -> Result<()> {
    let hudi_table = HudiTableBuilder::from_base_uri("/tmp/trips_table").build().await?;
    let options = ReadOptions::new().with_filters([("city", "=", "san_francisco")])?;
    let batches = hudi_table.read(&options).await?;
    let batch = concat_batches(&batches[0].schema(), &batches)?;
    let columns = vec!["rider", "city", "ts", "fare"];
    for col_name in columns {
        let idx = batch.schema().index_of(col_name).unwrap();
        println!("{col_name}: {:?}", batch.column(idx));
    }
    Ok(())
}

To run read-optimized (RO) query on Merge-on-Read (MOR) tables, set hoodie.read.use.read_optimized.mode in ReadOptions.

Python

from hudi import HudiReadOptions

batches = hudi_table.read(
    HudiReadOptions(hudi_options={"hoodie.read.use.read_optimized.mode": "true"})
)

Rust

let options = ReadOptions::new()
    .with_hudi_option("hoodie.read.use.read_optimized.mode", "true");
let batches = hudi_table.read(&options).await?;

Time-Travel Query

Time-travel query reads the data at a specific timestamp from the table. The table API also accepts column filters that drive partition + file pruning and row-level filtering.

Python

batches = hudi_table.read(
    HudiReadOptions(filters=[("city", "=", "san_francisco")])
    .with_as_of_timestamp("20241231123456789")
)

Rust

let options = ReadOptions::new()
    .with_as_of_timestamp("20241231123456789")
    .with_filters([("city", "=", "san_francisco")])?;
let batches = hudi_table.read(&options).await?;

Incremental Query

Incremental query reads the changed data from the table for a given time range, half-open (start, end]. On table version 8 and later (timeline layout v2) the bounds are commit completion times, so start from a write result's completion_time; on older tables they are requested times. The end defaults to everything committed so far.

Python

from hudi import HudiQueryType

# read the records between t1 (exclusive) and t2 (inclusive)
batches = hudi_table.read(
    HudiReadOptions()
    .with_query_type(HudiQueryType.Incremental)
    .with_start_timestamp(t1)
    .with_end_timestamp(t2)
)

# read the records after t1 (end defaults to everything committed so far)
batches = hudi_table.read(
    HudiReadOptions()
    .with_query_type(HudiQueryType.Incremental)
    .with_start_timestamp(t1)
)

# with column filters applied to the changed records
batches = hudi_table.read(
    HudiReadOptions(filters=[("city", "=", "san_francisco")])
    .with_query_type(HudiQueryType.Incremental)
    .with_start_timestamp(t1)
    .with_end_timestamp(t2)
)

Rust

use hudi::table::QueryType;

// read the records between t1 (exclusive) and t2 (inclusive)
let options = ReadOptions::new()
    .with_query_type(QueryType::Incremental)
    .with_start_timestamp(t1)
    .with_end_timestamp(t2);
let batches = hudi_table.read(&options).await?;

// read the records after t1 (end defaults to everything committed so far)
let options = ReadOptions::new()
    .with_query_type(QueryType::Incremental)
    .with_start_timestamp(t1);
let batches = hudi_table.read(&options).await?;

// with column filters applied to the changed records
let options = ReadOptions::new()
    .with_query_type(QueryType::Incremental)
    .with_start_timestamp(t1)
    .with_end_timestamp(t2)
    .with_filters([("city", "=", "san_francisco")])?;
let batches = hudi_table.read(&options).await?;

Incremental queries support the same timestamp formats as time-travel queries.

Streaming Read

Streaming reads yield RecordBatches one at a time without loading the full result into memory. The same ReadOptions knobs apply, plus batch_size and projection.

Python

options = (
    HudiReadOptions(
        filters=[("city", "=", "san_francisco")],
        projection=["rider", "city", "ts", "fare"],
    )
    .with_batch_size(4096)
)
for batch in hudi_table.read_stream(options):
    print(batch.num_rows)

Rust

use futures::StreamExt;

let options = ReadOptions::new()
    .with_filters([("city", "=", "san_francisco")])?
    .with_projection(["rider", "city", "ts", "fare"])
    .with_batch_size(4096)?;
let mut stream = hudi_table.read_stream(&options).await?;
while let Some(batch) = stream.next().await {
    let batch = batch?;
    println!("{}", batch.num_rows());
}

File Group Reading (Experimental)

File group reading allows you to read data from a specific file slice. This is useful when integrating with query engines, where the plan provides file paths.

Python

from hudi import HudiFileGroupReader

reader = HudiFileGroupReader(
    "/table/base/path", {"hoodie.read.start.timestamp": "0"})

# Returns a PyArrow RecordBatch
record_batch = reader.read_file_slice_from_paths("relative/path.parquet", [])

Rust

use hudi::file_group::reader::FileGroupReader;
use hudi::table::ReadOptions;

// Inside an async context
let reader = FileGroupReader::new_with_options(
    "/table/base/path", [("hoodie.read.start.timestamp", "0")]).await?;

// Returns an Arrow RecordBatch
let record_batch = reader
    .read_file_slice_from_paths(
        "relative/path.parquet",
        Vec::<&str>::new(),
        &ReadOptions::new(),
    )
    .await?;

C++

#include "cxx.h"
#include "src/lib.rs.h"
#include "arrow/c/abi.h"

// Functions may throw rust::Error on failure
auto reader = new_file_group_reader_with_options(
    "/table/base/path", {"hoodie.read.start.timestamp=0"});

// Returns an ArrowArrayStream pointer
std::vector<std::string> log_file_paths{};
ArrowArrayStream* stream_ptr = reader->read_file_slice_from_paths("relative/path.parquet", log_file_paths);

Writing Tables

hudi-rs includes a native, single-node write path that is storage-compatible with the Apache Hudi Spark writer: Spark reads tables written here and runs its table services on them (compaction, clustering, cleaning), and hudi-rs reads and writes tables Spark wrote. Writes target COPY_ON_WRITE and MERGE_ON_READ tables at table version 9 (Hudi 1.1.x; version 8 via with_table_version(8)), and maintain the metadata table (file listings, column/partition stats, record-level index) on every commit. See docs/writer-design.md for the full design.

Python exposes the same write APIs on HudiTable, accepting PyArrow tables, record batches and readers, or any object implementing the Arrow PyCapsule stream interface (__arrow_c_stream__, e.g. polars and pandas >= 2.2 DataFrames). Each write returns a HudiWriteResult. The C++ binding exposes the read APIs only.

Create a Table

Python

from hudi import HudiTable

table = HudiTable.create(
    "/tmp/trips_table",
    record_key="uuid",
    partition_by="city",
    ordering_fields="ts",     # event-time ordering
    table_type="cow",         # or "mor"
)

Rust

use hudi::table::Table;
use hudi::config::table::TableTypeValue;

let mut table = Table::create("/tmp/trips_table")
    .with_table_name("trips")
    .with_table_type(TableTypeValue::MergeOnRead)   // or CopyOnWrite
    .with_record_key_fields(["uuid"])
    .with_partition_fields(["city"])
    .with_ordering_fields(["ts"])                   // event-time ordering
    .create()
    .await?;

Append

append behaves like Hudi's bulk insert: it writes new file groups without an index lookup, splitting input into files bounded by hoodie.parquet.max.file.size. On tables below version 8 it takes a narrow path: unpartitioned append-only tables with hoodie.populate.meta.fields=false and no metadata table, one base file per call.

Python

import pyarrow as pa

# `table` is the HudiTable created above; the input is plain Arrow data.
batch = pa.record_batch(
    {"uuid": ["a", "b"], "city": ["san_francisco"] * 2, "ts": [1, 1], "fare": [30.0, 45.0]}
)
table.append(batch)

Rust

use arrow::array::RecordBatch;

let batch: RecordBatch = todo!("columns: uuid, city, ts, fare, ...");
table.append([batch]).await?;

Upsert

upsert tags incoming records against the table's index (the record-level index when enabled, otherwise a scan of the base files' keys): existing keys update their file group in place (a rewritten base file on COW, a log file on MOR). On COW, new keys fill existing small files first and then form new file groups; on MOR, they always form new file groups.

Python

result = table.upsert(batch)
print(f"requested_time={result.requested_time} updates={result.num_updates} inserts={result.num_inserts}")

# Partial upsert (copy-on-write only): matched rows take only these columns from the input.
table.upsert(batch, update_columns=["fare"])

Rust

let result = table.upsert([batch]).await?;
println!(
    "requested_time={} updates={} inserts={}",
    result.requested_time.as_deref().unwrap_or("none"),
    result.num_updates,
    result.num_inserts
);

Update and Delete

A write whose filter or keys match nothing succeeds without creating a commit and reports zero rows.

Python

# SQL-style update: each value must fit its column's type exactly.
table.update("city = 'san_francisco'", {"fare": 99.0})

# Delete by predicate (single comparison, or IN for multiple keys).
table.delete("uuid IN ('a')")

# Delete by key, optionally with the partition path.
table.delete_keys([("b", "city=san_francisco")])

Rust

use hudi::table::HoodieKey;

// SQL-style update: set columns from a one-row batch on matching rows.
table.update("city = 'san_francisco'", updates_batch).await?;

// Delete by predicate (single comparison, or IN for multiple keys).
table.delete("uuid = 'e96c4396-3fad-413a-a942-4cb36106d721'").await?;
table.delete("uuid IN ('a', 'b')").await?;

// Delete by key.
table.delete_keys([HoodieKey {
    record_key: "a".to_string(),
    partition_path: "city=san_francisco".to_string(),
}]).await?;

Overwrite

Python

# INSERT_OVERWRITE_TABLE: replace all file groups.
table.overwrite(batch)

# INSERT_OVERWRITE: replace only the partitions present in the input.
table.dynamic_partition_overwrite(batch)

Rust

// INSERT_OVERWRITE_TABLE: replace all file groups.
table.overwrite([batch]).await?;

// INSERT_OVERWRITE: replace only the partitions present in the input.
table.dynamic_partition_overwrite([batch]).await?;

Compaction (Experimental)

On a merge-on-read table, upserts, updates, and deletes land in log files. compact folds them into new base files: every file group whose latest file slice has log files gets a new base file of its merged rows, under one compaction commit. Rows keep their commit time, so incremental reads see no new rows from a compaction. When no file group has log files, nothing is written. A copy-on-write table has nothing to compact and returns an error.

Compaction can also run after delta commits once the number since the latest compaction reaches hoodie.compact.inline.max.delta.commits (default 5): in the write call with hoodie.compact.inline=true, or on a background task with hoodie.datasource.compaction.async.enable=true, which await_compaction waits for. Both are off by default. The metadata table compacts itself as commits accumulate.

Python

mor = HudiTable.create(
    "/tmp/trips_mor",
    record_key="uuid",
    table_type="mor",
    # Optional: also compact in the write call every 5 delta commits.
    hudi_options={"hoodie.compact.inline": "true"},
)
mor.append(batch)
mor.upsert(batch)  # updates go to log files

result = mor.compact()
print(result.num_file_groups, result.num_log_files, result.num_writes)

Rust

let mut mor = Table::create("/tmp/trips_mor")
    .with_table_name("trips")
    .with_table_type(TableTypeValue::MergeOnRead)
    .with_record_key_fields(["uuid"])
    // Optional: also compact in the write call every 5 delta commits.
    .with_option("hoodie.compact.inline", "true")
    .create()
    .await?;
mor.append([batch.clone()]).await?;
mor.upsert([batch]).await?; // updates go to log files

let result = mor.compact().await?;
println!("{} {} {}", result.num_file_groups, result.num_log_files, result.num_writes);

Behavior Notes

• Every write runs as a full Hudi transaction: instants are requested and completed with completion times, marker files enable automatic rollback of crashed writes, and the timeline is archived into the LSM history once it grows past hoodie.keep.max.commits.
• Writes update all metadata table partitions in the same transaction, so Spark readers with hoodie.metadata.enable=true see consistent listings and stats.
• Write parallelism is configurable with hoodie.write.task.parallelism (default 2 × cores), passed via with_option.

Query Engine Integration

Hudi-rs provides APIs to support integration with query engines. The sections below highlight some commonly used APIs.

Table API

Create a Hudi table instance using its constructor or the TableBuilder API.

All read APIs accept a ReadOptions (Rust) / HudiReadOptions (Python) value. It stores three fields — filters, projection, and hudi_options — and exposes chainable with_* builders for the rest. The available knobs:

• query_type (with_query_type) — Snapshot (default) or Incremental. Drives dispatch in read, read_stream, and get_file_slices.
• filters — column filters as (field, op, value) tuples. The field can be any column (partition or data). Used for partition pruning, file-level stats pruning (copy-on-write and read-optimized reads), file-group selection through the record-level index for =/IN filters on the record key (latest snapshot, not read-optimized), predicate pushdown into the parquet reader, and row-level filtering.
• projection — columns to return. Streaming pushes the projection down to the parquet reader; eager reads project after merging.
• batch_size (with_batch_size) — rows per batch (streaming only; eager reads return one batch per file slice).
• as_of_timestamp (with_as_of_timestamp) — snapshot/time-travel timestamp (defaults to latest commit).
• start_timestamp / end_timestamp (with_start_timestamp / with_end_timestamp) — incremental range, bounding completion times on table version 8 and later (defaults to the beginning of the timeline…everything committed so far).
• hudi_options — Hudi configs for this read (e.g. hoodie.read.use.read_optimized.mode). A config that selects which read to perform — hoodie.read.query.type and the as-of/start/end timestamps — is per-read only and is dropped when set on the table. The rest describe how to read: set them on the table and override them here. See Read configs.

Read configs

Read configs reach a read either through ReadOptions / HudiReadOptions or, for those scoped table or read, through the table (TableBuilder, hoodie.properties, hudi-defaults.conf), where a per-read value wins. A per read config set on the table is dropped: baked in there, it would silently redirect every later read.

A table whose hoodie.record.merge.mode is CUSTOM needs a merger for its payload class. When neither reader has one, the read fails rather than returning wrong rows; set hoodie.read.file.group.reader.version=1 to read it the way it was read before, unless its base files are HFile, which version 1 cannot read.

Base files are read from Parquet, Lance (hoodie.table.base.file.format=lance), and HFile (metadata tables only).

File Group API

Create a Hudi file group reader instance using its constructor or the Hudi table API create_file_group_reader_with_options().

Apache DataFusion

Enabling the hudi crate with datafusion feature will provide a DataFusion extension to query Hudi tables.

Each hudi-rs release targets one DataFusion major, and with it one arrow major. The Rust hudi crate's TableProvider and the Python hudi[datafusion] extra both have to match the DataFusion version of the application they plug into, so pick the hudi-rs release that pairs with yours. Readers that take Arrow data out through the C data interface (Python read(), the C++ and JVM bindings) are not tied to these versions. The hudi[datafusion] extra first shipped in 0.5.

On macOS arm64, a debug build of an application that links hudi with the datafusion feature can fail with ld: B/BL out of range ... (max +/-128MB): unoptimized dependencies push the binary's code section past the arm64 branch range. Optimize dependencies in the dev profile:

[profile.dev.package."*"]
opt-level = 1

This workspace applies the same setting from build-wrapper.sh on macOS arm64 only, since it slows compilation noticeably. It goes away once a Lance release shares the workspace's arrow major (#844).

cargo new my_project --bin && cd my_project
cargo add tokio@1 datafusion@55  # the DataFusion major that pairs with the hudi release, see the table above
cargo add hudi --features datafusion

pip install hudi[datafusion]

Rust

use std::sync::Arc;

use datafusion::error::Result;
use datafusion::prelude::{DataFrame, SessionContext};
use hudi::HudiDataSource;

#[tokio::main]
async fn main() -> Result<()> {
    let ctx = SessionContext::new();
    let hudi = HudiDataSource::new_with_options(
        "/tmp/trips_table",
        [("hoodie.read.input.partitions", "5")]).await?;
    ctx.register_table("trips_table", Arc::new(hudi))?;
    let df: DataFrame = ctx.sql("SELECT * from trips_table where city = 'san_francisco'").await?;
    df.show().await?;
    Ok(())
}

Python

from datafusion import SessionContext
from hudi import HudiDataFusionDataSource

table = HudiDataFusionDataSource(
    "/tmp/trips_table", [("hoodie.read.input.partitions", "5")]
)
ctx = SessionContext()
ctx.register_table("trips", table)
ctx.sql("SELECT max(fare), city from trips group by city order by 1 desc").show()

Other Integrations

Hudi is also integrated with

• Daft
• Ray

Work with cloud storage

Ensure cloud storage credentials are set properly as environment variables, e.g., AWS_*, AZURE_*, or GOOGLE_*. Relevant storage environment variables will then be picked up. The target table's base uri with schemes such as s3://, az://, or gs:// will be processed accordingly.

Alternatively, you can pass the storage configuration as options via Table APIs.

Python

from hudi import HudiTableBuilder

hudi_table = (
    HudiTableBuilder
    .from_base_uri("s3://bucket/trips_table")
    .with_option("aws_region", "us-west-2")
    .build()
)

Rust

use hudi::error::Result;
use hudi::table::builder::TableBuilder as HudiTableBuilder;

#[tokio::main]
async fn main() -> Result<()> {
    let hudi_table = HudiTableBuilder::from_base_uri("s3://bucket/trips_table")
        .with_option("aws_region", "us-west-2")
        .build().await?;
    Ok(())
}

Project layout

Contributing

Check out the contributing guide for all the details about making contributions to the project.