Crate hudi 

A native Rust implementation of Apache Hudi.

This crate creates, writes, and reads Hudi tables on local disk or cloud object storage (s3://, az://, gs://), exchanging data as Arrow RecordBatches. It re-exports hudi_core, and with the datafusion feature also the DataFusion table provider from hudi-datafusion. table::Table is the entry point for both reads and writes.

Resources

• Quick start: the Python and Rust quick start guide on the Hudi website, and the shorter README quick start.
• Rust API reference: this crate, hudi-core, and hudi-datafusion.
• Python API reference: the docstrings in the package’s type stubs, python/hudi/_internal.pyi.
• Runnable Rust example: crates/core/examples/table_writes.rs.
• Reader semantics: docs/reader-spec.md. Writer design: docs/writer-design.md.
• Packages: crates.io and PyPI. Source: github.com/apache/hudi-rs.
• Hudi concepts (timeline, table types, query types): hudi.apache.org/docs/overview.

Features

• datafusion (off by default): adds HudiDataSource, a DataFusion TableProvider, and HudiTableFactory for CREATE EXTERNAL TABLE ... STORED AS HUDI.
• spill-rocksdb (on by default): the on-disk tier of the merge-on-read merge map, used when a merge exceeds hoodie.memory.merge.max.size. RocksDB is built from source with bindgen, so it needs libclang and a C++ toolchain. With default-features = false the tier is gone, and a merge past the budget fails with CoreError::Unsupported naming the config instead of spilling.

Create a table and write to it

A write that changes data is one Hudi commit. Table::append inserts without an index lookup; Table::upsert updates rows whose record key exists and inserts the rest; Table::delete and Table::delete_keys remove rows by predicate or by key.

use hudi::config::table::TableTypeValue;
use hudi::table::{HoodieKey, Table};

let mut table = Table::create(base_uri)
    .with_table_name("trips")
    .with_table_type(TableTypeValue::CopyOnWrite) // or MergeOnRead
    .with_record_key_fields(["id"])
    .with_partition_fields(["city"])
    .with_ordering_fields(["ts"]) // the larger `ts` wins when keys collide
    .create()
    .await?;

// `trips` builds a batch with columns `id`, `city`, and `ts`.
table.append([trips(&[("a", "sf", 1), ("b", "sf", 1), ("c", "nyc", 1)])]).await?;

let result = table.upsert([trips(&[("b", "sf", 2), ("d", "nyc", 2)])]).await?;
assert_eq!((result.num_updates, result.num_inserts), (1, 1));

let result = table.delete("id = 'a'").await?;
assert_eq!(result.num_deletes, 1);

let result = table
    .delete_keys([HoodieKey {
        record_key: "c".to_string(),
        partition_path: "city=nyc".to_string(),
    }])
    .await?;
assert_eq!(result.num_deletes, 1);

Compact a merge-on-read table

On a merge-on-read table, upserts, updates, and deletes write log files. Table::compact (experimental) folds them into new base files under one compaction commit. Rows keep their commit time, so incremental reads see no new rows from it. Compaction can also run after every hoodie.compact.inline.max.delta.commits (default 5) delta commits: in the write call with hoodie.compact.inline=true, or on a background task with hoodie.datasource.compaction.async.enable=true, which Table::await_compaction waits for.

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

let mut table = Table::create(base_uri)
    .with_table_name("trips")
    .with_table_type(TableTypeValue::MergeOnRead)
    .with_record_key_fields(["id"])
    .with_partition_fields(["city"])
    .with_ordering_fields(["ts"])
    .create()
    .await?;
table.append([trips(&[("a", "sf", 1), ("b", "nyc", 1)])]).await?;
table.upsert([trips(&[("a", "sf", 2)])]).await?; // writes a log file in `city=sf`

let result = table.compact().await?;
assert_eq!((result.num_file_groups, result.num_log_files), (1, 1));

// Nothing left to compact: no commit is created.
assert!(table.compact().await?.requested_time.is_none());

Snapshot read

Table::new opens an existing table. A read returns the latest committed state, narrowed by optional filters and a projection from ReadOptions.

use hudi::table::{ReadOptions, Table};
let table = Table::new(base_uri).await?;
let options = ReadOptions::new()
    .with_filters([("city", "=", "sf")])?
    .with_projection(["id", "ts"]);
let batches = table.read(&options).await?;
let num_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
assert_eq!(num_rows, 2);

Table::read_stream yields the same snapshot one batch at a time.

Time-travel read

ReadOptions::with_as_of_timestamp reads the table as of an earlier commit. It accepts a Hudi instant time (yyyyMMddHHmmssSSS or yyyyMMddHHmmss), Unix epoch time, or RFC 3339 with a timezone offset.

use hudi::table::{ReadOptions, Table};
let first = table.append([trips(&[("a", "sf", 1), ("b", "sf", 1)])]).await?;
table.append([trips(&[("c", "nyc", 2)])]).await?;

// A write's requested time names its commit for time travel.
let as_of = first.requested_time.expect("the append made a commit");
let options = ReadOptions::new().with_as_of_timestamp(as_of);
let batches = table.read(&options).await?;
let num_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
assert_eq!(num_rows, 2);

Incremental read

An incremental query returns the records changed by commits in the half-open window (start, end]; end defaults to everything committed so far. On tables at version 8 and above (timeline layout 2) the window bounds commit completion times, so start from a write’s completion_time, not its requested_time.

use hudi::table::{QueryType, ReadOptions, Table};
let first = table.append([trips(&[("a", "sf", 1), ("b", "sf", 1)])]).await?;
table.upsert([trips(&[("b", "sf", 2), ("c", "nyc", 2)])]).await?;

let start = first.completion_time.ok_or("layout 2 records a completion time")?;
let options = ReadOptions::new()
    .with_query_type(QueryType::Incremental)
    .with_start_timestamp(start);
let changed = table.read(&options).await?;
let num_rows: usize = changed.iter().map(|b| b.num_rows()).sum();
assert_eq!(num_rows, 2); // `b` updated, `c` inserted

Query with DataFusion

With the datafusion feature, register a table with a DataFusion SessionContext and query it with SQL.

use datafusion::prelude::SessionContext;
use hudi::HudiDataSource;

let ctx = SessionContext::new();
let provider =
    HudiDataSource::new_with_options(base_uri, [("hoodie.read.input.partitions", "2")]).await?;
ctx.register_table("trips", Arc::new(provider))?;
let df = ctx.sql("SELECT id, ts FROM trips WHERE city = 'sf'").await?;
let batches = df.collect().await?;
let num_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
assert_eq!(num_rows, 2);

Modules

config

Hudi configurations.

error

The error and result types returned by hudi-core APIs.

expr

Column filter expressions for reads and writes.

file_group

This module is for File Group related models and APIs.

hfile

HFile reader implementation.

index

Record-location indexes used by write operations.

keygen

Key generator implementations: record key encoding, and transforms of user filters to partition filters.

merge

metadata

schema

statistics

Statistics module for column statistics pruning.

storage

This module is responsible for interacting with the storage layer.

table

The Table API for reading and writing a Hudi table.

timeline

util

write

Write APIs for Hudi tables.

Structs

HoodieKey

A record key and its partition path.

HudiDataSource

A DataFusion table provider for Apache Hudi tables.

HudiTableFactory

HudiTableFactory is responsible for creating and configuring Hudi tables.

SimpleIndex

A single-node index that scans latest file slices to locate record keys.

UpsertOptions

Options controlling an upsert.

WriteResult

Result of a write on Table, shared by every write verb.