Apache Hudi

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

Area	Supported
Table types	Copy-on-Write (COW), Merge-on-Read (MOR)
Table versions	Read: 6, 8, 9 (Hudi 0.14 and later). Write: 8, 9 (Hudi 1.0 and later), plus a narrow append path on 6
Queries	Snapshot, time travel, incremental, read-optimized (MOR), streaming snapshot reads, column filters with pruning
Writes	Create, append, upsert, update, delete (by predicate or key), overwrite, dynamic partition overwrite
Table services	Merge-on-read compaction (experimental: on demand, inline, or in the background), rollback of failed writes, timeline archival
Metadata table	File listings, column and partition stats, record-level index, kept up to date on every write and compacted as it grows (on by default)
Base file formats	Parquet (read and write), Lance (read), HFile (metadata table)
Storage	Local file system, Amazon S3, Azure Blob Storage, Google Cloud Storage
APIs	Rust and Python (read and write); C++ and a C ABI (hudi-ffi) for file-slice reads
Query engines	Apache DataFusion (datafusion feature in Rust, included in Python); see Other Integrations

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

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

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

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]; completion times on table version 8 and later, requested times before.
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 per DataFusion partition; a Table read defaults to the available cores.
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

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

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

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

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

Path	Contents
crates/hudi	The hudi crate: re-exports hudi-core, and hudi-datafusion with the datafusion feature
crates/core	hudi-core: tables, timeline, file groups, metadata table, readers, writers
crates/datafusion	hudi-datafusion: the DataFusion TableProvider
crates/ffi	hudi-ffi: a C ABI for engines, over the Arrow C Data Interface
python	The hudi Python package (PyO3)
cpp	The C++ binding (cxx)
crates/test	Shared test fixtures
benchmark	Benchmark harnesses
docs	Reader spec and writer design

Contributing

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