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

The Hudi-rs project aims to standardize the core Apache Hudi APIs, and broaden the Hudi integration in the data ecosystems for a diverse range of users and projects.

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

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.

Reading Tables

Note

These examples expect a Hudi table exists at /tmp/trips_table, created using the quick start guide.

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?;

Supported timestamp formats

The supported formats for the timestamp argument are:

• Hudi Timeline format (highest matching precedence): yyyyMMddHHmmssSSS or yyyyMMddHHmmss.
• Unix epoch time in seconds, milliseconds, microseconds, or nanoseconds.
• RFC 3339 / ISO 8601 with timezone offset, including:
  • yyyy-MM-dd'T'HH:mm:ss.SSS+00:00
  • yyyy-MM-dd'T'HH:mm:ss.SSSZ
  • yyyy-MM-dd'T'HH:mm:ss+00:00
  • yyyy-MM-dd'T'HH:mm:ssZ

Timestamp strings without a timezone offset (for example yyyy-MM-dd'T'HH:mm:ss) and date-only strings (for example yyyy-MM-dd) are not accepted.

Incremental Query

Incremental query reads the changed data from the table for a given time range.

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 the latest commit)
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 the latest commit)
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 (Rust)

hudi-rs includes a native, single-node write path that is storage-compatible with the Apache Hudi Spark writer: tables written here are readable — and serviceable (compaction, clustering, cleaning) — by Spark, and vice versa. 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.

Note

Write APIs are currently Rust-only; Python and C++ bindings expose the read APIs shown above.

Create a Table

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.

use arrow_array::RecordBatch;

let batch: RecordBatch = /* columns: uuid, city, ts, fare, ... */
# unimplemented!();
table.append([batch]).await?;

Upsert

upsert tags incoming records against the record-level index: existing keys update their file group in place (a rewritten base file on COW, a log file on MOR); new keys pack into existing small files first and then form new file groups.

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

Update and Delete

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

// 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?;

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 (snapshot only), 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 (defaults to earliest…latest).
• 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.

Stage	API	Description
Query planning	get_file_slices(options)	Get the file slices the read targets, dispatched on options.query_type. To bucket for parallel reads, call hudi::util::collection::split_into_chunks on the result.
	compute_table_stats(options)	Estimated (num_rows, byte_size) for scan planning, derived from the metadata table. Snapshot only. Returns None for incremental queries, and whenever the estimate cannot be computed (no metadata table, non-Parquet base files, footer sampling failure).
Query execution	create_file_group_reader_with_options(read_options, extra_storage_overrides)	Create a file group reader with the table's configs. In Python both args are optional; in Rust read_options is an Option and extra_storage_overrides takes a (possibly empty) iterator. Timestamps are resolved automatically (e.g. AsOfTimestamp → EndTimestamp), so callers can pass the same options used for get_file_slices.
	read(options) / read_stream(options)	Record-read APIs. Dispatch on options.query_type. read_stream errors on Incremental for now. Per-slice streaming lives on FileGroupReader.

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.

Config	Default	Scope	Notes
hoodie.read.query.type	snapshot	per read	snapshot or incremental.
hoodie.read.as.of.timestamp	latest commit	per read	Snapshot time-travel point.
hoodie.read.start.timestamp / hoodie.read.end.timestamp	earliest / latest	per read	Incremental window, half-open (start, end].
hoodie.read.file.group.reader.version	2	table or read	Which file group reader merges a slice. Version 2 is the default; a read it cannot serve falls back to version 1.
hoodie.read.use.read_optimized.mode	false	table or read	Read base files only, skipping the log files, on MOR tables.
hoodie.read.stream.batch_size	1024	table or read	Rows per batch for streaming reads.
hoodie.read.file.slice.read.concurrency	4	table or read	File slices read concurrently.
hoodie.read.scan.max.memory.size	unset	table or read	Total bytes a whole scan may use for concurrent slice reads; when set, the concurrency is derived from it. Reaching the limit lowers throughput, it never fails the read.
hoodie.read.input.partitions	0	table or read	How many partitions the DataFusion table provider buckets the file slices into. 0 defers to DataFusion's target_partitions. On the parquet-source path (copy-on-write, read-optimized merge-on-read) DataFusion may split a bucket further once it exceeds datafusion.optimizer.repartition_file_min_size, unless datafusion.optimizer.repartition_file_scans is off.
hoodie.merge.use.record.positions	false	table or read	Match a log record to the base row it updates by position rather than by record key. Honored by reader version 2 only; a log block written without positions is merged by key regardless.

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().

Stage	API	Description
Query execution	read_file_slice()	Read records from a given file slice; based on the configs, read records from only base file, or from base file and log files, and merge records based on the configured strategy.
	read_file_slice_from_paths()	Read records from an explicit base file path and a list of log file paths. Pass an empty log path list to read just the base file.
	read_file_slice_stream()	Streaming version of read_file_slice(). A base-file-only or read-optimized slice streams straight from the base file. A MOR slice with log files streams merged chunks under file group reader version 2 (the default), and collapses to a single merged batch under version 1, whose merge has no incremental form.
	read_file_slice_from_paths_stream()	Streaming version of read_file_slice_from_paths().

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.

hudi-rs	DataFusion	arrow	Python datafusion package
0.6 (unreleased)	55	59	>=55.0.0,<56
0.5	54	58	>=54.0.0,<55
0.4	46	54	not available
0.3	43	53	not available

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

Add crate hudi with datafusion feature to your application to query a Hudi table.

cargo new my_project --bin && cd my_project
cargo add tokio@1 datafusion@55
cargo add hudi --features datafusion

Update src/main.rs with the code snippet below then cargo run.

Add python hudi with datafusion feature to your application to query a Hudi table.

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(())
}

Contributing

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