This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hudi-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 85810b05 docs(rust): document the user-facing API and build docs.rs
with datafusion (#938)
85810b05 is described below
commit 85810b05111471e1706978b8564bb2c2733593c1
Author: Y Ethan Guo <[email protected]>
AuthorDate: Fri Oct 30 13:49:25 2026 -0700
docs(rust): document the user-facing API and build docs.rs with datafusion
(#938)
---
crates/core/src/config/error.rs | 8 +
crates/core/src/config/internal.rs | 1 +
crates/core/src/config/mod.rs | 20 +-
crates/core/src/config/plan.rs | 1 +
crates/core/src/config/read.rs | 2 +-
crates/core/src/config/read_options.rs | 8 +-
crates/core/src/config/table.rs | 8 +
crates/core/src/error.rs | 29 +++
crates/core/src/expr/filter.rs | 63 +++++-
crates/core/src/expr/mod.rs | 18 +-
crates/core/src/file_group/base_file/mod.rs | 4 +-
crates/core/src/file_group/file_slice.rs | 19 +-
crates/core/src/file_group/log_file/reader.rs | 4 +-
crates/core/src/file_group/mod.rs | 2 +
crates/core/src/file_group/reader.rs | 28 ++-
crates/core/src/hfile/reader.rs | 4 +-
crates/core/src/index/mod.rs | 7 +
crates/core/src/lib.rs | 78 +++++--
crates/core/src/metadata/table/column_stats.rs | 2 +-
crates/core/src/metadata/table/hash.rs | 2 +-
crates/core/src/metadata/table/mod.rs | 2 +-
crates/core/src/schema/avsc.rs | 2 +-
crates/core/src/schema/batch_evolution.rs | 2 +-
crates/core/src/schema/extended_promotion.rs | 4 +-
crates/core/src/schema/resolver.rs | 2 +-
crates/core/src/storage/file_metadata.rs | 4 +-
crates/core/src/table/builder.rs | 23 +-
crates/core/src/table/mod.rs | 163 +++++++------
crates/core/src/table/partition.rs | 7 +
crates/core/src/timeline/instant.rs | 2 +-
crates/core/src/timeline/loader.rs | 2 +-
crates/core/src/timeline/mod.rs | 8 +-
crates/core/src/timeline/selector.rs | 2 +-
crates/core/src/write/create.rs | 24 ++
crates/core/src/write/lock.rs | 1 +
crates/core/src/write/mod.rs | 11 +-
crates/datafusion/src/lib.rs | 35 ++-
crates/hudi/Cargo.toml | 8 +-
crates/hudi/src/lib.rs | 302 +++++++++++++++++++++++++
39 files changed, 761 insertions(+), 151 deletions(-)
diff --git a/crates/core/src/config/error.rs b/crates/core/src/config/error.rs
index 6d852f6c..186e0d31 100644
--- a/crates/core/src/config/error.rs
+++ b/crates/core/src/config/error.rs
@@ -16,12 +16,16 @@
* specific language governing permissions and limitations
* under the License.
*/
+//! The error type of config parsing and validation.
+
use std::num::ParseIntError;
use std::str::ParseBoolError;
use thiserror::Error;
+/// Result type of config parsing, failing with [`ConfigError`] by default.
pub type Result<T, E = ConfigError> = std::result::Result<T, E>;
+/// Errors from parsing or validating a config.
#[derive(Error, Debug)]
pub enum ConfigError {
/// ParseBool(key, value, source)
@@ -36,12 +40,16 @@ pub enum ConfigError {
#[error("Failed to parse line: {0}'")]
ParseLine(String),
+ /// InvalidValue(value or message): the value is not one the config
accepts. The payload is
+ /// the rejected value or a message describing it.
#[error("Config value '{0}' is not valid")]
InvalidValue(String),
+ /// NotFound(key): the config is not set.
#[error("Config '{0}' not found")]
NotFound(String),
+ /// UnsupportedValue(message): the value is valid but not supported by
hudi-rs.
#[error("Config value '{0}' is not supported")]
UnsupportedValue(String),
}
diff --git a/crates/core/src/config/internal.rs
b/crates/core/src/config/internal.rs
index 8d8e8184..51cfbe1d 100644
--- a/crates/core/src/config/internal.rs
+++ b/crates/core/src/config/internal.rs
@@ -45,6 +45,7 @@ use crate::config::{ConfigParser, HudiConfigValue};
///
#[derive(Clone, Debug, PartialEq, Eq, Hash, EnumIter)]
pub enum HudiInternalConfig {
+ /// Skip validating table configs when a table is opened. Defaults to
`false`.
SkipConfigValidation,
/// Enable reading archived timeline (v1) and LSM history (v2).
///
diff --git a/crates/core/src/config/mod.rs b/crates/core/src/config/mod.rs
index 9f270a76..886d36d5 100644
--- a/crates/core/src/config/mod.rs
+++ b/crates/core/src/config/mod.rs
@@ -16,7 +16,13 @@
* specific language governing permissions and limitations
* under the License.
*/
-//! Hudi Configurations.
+//! Hudi configurations.
+//!
+//! Each config is an enum variant that knows its `hoodie.*` key, default, and
how to parse its
+//! value: [`table::HudiTableConfig`] (mostly persisted in
`hoodie.properties`),
+//! [`read::HudiReadConfig`], [`plan::HudiPlanConfig`], and
[`internal::HudiInternalConfig`].
+//! [`HudiConfigs`] holds the resolved key-value options of a table, and
+//! [`read_options::ReadOptions`] those of a single read.
use std::any::type_name;
use std::collections::HashMap;
use std::sync::Arc;
@@ -39,15 +45,20 @@ pub mod util;
pub use read_options::{QueryType, ReadOptions};
+/// Environment variable naming the directory that holds `hudi-defaults.conf`,
whose options a
+/// table picks up when they are not set otherwise. Defaults to
`/etc/hudi/conf` when unset.
pub const HUDI_CONF_DIR: &str = "HUDI_CONF_DIR";
/// An alternative key for a configuration, optionally marked as deprecated.
pub struct ConfigAlias {
+ /// The alternative key.
pub key: &'static str,
+ /// Whether using this key logs a deprecation warning.
pub deprecated: bool,
}
impl ConfigAlias {
+ /// An alias that is not deprecated.
pub const fn new(key: &'static str) -> Self {
Self {
key,
@@ -55,6 +66,7 @@ impl ConfigAlias {
}
}
+ /// A deprecated alias.
pub const fn deprecated(key: &'static str) -> Self {
Self {
key,
@@ -71,6 +83,7 @@ pub trait ConfigParser: AsRef<str> {
/// Supplies the default value of the configuration.
fn default_value(&self) -> Option<Self::Output>;
+ /// The config key, such as `hoodie.table.name`.
fn key(&self) -> String {
self.as_ref().to_string()
}
@@ -136,10 +149,15 @@ pub trait ConfigParser: AsRef<str> {
/// All possible data types for Hudi Configuration values.
#[derive(Clone, Debug)]
pub enum HudiConfigValue {
+ /// A boolean value.
Boolean(bool),
+ /// A signed integer value.
Integer(isize),
+ /// An unsigned integer value.
UInteger(usize),
+ /// A string value.
String(String),
+ /// A comma-separated list, split into its items.
List(Vec<String>),
}
diff --git a/crates/core/src/config/plan.rs b/crates/core/src/config/plan.rs
index b2664eb9..d9746b88 100644
--- a/crates/core/src/config/plan.rs
+++ b/crates/core/src/config/plan.rs
@@ -54,6 +54,7 @@ pub enum HudiPlanConfig {
}
impl HudiPlanConfig {
+ /// The config key, usable in `const` contexts.
pub const fn key_str(&self) -> &'static str {
match self {
Self::ListingParallelism => "hoodie.plan.listing.parallelism",
diff --git a/crates/core/src/config/read.rs b/crates/core/src/config/read.rs
index bec51459..593aaed3 100644
--- a/crates/core/src/config/read.rs
+++ b/crates/core/src/config/read.rs
@@ -113,7 +113,7 @@ pub enum HudiReadConfig {
/// Number of input partitions to read the data in parallel.
///
- /// For processing 100 files, [InputPartitions] being 5 will produce 5
partitions, with each partition having 20 files.
+ /// For processing 100 files, [Self::InputPartitions] being 5 will produce
5 partitions, with each partition having 20 files.
///
/// On the parquet-source path (copy-on-write, or read-optimized
merge-on-read)
/// DataFusion may split these buckets further once one exceeds
diff --git a/crates/core/src/config/read_options.rs
b/crates/core/src/config/read_options.rs
index 2cec1214..6860c449 100644
--- a/crates/core/src/config/read_options.rs
+++ b/crates/core/src/config/read_options.rs
@@ -275,10 +275,6 @@ impl ReadOptions {
}
}
- /// The target batch size (rows per batch) for streaming reads, if set.
- /// Errors if the stored string is not a valid `usize` or if the value is
`0`
- /// (a zero-row batch yields no batches at the parquet stream reader and is
- /// almost certainly a caller mistake).
/// Fill in defaults from the given configs for keys not already set.
///
/// Values already present in this `ReadOptions` take precedence.
@@ -296,6 +292,10 @@ impl ReadOptions {
Ok(resolved)
}
+ /// The target batch size (rows per batch) for streaming reads, if set.
+ /// Errors if the stored string is not a valid `usize` or if the value is
`0`
+ /// (a zero-row batch yields no batches at the parquet stream reader and is
+ /// almost certainly a caller mistake).
pub fn batch_size(&self) -> crate::Result<Option<usize>> {
let key = HudiReadConfig::StreamBatchSize.as_ref();
match self.hudi_options.get(key) {
diff --git a/crates/core/src/config/table.rs b/crates/core/src/config/table.rs
index 5a69eab8..c67c19ad 100644
--- a/crates/core/src/config/table.rs
+++ b/crates/core/src/config/table.rs
@@ -359,10 +359,14 @@ impl ConfigParser for HudiTableConfig {
}
/// Config value for [HudiTableConfig::TableType].
+///
+/// Parsing is case-insensitive and also accepts `cow` / `mor` and hyphenated
forms.
#[derive(Clone, Debug, PartialEq, AsRefStr)]
pub enum TableTypeValue {
+ /// `COPY_ON_WRITE`: updates rewrite base files.
#[strum(serialize = "COPY_ON_WRITE")]
CopyOnWrite,
+ /// `MERGE_ON_READ`: updates go to log files, merged with base files when
read.
#[strum(serialize = "MERGE_ON_READ")]
MergeOnRead,
}
@@ -382,11 +386,13 @@ impl FromStr for TableTypeValue {
/// Config value for [HudiTableConfig::BaseFileFormat].
#[derive(Clone, Debug, PartialEq, AsRefStr)]
pub enum BaseFileFormatValue {
+ /// Config value `parquet`, the default.
#[strum(serialize = "parquet")]
Parquet,
/// HFile format - only valid for metadata tables.
#[strum(serialize = "hfile")]
HFile,
+ /// Config value `lance`.
#[strum(serialize = "lance")]
Lance,
}
@@ -471,9 +477,11 @@ impl FromStr for BaseFileFormatValue {
/// Config value for [HudiTableConfig::TimelineTimezone].
#[derive(Clone, Debug, PartialEq, AsRefStr, Default)]
pub enum TimelineTimezoneValue {
+ /// Instant times are in UTC, assumed when the config is unset.
#[strum(serialize = "utc")]
#[default]
UTC,
+ /// Instant times are in the local timezone of the process.
#[strum(serialize = "local")]
Local,
}
diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs
index c49e7f3e..ffd39158 100644
--- a/crates/core/src/error.rs
+++ b/crates/core/src/error.rs
@@ -16,26 +16,35 @@
* specific language governing permissions and limitations
* under the License.
*/
+//! The error and result types returned by hudi-core APIs.
+
use crate::config::error::ConfigError;
use crate::storage::error::StorageError;
use thiserror::Error;
+/// Result type of hudi-core APIs, failing with [`CoreError`] by default.
pub type Result<T, E = CoreError> = std::result::Result<T, E>;
+/// Errors returned by hudi-core APIs.
#[derive(Error, Debug)]
pub enum CoreError {
+ /// An Arrow operation failed (schema mismatch, cast, or batch
construction).
#[error(transparent)]
ArrowError(#[from] arrow::error::ArrowError),
+ /// Avro encoding or decoding failed.
#[error(transparent)]
AvroError(#[from] apache_avro::Error),
+ /// A config is missing or has an invalid value.
#[error("Config error: {0}")]
Config(#[from] ConfigError),
+ /// Commit metadata on the timeline could not be parsed or lacks a
required field.
#[error("Commit metadata error: {0}")]
CommitMetadata(String),
+ /// Records could not be merged.
#[error("{0}")]
MergeRecordError(String),
@@ -43,60 +52,80 @@ pub enum CoreError {
#[error("{0}")]
MemoryLimitExceeded(String),
+ /// A schema or data type is invalid for the operation, such as a filter
on an unknown
+ /// column or a write whose columns do not match the table.
#[error("Data type error: {0}")]
Schema(String),
+ /// The table schema could not be resolved.
#[error("{0}")]
SchemaNotFound(String),
+ /// A file group, file slice, or data file name is invalid.
#[error("File group error: {0}")]
FileGroup(String),
+ /// Reading or merging a file slice failed.
#[error("{0}")]
ReadFileSliceError(String),
+ /// A log file does not follow the Hudi log format.
#[error("{0}")]
LogFormatError(String),
+ /// A log block could not be decoded.
#[error("{0}")]
LogBlockError(String),
+ /// Reading or writing an HFile failed.
#[error("HFile error: {0}")]
HFile(String),
+ /// Reading or writing the metadata table failed.
#[error("Metadata table error: {0}")]
MetadataTable(String),
+ /// A partition path does not match the table's partition fields.
#[error("{0}")]
InvalidPartitionPath(String),
+ /// An argument or stored value is invalid.
#[error("{0}")]
InvalidValue(String),
+ /// Reading or writing a Parquet file failed.
#[error(transparent)]
ParquetError(#[from] parquet::errors::ParquetError),
+ /// An I/O error, raised mostly while reading log files.
#[error("{0}")]
ReadLogFileError(#[from] std::io::Error),
+ /// An object store operation failed.
#[error("Storage error: {0}")]
Storage(#[from] StorageError),
+ /// The timeline could not be loaded or holds an invalid instant.
#[error("Timeline error: {0}")]
Timeline(String),
+ /// The timeline has no completed commit.
#[error("Timeline has no completed commit.")]
TimelineNoCommit,
+ /// A timestamp string could not be parsed.
#[error("{0}")]
TimestampParsingError(String),
+ /// The table, config, or operation is not supported.
#[error("{0}")]
Unsupported(String),
+ /// A write was rejected or failed.
#[error("Write error: {0}")]
Write(String),
+ /// Bytes expected to be UTF-8 were not.
#[error(transparent)]
Utf8Error(#[from] std::str::Utf8Error),
}
diff --git a/crates/core/src/expr/filter.rs b/crates/core/src/expr/filter.rs
index afbf3cd7..a2cdf903 100644
--- a/crates/core/src/expr/filter.rs
+++ b/crates/core/src/expr/filter.rs
@@ -17,6 +17,8 @@
* under the License.
*/
+//! The [`Filter`] type and helpers to build and evaluate filters.
+
use crate::Result;
use crate::error::CoreError;
use crate::expr::ExprOperator;
@@ -30,14 +32,28 @@ use arrow_schema::{DataType, Field, Schema};
use std::collections::HashSet;
use std::str::FromStr;
+/// A comparison of one column against one or more values, such as `city =
'sf'`.
+///
+/// Values are kept as strings and cast to the column's Arrow type when the
filter is evaluated.
+/// Build one with [`Filter::new`], from a `(field, operator, value)` tuple
with
+/// `Filter::try_from`, or with [`col`].
#[derive(Debug, Clone)]
pub struct Filter {
+ /// Column name.
pub field: String,
+ /// Comparison operator.
pub operator: ExprOperator,
+ /// Values to compare against: exactly one, or at least one for `IN` /
`NOT IN`.
pub values: Vec<String>,
}
impl Filter {
+ /// Creates a filter, checking the number of values against the operator.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`CoreError::Schema`] when `IN` / `NOT IN` gets no values, or
another
+ /// operator does not get exactly one.
pub fn new(field: String, operator: ExprOperator, values: Vec<String>) ->
Result<Self> {
if operator.is_multi_value() {
if values.is_empty() {
@@ -60,6 +76,7 @@ impl Filter {
}
impl Filter {
+ /// The filter with its operator negated (see [`ExprOperator::negate`]).
pub fn negate(&self) -> Option<Self> {
self.operator.negate().map(|op| Self {
operator: op,
@@ -156,6 +173,10 @@ fn escape_in_value(v: &str) -> String {
out
}
+/// Parses `(field, operator, value)` tuples into filters, as
`Filter::try_from` does.
+///
+/// For `IN` / `NOT IN` the value is a comma-separated list; `\,` is a literal
comma and `\\`
+/// a literal backslash.
pub fn from_str_tuples<I, S>(tuples: I) -> Result<Vec<Filter>>
where
I: IntoIterator<Item = (S, S, S)>,
@@ -167,7 +188,7 @@ where
.collect()
}
-/// Evaluate a slice of [`Filter`]s against a [`RecordBatch`] and return a
row-level mask
+/// Evaluate a slice of [`Filter`]s against a
[`RecordBatch`](arrow_array::RecordBatch) and return a row-level mask
/// where `true` indicates the row should be retained.
///
/// This is a low-level primitive: filters whose field is not present in the
batch are
@@ -224,19 +245,29 @@ where
Ok(())
}
+/// A column to build [`Filter`]s on; see [`col`].
+///
+/// The builders do not check the number of values, so an empty
+/// [`in_list`](Self::in_list) is not rejected here the way [`Filter::new`]
rejects it. Such a
+/// filter panics when evaluated, for example by [`filters_to_row_mask`]
during a read that has
+/// it in [`ReadOptions::filters`](crate::table::ReadOptions::filters).
pub struct FilterField {
+ /// Column name.
pub name: String,
}
impl FilterField {
+ /// Creates a field for the column `name`.
pub fn new(name: impl Into<String>) -> Self {
Self { name: name.into() }
}
+ /// The column name.
pub fn name(&self) -> &str {
&self.name
}
+ /// `field = value`
pub fn eq(&self, value: impl Into<String>) -> Filter {
Filter {
field: self.name.clone(),
@@ -245,6 +276,7 @@ impl FilterField {
}
}
+ /// `field != value`
pub fn ne(&self, value: impl Into<String>) -> Filter {
Filter {
field: self.name.clone(),
@@ -253,6 +285,7 @@ impl FilterField {
}
}
+ /// `field < value`
pub fn lt(&self, value: impl Into<String>) -> Filter {
Filter {
field: self.name.clone(),
@@ -261,6 +294,7 @@ impl FilterField {
}
}
+ /// `field <= value`
pub fn lte(&self, value: impl Into<String>) -> Filter {
Filter {
field: self.name.clone(),
@@ -269,6 +303,7 @@ impl FilterField {
}
}
+ /// `field > value`
pub fn gt(&self, value: impl Into<String>) -> Filter {
Filter {
field: self.name.clone(),
@@ -277,6 +312,7 @@ impl FilterField {
}
}
+ /// `field >= value`
pub fn gte(&self, value: impl Into<String>) -> Filter {
Filter {
field: self.name.clone(),
@@ -285,6 +321,7 @@ impl FilterField {
}
}
+ /// `field IN (values)`
pub fn in_list<I, S>(&self, values: I) -> Filter
where
I: IntoIterator<Item = S>,
@@ -297,6 +334,7 @@ impl FilterField {
}
}
+ /// `field NOT IN (values)`
pub fn not_in_list<I, S>(&self, values: I) -> Filter
where
I: IntoIterator<Item = S>,
@@ -310,14 +348,29 @@ impl FilterField {
}
}
+/// Starts a [`Filter`] on the column `name`.
+///
+/// ```
+/// use hudi_core::expr::filter::col;
+///
+/// let filter = col("city").in_list(["sf", "nyc"]);
+/// assert_eq!(filter.values, vec!["sf", "nyc"]);
+/// ```
pub fn col(name: impl Into<String>) -> FilterField {
FilterField::new(name)
}
+/// A [`Filter`] resolved against a schema, with its values cast to the
column's type.
+///
+/// Created with `SchemableFilter::try_from((filter, &schema))`, which fails
with
+/// [`CoreError::Schema`] when the column is missing or a value cannot be cast.
#[derive(Debug, Clone)]
pub struct SchemableFilter {
+ /// The schema field the filter applies to.
pub field: Field,
+ /// Comparison operator.
pub operator: ExprOperator,
+ /// Values cast to the field's data type.
pub values: Vec<Scalar<ArrayRef>>,
}
@@ -345,6 +398,8 @@ impl TryFrom<(Filter, &Schema)> for SchemableFilter {
}
impl SchemableFilter {
+ /// Casts a string value to `data_type`, failing with
[`CoreError::Schema`] rather than
+ /// producing a null when the value does not parse.
pub fn cast_value(value: &[&str; 1], data_type: &DataType) ->
Result<Scalar<ArrayRef>> {
let cast_options = CastOptions {
safe: false,
@@ -359,6 +414,12 @@ impl SchemableFilter {
))
}
+ /// Evaluates the filter against `value` (typically a column), returning
`true` for rows
+ /// that match.
+ ///
+ /// # Panics
+ ///
+ /// Panics if [`values`](Self::values) is empty.
pub fn apply_comparison(&self, value: &dyn Datum) -> Result<BooleanArray> {
match self.operator {
ExprOperator::Eq => eq(value, &self.values[0]),
diff --git a/crates/core/src/expr/mod.rs b/crates/core/src/expr/mod.rs
index cc7c84ca..6b596201 100644
--- a/crates/core/src/expr/mod.rs
+++ b/crates/core/src/expr/mod.rs
@@ -17,6 +17,11 @@
* under the License.
*/
+//! Column filter expressions for reads and writes.
+//!
+//! A [`filter::Filter`] compares one column against string values with an
[`ExprOperator`].
+//! Values are cast to the column's Arrow type when the filter is evaluated.
+
pub mod filter;
use crate::error::CoreError;
@@ -26,16 +31,24 @@ use std::cmp::PartialEq;
use std::fmt::{Display, Formatter, Result as FmtResult};
use std::str::FromStr;
-/// An operator that represents a comparison operation used in a partition
filter expression.
+/// A comparison operator in a [`filter::Filter`].
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum ExprOperator {
+ /// `=`
Eq,
+ /// `!=`
Ne,
+ /// `<`
Lt,
+ /// `<=`
Lte,
+ /// `>`
Gt,
+ /// `>=`
Gte,
+ /// `IN`: equal to any of the values.
In,
+ /// `NOT IN`: equal to none of the values.
NotIn,
}
@@ -56,6 +69,7 @@ impl Display for ExprOperator {
}
impl ExprOperator {
+ /// The token for each operator, as accepted by [`FromStr`]
(case-insensitive).
pub const TOKEN_OP_PAIRS: [(&'static str, ExprOperator); 8] = [
("=", ExprOperator::Eq),
("!=", ExprOperator::Ne),
@@ -72,7 +86,7 @@ impl ExprOperator {
matches!(self, ExprOperator::In | ExprOperator::NotIn)
}
- /// Negates the operator.
+ /// The logical negation, such as `<` for `>=`. Always `Some` for the
current operators.
pub fn negate(&self) -> Option<ExprOperator> {
match self {
ExprOperator::Eq => Some(ExprOperator::Ne),
diff --git a/crates/core/src/file_group/base_file/mod.rs
b/crates/core/src/file_group/base_file/mod.rs
index e72619d3..f08b710a 100644
--- a/crates/core/src/file_group/base_file/mod.rs
+++ b/crates/core/src/file_group/base_file/mod.rs
@@ -30,10 +30,10 @@ use crate::timeline::completion_time::CompletionTimeView;
use std::fmt::Display;
use std::str::FromStr;
-/// Hudi Base file, part of a [FileSlice].
+/// Hudi Base file, part of a
[FileSlice](crate::file_group::file_slice::FileSlice).
#[derive(Clone, Debug)]
pub struct BaseFile {
- /// The id of the enclosing [FileGroup].
+ /// The id of the enclosing [FileGroup](crate::file_group::FileGroup).
pub file_id: String,
/// Monotonically increasing token for every attempt to write the
[BaseFile].
diff --git a/crates/core/src/file_group/file_slice.rs
b/crates/core/src/file_group/file_slice.rs
index 5e8088d5..36fc8b1f 100644
--- a/crates/core/src/file_group/file_slice.rs
+++ b/crates/core/src/file_group/file_slice.rs
@@ -16,6 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
+//! [`FileSlice`], the unit a file group reader reads.
+
use crate::Result;
use crate::error::CoreError;
use crate::file_group::base_file::BaseFile;
@@ -43,7 +45,9 @@ pub struct FileSlice {
/// The instant that created this slice: its base file's commit, or — with
/// no base file — the earliest log file's own instant.
creation_instant_time: String,
+ /// The log files that apply on top of the base file, in [`LogFile`] order.
pub log_files: BTreeSet<LogFile>,
+ /// The partition path relative to the table base, empty for an
unpartitioned table.
pub partition_path: String,
/// Column statistics from the base file's Parquet footer.
///
@@ -74,6 +78,8 @@ impl PartialEq for FileSlice {
impl Eq for FileSlice {}
impl FileSlice {
+ /// A slice with `base_file` and no log files, taking its file id and
creation instant from
+ /// the base file.
pub fn new(base_file: BaseFile, partition_path: String) -> Self {
Self {
file_id: base_file.file_id.clone(),
@@ -101,11 +107,18 @@ impl FileSlice {
}
}
+ /// Whether the slice has any log files.
#[inline]
pub fn has_log_file(&self) -> bool {
!self.log_files.is_empty()
}
+ /// Adds the log files of `other`, which must be the same slice.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`CoreError::FileGroup`] when `other` differs in base file,
partition path, file
+ /// id, or creation instant.
pub fn merge(&mut self, other: &FileSlice) -> Result<()> {
if self != other {
return Err(CoreError::FileGroup(format!(
@@ -139,7 +152,7 @@ impl FileSlice {
self.relative_path_for_file(file_name)
}
- /// Returns the enclosing [FileGroup]'s id.
+ /// Returns the enclosing [FileGroup](crate::file_group::FileGroup)'s id.
#[inline]
pub fn file_id(&self) -> &str {
&self.file_id
@@ -147,7 +160,7 @@ impl FileSlice {
/// Returns the instant time that marks the [FileSlice] creation.
///
- /// This is also an instant time stored in the [Timeline].
+ /// This is also an instant time stored in the
[Timeline](crate::timeline::Timeline).
#[inline]
pub fn creation_instant_time(&self) -> &str {
&self.creation_instant_time
@@ -157,7 +170,7 @@ impl FileSlice {
///
/// Use this for I/O cost estimation and split sizing (e.g., balancing
/// parallel read splits by storage volume). For estimated in-memory size
- /// and record count for query planning, use [`Table::compute_table_stats`]
+ /// and record count for query planning, use
[`Table::compute_table_stats`](crate::table::Table::compute_table_stats)
/// instead — it returns values derived from the base file only.
///
/// Files whose `file_metadata` is `None` contribute 0 to the sum.
diff --git a/crates/core/src/file_group/log_file/reader.rs
b/crates/core/src/file_group/log_file/reader.rs
index 0d09ba6c..e4c36d4e 100644
--- a/crates/core/src/file_group/log_file/reader.rs
+++ b/crates/core/src/file_group/log_file/reader.rs
@@ -191,7 +191,7 @@ impl LogFileReader {
/// Push a predicate into this file's parquet log blocks.
///
/// Left unset, no predicate is pushed and every row is read. Whether
pushing
- /// is sound is the caller's decision — see [`Decoder::with_row_filter`].
+ /// is sound is the caller's decision (see `Decoder::with_row_filter`).
pub fn with_row_filter(mut self, row_filter: Option<RowFilterBuilder>) ->
Self {
self.row_filter = row_filter;
self
@@ -207,7 +207,7 @@ impl LogFileReader {
}
/// Resolve Avro blocks up to this schema. See
- /// [`Decoder::with_reader_schema`].
+ /// `Decoder::with_reader_schema`.
pub fn with_reader_schema(mut self, reader_schema_json: Option<String>) ->
Self {
self.reader_schema_json = reader_schema_json;
self
diff --git a/crates/core/src/file_group/mod.rs
b/crates/core/src/file_group/mod.rs
index b2ca4402..388a27af 100644
--- a/crates/core/src/file_group/mod.rs
+++ b/crates/core/src/file_group/mod.rs
@@ -23,9 +23,11 @@
pub mod admission;
pub mod base_file;
pub mod builder;
+#[warn(missing_docs)]
pub mod file_slice;
pub mod log_file;
pub(crate) mod pushdown;
+#[warn(missing_docs)]
pub mod reader;
pub(crate) mod reader_v2;
pub mod record_batches;
diff --git a/crates/core/src/file_group/reader.rs
b/crates/core/src/file_group/reader.rs
index e2996e28..ac6a2cf6 100644
--- a/crates/core/src/file_group/reader.rs
+++ b/crates/core/src/file_group/reader.rs
@@ -16,6 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
+//! [`FileGroupReader`] reads and merges the records of one file slice.
+
use crate::Result;
use crate::config::HudiConfigs;
use crate::config::internal::HudiInternalConfig;
@@ -120,14 +122,6 @@ impl FileGroupReader {
})
}
- /// Creates a new reader with the given base URI and options.
- ///
- /// # Arguments
- /// * `base_uri` - The base URI of the file group's residing table.
- /// * `options` - Additional options for the reader.
- ///
- /// # Notes
- /// This API uses [`OptionResolver`] that loads table properties from
storage to resolve options.
/// Refuse a scan-wide memory budget on a reader that reads one file slice.
///
/// `hoodie.read.scan.max.memory.size` bounds a scan by deciding how many
slices
@@ -155,6 +149,24 @@ impl FileGroupReader {
Ok(())
}
+ /// Creates a new reader with the given base URI and options.
+ ///
+ /// # Arguments
+ /// * `base_uri` - The base URI of the file group's residing table.
+ /// * `options` - Additional options for the reader.
+ ///
+ /// # Notes
+ /// This API uses [`OptionResolver`] that loads table properties from
storage to resolve options.
+ ///
+ /// A reader built this way has no timeline, so it reads with the base
file's schema and does
+ /// not check whether a log block's instant has committed. Prefer
+ ///
[`Table::create_file_group_reader_with_options`](crate::table::Table::create_file_group_reader_with_options)
+ /// when a [`Table`](crate::table::Table) is at hand.
+ ///
+ /// # Errors
+ ///
+ /// Fails when the table properties cannot be loaded, and with
[`CoreError::InvalidValue`] when
+ /// `hoodie.read.scan.max.memory.size` is set, since this reader reads one
slice per call.
pub async fn new_with_options<I, K, V>(base_uri: &str, options: I) ->
Result<Self>
where
I: IntoIterator<Item = (K, V)>,
diff --git a/crates/core/src/hfile/reader.rs b/crates/core/src/hfile/reader.rs
index 2cdc412b..219365e9 100644
--- a/crates/core/src/hfile/reader.rs
+++ b/crates/core/src/hfile/reader.rs
@@ -294,7 +294,7 @@ impl HFileReader {
/// Open an HFile without holding it.
///
/// Two ranged reads before any data is touched: the trailer, which is the
- /// final [`TRAILER_SIZE`] bytes, then the load-on-open section it points
at,
+ /// final `TRAILER_SIZE` bytes, then the load-on-open section it points at,
/// which carries the index and the file info. Data blocks are read later,
/// through [`Self::read_records_batched`].
pub async fn open_ranged(storage: &Storage, relative_path: &str) ->
Result<Self> {
@@ -1480,7 +1480,7 @@ impl HFileReader {
/// Keys must be sorted in ascending order. This method efficiently
/// scans forward through the file to find matching keys.
///
- /// Returns a vector of (key, Option<HFileRecord>) tuples where
+ /// Returns a vector of `(key, Option<HFileRecord>)` tuples where
/// the Option is Some if the key was found.
pub fn lookup_records(&mut self, keys: &[&str]) -> Result<Vec<(String,
Option<HFileRecord>)>> {
let mut results = Vec::with_capacity(keys.len());
diff --git a/crates/core/src/index/mod.rs b/crates/core/src/index/mod.rs
index 806e8321..e9534e5a 100644
--- a/crates/core/src/index/mod.rs
+++ b/crates/core/src/index/mod.rs
@@ -33,7 +33,14 @@ use crate::table::Table;
/// A record key and its partition path.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct HoodieKey {
+ /// The record key as stored, such as `a:1,b:x` for a composite key.
pub record_key: String,
+ /// The partition path relative to the table base, such as `city=sf`;
empty for an
+ /// unpartitioned table.
+ ///
+ /// [`Table::delete_keys`](crate::table::Table::delete_keys) locates keys
by record key. On a
+ /// copy-on-write table, an empty `partition_path` there deletes the key
in any partition and a
+ /// non-empty one only in that partition.
pub partition_path: String,
}
diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs
index 72e6b4f5..9055e32d 100644
--- a/crates/core/src/lib.rs
+++ b/crates/core/src/lib.rs
@@ -16,34 +16,69 @@
* specific language governing permissions and limitations
* under the License.
*/
-//! Crate `hudi-core`.
+//! Core library of hudi-rs, a native Rust implementation of [Apache
Hudi](https://hudi.apache.org).
//!
-//! # The [config] module is responsible for managing configurations.
+//! Most applications depend on the [`hudi`](https://docs.rs/hudi) crate,
which re-exports this one;
+//! the [Python and Rust quick start guide][quick start guide] walks through
it, and the `hudi`
+//! crate docs list every resource (API references, quick starts, examples,
specs). The main entry
+//! points:
//!
-//! **Example**
+//! - [`table::Table`] opens a table ([`Table::new`](table::Table::new),
+//! [`TableBuilder`](table::builder::TableBuilder)) or creates one
+//! ([`Table::create`](table::Table::create)), and carries both the read API
(`read`,
+//! `read_stream`, `get_file_slices`) and the write API (`append`, `upsert`,
`delete`, `update`,
+//! `overwrite`, and the experimental `compact` for merge-on-read tables).
+//! - [`table::ReadOptions`] selects the query type ([`table::QueryType`]),
timestamps, filters,
+//! and projection for every read.
+//! - [`write`](mod@write) holds the types the write API takes and returns,
such as [`WriteResult`] and
+//! [`UpsertOptions`].
+//! - [`file_group::reader::FileGroupReader`] reads one file slice at a time,
for query engines
+//! that plan their own scans with
[`Table::get_file_slices`](table::Table::get_file_slices).
+//! - [`config`] defines the typed table and read configs, such as
+//! [`HudiTableConfig`](config::table::HudiTableConfig) and
+//! [`HudiReadConfig`](config::read::HudiReadConfig).
//!
-//! ```rust
-//! use hudi_core::config::read::HudiReadConfig::InputPartitions;
-//! use hudi_core::table::Table as HudiTable;
+//! Most fallible APIs return [`error::Result`], whose error type is
[`error::CoreError`]; config
+//! parsing returns [`config::error::Result`].
//!
-//! # #[tokio::main]
-//! # async fn main() {
-//! let options = [(InputPartitions, "2")];
-//! HudiTable::new_with_options("/tmp/hudi_data", options).await;
-//! # }
-//! ```
+//! [quick start guide]:
https://hudi.apache.org/docs/python-rust-quick-start-guide
//!
-//! # The [table] module is responsible for managing Hudi tables.
+//! # Example
//!
-//! **Example**
+//! Plan a snapshot read as file slices and read each slice with a file group
reader, the way a
+//! query engine distributes a scan.
//!
-//! create hudi table
-//! ```rust
-//! use hudi_core::table::Table;
+//! ```
+//! # use std::sync::Arc;
+//! # use arrow::array::{ArrayRef, RecordBatch, StringArray};
+//! # use arrow::datatypes::{DataType, Field, Schema};
+//! use hudi_core::config::util::empty_options;
+//! use hudi_core::table::{ReadOptions, Table};
//!
-//! pub async fn test() {
-//! let hudi_table = Table::new("/tmp/hudi_data").await.unwrap();
+//! # #[tokio::main]
+//! # async fn main() -> hudi_core::error::Result<()> {
+//! # let dir = tempfile::tempdir()?;
+//! # let base_uri = dir.path().to_str().unwrap();
+//! # let mut writer = Table::create(base_uri)
+//! # .with_table_name("docs")
+//! # .with_record_key_fields(["id"])
+//! # .create()
+//! # .await?;
+//! # let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Utf8,
false)]));
+//! # let ids: ArrayRef = Arc::new(StringArray::from(vec!["a", "b"]));
+//! # writer.append([RecordBatch::try_new(schema, vec![ids])?]).await?;
+//! let table = Table::new(base_uri).await?;
+//! let options = ReadOptions::new();
+//! let reader = table
+//! .create_file_group_reader_with_options(Some(&options), empty_options())
+//! .await?;
+//! let mut num_rows = 0;
+//! for slice in table.get_file_slices(&options).await? {
+//! num_rows += reader.read_file_slice(&slice, &options).await?.num_rows();
//! }
+//! assert_eq!(num_rows, 2);
+//! # Ok(())
+//! # }
//! ```
// The write-path async fns (e.g. `Table::delete_keys`) nest deeply enough
that computing their
@@ -52,8 +87,11 @@
mod avro_to_arrow;
mod bloom;
+#[warn(missing_docs)]
pub mod config;
+#[warn(missing_docs)]
pub mod error;
+#[warn(missing_docs)]
pub mod expr;
pub mod file_group;
pub mod hfile;
@@ -65,9 +103,11 @@ mod record;
pub mod schema;
pub mod statistics;
pub mod storage;
+#[warn(missing_docs)]
pub mod table;
pub mod timeline;
pub mod util;
+#[warn(missing_docs)]
pub mod write;
pub use index::{HoodieKey, SimpleIndex};
diff --git a/crates/core/src/metadata/table/column_stats.rs
b/crates/core/src/metadata/table/column_stats.rs
index 494bb1db..458c950b 100644
--- a/crates/core/src/metadata/table/column_stats.rs
+++ b/crates/core/src/metadata/table/column_stats.rs
@@ -199,7 +199,7 @@ pub fn column_stats_for_file(
/// Aggregate file-level ranges into partition-level stats (same column name).
///
/// Ranges must already be expressed in their column's type (see
-/// [`coerce_stat_value`]); values of different types fail the aggregation
+/// `coerce_stat_value`); values of different types fail the aggregation
/// rather than producing a bound ordered by type instead of by value.
pub fn aggregate_partition_stats(
partition_path: &str,
diff --git a/crates/core/src/metadata/table/hash.rs
b/crates/core/src/metadata/table/hash.rs
index d6590ba2..3acbff25 100644
--- a/crates/core/src/metadata/table/hash.rs
+++ b/crates/core/src/metadata/table/hash.rs
@@ -36,7 +36,7 @@ pub fn partition_identifier(partition_path: &str) -> &str {
}
}
-/// XXHash64 (seed [`HASH_SEED`]) as big-endian bytes, Java `HashID` BITS_64.
+/// XXHash64 (seed `HASH_SEED`) as big-endian bytes, Java `HashID` BITS_64.
pub fn xxhash64_bytes(message: &str) -> [u8; 8] {
xxh64(message.as_bytes(), HASH_SEED).to_be_bytes()
}
diff --git a/crates/core/src/metadata/table/mod.rs
b/crates/core/src/metadata/table/mod.rs
index 7d2ad912..8a563a1a 100644
--- a/crates/core/src/metadata/table/mod.rs
+++ b/crates/core/src/metadata/table/mod.rs
@@ -170,7 +170,7 @@ impl Table {
///
/// Records are returned with normalized partition keys. For
non-partitioned tables,
/// the key is "" (empty string) instead of the internal "."
representation.
- /// Normalization happens at decode time in
[`decode_files_partition_record_with_schema`].
+ /// Normalization happens at decode time in
`decode_files_partition_record_with_schema`.
///
/// # Note
/// Must be called on a DATA table, not a METADATA table.
diff --git a/crates/core/src/schema/avsc.rs b/crates/core/src/schema/avsc.rs
index e695aa35..aa801417 100644
--- a/crates/core/src/schema/avsc.rs
+++ b/crates/core/src/schema/avsc.rs
@@ -63,7 +63,7 @@ fn parse_vendored(avsc: &str, name: &str) -> Result<Schema> {
/// Serialize `value` to Avro OCF using a Java writer schema.
///
-/// Goes JSON → untyped Avro [`Value`] → [`Value::resolve`] so union indexes
and
+/// Goes JSON → untyped Avro `Value` → `Value::resolve` so union indexes and
/// field order match the schema (unlike `to_value`/`append_ser`, which assume
/// null-first Option unions and Serialize field order).
pub fn encode_with_schema<T: serde::Serialize>(value: &T, schema: &Schema) ->
Result<Vec<u8>> {
diff --git a/crates/core/src/schema/batch_evolution.rs
b/crates/core/src/schema/batch_evolution.rs
index 0bd8978a..c6d4bd26 100644
--- a/crates/core/src/schema/batch_evolution.rs
+++ b/crates/core/src/schema/batch_evolution.rs
@@ -232,7 +232,7 @@ fn is_repair_target(table: &DataType) -> bool {
/// judged from the TABLE schema alone.
///
/// Meant to be computed ONCE per scan by whoever supplies the predicate, and
handed
-/// to the reader via
[`HoodieFileGroupReaderBuilder::with_repair_risk_columns`]
+/// to the reader via `HoodieFileGroupReaderBuilder::with_repair_risk_columns`
/// (crate::file_group::reader_v2). Two properties earn it that place:
///
/// * **It is usually empty.** Spark's `TimestampType` maps to micros, so a
diff --git a/crates/core/src/schema/extended_promotion.rs
b/crates/core/src/schema/extended_promotion.rs
index 052956bb..c12def24 100644
--- a/crates/core/src/schema/extended_promotion.rs
+++ b/crates/core/src/schema/extended_promotion.rs
@@ -44,7 +44,7 @@
//!
//! ## Union handling — keyed on the reader, throws on malformed
//! Like Java, the UNION case is keyed on the READER type (Java line 1482):
only when
-//! the reader is a union are both sides unwrapped via
[`actual_schema_from_union`]
+//! the reader is a union are both sides unwrapped via
`actual_schema_from_union`
//! (Java's `getActualSchemaFromUnion(.., null)`). A writer union feeding a
*plain*
//! reader is therefore NOT unwrapped — `physical_type` reports it as `UNION`,
so a
//! `[null,int]` writer → plain `long` reader rewrites, matching Java's
@@ -237,7 +237,7 @@ fn needs_rewrite_to_string(writer: &Schema, reader_is_enum:
bool) -> bool {
/// See module docs. Port of Java's
`recordNeedsRewriteForExtendedAvroTypePromotion`
/// (`HoodieAvroUtils.java:1451-1494`).
///
-/// Returns `Err` for a malformed union (see [`actual_schema_from_union`]),
exactly
+/// Returns `Err` for a malformed union (see `actual_schema_from_union`),
exactly
/// where Java throws `HoodieAvroSchemaException`.
pub fn record_needs_rewrite_for_extended_promotion(
writer: &Schema,
diff --git a/crates/core/src/schema/resolver.rs
b/crates/core/src/schema/resolver.rs
index 852e2d7d..6f190177 100644
--- a/crates/core/src/schema/resolver.rs
+++ b/crates/core/src/schema/resolver.rs
@@ -61,7 +61,7 @@ pub async fn resolve_schema(table: &Table) -> Result<Schema> {
/// ### Note
///
/// - For resolving Avro schema, we don't read the schema from a base file
like we do when resolving Arrow schema.
-/// - Avro schema does not contain [`MetaField`]s.
+/// - Avro schema does not contain
[`MetaField`](crate::metadata::meta_field::MetaField)s.
pub async fn resolve_avro_schema(table: &Table) -> Result<String> {
match table.get_timeline().get_latest_avro_schema().await {
Ok(schema) => Ok(schema),
diff --git a/crates/core/src/storage/file_metadata.rs
b/crates/core/src/storage/file_metadata.rs
index ae1e453f..a83bdcfd 100644
--- a/crates/core/src/storage/file_metadata.rs
+++ b/crates/core/src/storage/file_metadata.rs
@@ -26,12 +26,12 @@ pub struct FileMetadata {
pub size: u64,
/// Uncompressed size in bytes in memory.
- /// Currently populated using
[crate::statistics::estimator::FileStatsEstimator]
+ /// Currently populated using `FileStatsEstimator`
/// when an estimator is available; otherwise stays 0.
pub byte_size: i64,
/// Number of records in the file.
- /// Currently populated using
[crate::statistics::estimator::FileStatsEstimator]
+ /// Currently populated using `FileStatsEstimator`
/// when an estimator is available; otherwise stays 0.
pub num_records: i64,
}
diff --git a/crates/core/src/table/builder.rs b/crates/core/src/table/builder.rs
index a798ec64..1c5effe9 100644
--- a/crates/core/src/table/builder.rs
+++ b/crates/core/src/table/builder.rs
@@ -17,6 +17,8 @@
* under the License.
*/
+//! [`TableBuilder`] opens a [`Table`] with options.
+
use paste::paste;
use std::collections::HashMap;
use std::path::PathBuf;
@@ -35,7 +37,11 @@ use crate::table::validation::validate_configs;
use crate::timeline::Timeline;
use crate::util::collection::extend_if_absent;
-/// Builder for creating a [Table] instance.
+/// Builder that opens an existing [`Table`] with options.
+///
+/// `with_hudi_option(s)` set Hudi configs, `with_storage_option(s)` set
storage options, and
+/// `with_option(s)` take either, routing keys that start with `hoodie.` to
Hudi configs.
+/// [`OptionResolver::resolve_options`] describes how these combine with the
table's own configs.
#[derive(Debug, Clone)]
pub struct TableBuilder {
option_resolver: OptionResolver,
@@ -44,9 +50,13 @@ pub struct TableBuilder {
/// Resolver for options including Hudi options, storage options, and generic
options.
#[derive(Debug, Clone)]
pub struct OptionResolver {
+ /// The table's base URI.
pub base_uri: String,
+ /// Hudi configs (`hoodie.*`).
pub hudi_options: HashMap<String, String>,
+ /// Storage options, such as cloud credentials.
pub storage_options: HashMap<String, String>,
+ /// Options not yet split into Hudi configs and storage options.
pub options: HashMap<String, String>,
}
@@ -102,6 +112,15 @@ impl TableBuilder {
TableBuilder { option_resolver }
}
+ /// Resolves the options, validates the table configs, and loads the
timeline.
+ ///
+ /// Read configs that select which read to perform (query type and
timestamps) are dropped
+ /// with a warning; pass them in
[`ReadOptions`](crate::table::ReadOptions) instead.
+ ///
+ /// # Errors
+ ///
+ /// Fails, among other cases, when `.hoodie/hoodie.properties` cannot be
read, or the table
+ /// configs are invalid or unsupported (for example a table version other
than 6, 8, or 9).
pub async fn build(&mut self) -> Result<Table> {
let option_resolver = &mut self.option_resolver;
@@ -238,7 +257,7 @@ impl OptionResolver {
/// - Cloud storage options specified via env vars
/// 4. Global Hudi configs
///
- /// [note] Error may occur when 1 and 2 have conflicts.
+ /// Note: an error may occur when 1 and 2 have conflicts.
pub async fn resolve_options(&mut self) -> Result<()> {
self.resolve_user_provided_options();
diff --git a/crates/core/src/table/mod.rs b/crates/core/src/table/mod.rs
index cd61dcfd..a6cbf845 100644
--- a/crates/core/src/table/mod.rs
+++ b/crates/core/src/table/mod.rs
@@ -16,77 +16,17 @@
* specific language governing permissions and limitations
* under the License.
*/
-//! This module is responsible for Hudi table APIs.
+//! The [`Table`] API for reading and writing a Hudi table.
//!
-//! It provides a quick entry point for reading Hudi table metadata and data,
-//! facilitating adaptation and compatibility across various engines.
+//! Open an existing table with [`Table::new`] or [`builder::TableBuilder`],
or create one with
+//! [`Table::create`]. Reads take [`ReadOptions`]; [`Table::read`] and
[`Table::read_stream`] return
+//! records, while [`Table::get_file_slices`] and
[`Table::create_file_group_reader_with_options`]
+//! let a query engine plan and distribute the scan itself. A write
([`Table::append`],
+//! [`Table::upsert`], [`Table::delete`], [`Table::update`],
[`Table::overwrite`]) that changes
+//! data creates one commit and refreshes the handle so later reads see it. On
a merge-on-read
+//! table, [`Table::compact`] folds log files into new base files.
//!
-//! **Example**
-//! 1. create hudi table
-//! ```rust
-//! use url::Url;
-//! use hudi_core::table::Table;
-//!
-//! pub async fn test() {
-//! let base_uri = Url::from_file_path("/tmp/hudi_data").unwrap();
-//! let hudi_table = Table::new(base_uri.path()).await.unwrap();
-//! }
-//! ```
-//! 2. get hudi table schema(arrow_schema::Schema)
-//! ```rust
-//! use url::Url;
-//! use hudi_core::table::Table;
-//!
-//! pub async fn test() {
-//! use arrow_schema::Schema;
-//! let base_uri = Url::from_file_path("/tmp/hudi_data").unwrap();
-//! let hudi_table = Table::new(base_uri.path()).await.unwrap();
-//! let schema = hudi_table.get_schema().await.unwrap();
-//! }
-//! ```
-//! 3. read hudi table
-//! ```rust
-//! use url::Url;
-//! use hudi_core::table::{ReadOptions, Table};
-//!
-//! pub async fn test() {
-//! let base_uri = Url::from_file_path("/tmp/hudi_data").unwrap();
-//! let hudi_table = Table::new(base_uri.path()).await.unwrap();
-//! let record_read = hudi_table.read(&ReadOptions::new()).await.unwrap();
-//! }
-//! ```
-//! 4. get file slice
-//! Users can obtain metadata to customize reading methods, read in
batches, perform parallel reads, and more.
-//! ```rust
-//! use url::Url;
-//! use hudi_core::table::{ReadOptions, Table};
-//! use hudi_core::storage::util::parse_uri;
-//! use hudi_core::storage::util::join_url_segments;
-//!
-//! pub async fn test() {
-//! let base_uri = Url::from_file_path("/tmp/hudi_data").unwrap();
-//! let hudi_table = Table::new(base_uri.path()).await.unwrap();
-//! let flat_slices = hudi_table
-//! .get_file_slices(&ReadOptions::new())
-//! .await.unwrap();
-//! let file_slices =
hudi_core::util::collection::split_into_chunks(flat_slices, 2);
-//! // define every parquet task reader how many slice
-//! let mut parquet_file_groups: Vec<Vec<String>> = Vec::new();
-//! for file_slice_vec in file_slices {
-//! let file_group_vec = file_slice_vec
-//! .iter()
-//! .filter_map(|f| {
-//! // None when the slice's records live entirely in log
-//! // files, so there is no base file to read.
-//! let relative_path =
f.base_file_relative_path().unwrap()?;
-//! let url = join_url_segments(&base_uri,
&[relative_path.as_str()]).unwrap();
-//! Some(url.path().to_string())
-//! })
-//! .collect();
-//! parquet_file_groups.push(file_group_vec)
-//! }
-//! }
-//! ```
+//! See the crate-level docs for examples.
pub mod builder;
pub mod file_pruner;
@@ -134,12 +74,19 @@ use std::sync::Arc;
use tokio::sync::OnceCell;
use url::Url;
-/// The main struct that provides table APIs for interacting with a Hudi table.
+/// A handle to a Hudi table, with the read and write APIs.
+///
+/// Clones share configs and caches. Write methods take `&mut self` and refresh
+/// that handle's timeline; reopen or rebuild other handles to see the new
commit.
#[derive(Debug)]
pub struct Table {
+ /// Resolved Hudi configs: user options, `hoodie.properties`, and
`hudi-defaults.conf`.
pub hudi_configs: Arc<HudiConfigs>,
+ /// Options for the storage backend, such as cloud credentials.
pub storage_options: Arc<HashMap<String, String>>,
+ /// The table's timeline of commits.
pub timeline: Timeline,
+ /// Listing of the table's file groups and file slices.
pub file_system_view: FileSystemView,
/// Cached metadata table instance, lazily initialized on first use.
/// Shared across clones via `Arc`; [`Table::clear_read_caches`] gives this
@@ -354,12 +301,19 @@ impl Table {
None
}
- /// Create hudi table by base_uri
+ /// Opens the table at `base_uri`, a local path or an object store URI
(`s3://`, `az://`,
+ /// `gs://`, `file://`).
+ ///
+ /// Equivalent to `TableBuilder::from_base_uri(base_uri).build()`; see
+ /// [`builder::TableBuilder`] for how options are resolved.
pub async fn new(base_uri: &str) -> Result<Self> {
TableBuilder::from_base_uri(base_uri).build().await
}
- /// Create hudi table with options
+ /// Opens the table at `base_uri` with extra options.
+ ///
+ /// Keys starting with `hoodie.` are Hudi configs; all others are storage
options, such as
+ /// `aws_region`.
pub async fn new_with_options<I, K, V>(base_uri: &str, options: I) ->
Result<Self>
where
I: IntoIterator<Item = (K, V)>,
@@ -375,14 +329,18 @@ impl Table {
/// Start creating a new Hudi table at `base_uri` (does not open until
[`TableCreateBuilder::create`]).
///
/// # Example
- /// ```no_run
- /// # async fn example() -> hudi_core::error::Result<()> {
+ /// ```
+ /// # #[tokio::main]
+ /// # async fn main() -> hudi_core::error::Result<()> {
+ /// # let dir = tempfile::tempdir()?;
+ /// # let base_uri = dir.path().to_str().unwrap();
/// use hudi_core::table::Table;
- /// let table = Table::create("/tmp/hudi_table")
+ /// let table = Table::create(base_uri)
/// .with_table_name("trips")
/// .with_record_key_fields(["id"])
/// .create()
/// .await?;
+ /// assert_eq!(table.table_name(), "trips");
/// # Ok(())
/// # }
/// ```
@@ -412,7 +370,12 @@ impl Table {
crate::write::upsert_batches(self, &batches, options).await
}
- /// Replace all data in a copy-on-write table.
+ /// Replaces all data in the table with `batches`
(`INSERT_OVERWRITE_TABLE`), as one
+ /// replacecommit.
+ ///
+ /// # Errors
+ ///
+ /// Fails, among other cases, when `batches` is empty or does not match
the table schema.
pub async fn overwrite(
&mut self,
batches: impl IntoIterator<Item = RecordBatch>,
@@ -421,7 +384,13 @@ impl Table {
crate::write::overwrite_batches(self, &batches).await
}
- /// Replace only the partitions present in `batches` in a partitioned
copy-on-write table.
+ /// Replaces only the partitions present in `batches`
(`INSERT_OVERWRITE`), leaving the others
+ /// untouched, as one replacecommit.
+ ///
+ /// # Errors
+ ///
+ /// Fails, among other cases, with [`CoreError::Unsupported`] on an
unpartitioned table, and
+ /// when `batches` is empty or does not match the table schema.
pub async fn dynamic_partition_overwrite(
&mut self,
batches: impl IntoIterator<Item = RecordBatch>,
@@ -455,9 +424,9 @@ impl Table {
crate::write::delete_keys(self, &keys).await
}
- /// Compact a merge-on-read table: every file group whose latest file
- /// slice carries log files gets a new base file holding its merged rows,
- /// under one compaction instant that completes as a `commit`.
+ /// Compact a merge-on-read table (experimental): every file group whose
+ /// latest file slice carries log files gets a new base file holding its
+ /// merged rows, under one compaction instant that completes as a `commit`.
///
/// Rows keep their `_hoodie_commit_time`, so incremental reads see no new
/// rows from a compaction. Returns a result whose `requested_time` is
@@ -471,8 +440,8 @@ impl Table {
///
/// # Errors
///
- /// Returns
[`CoreError::Unsupported`](crate::error::CoreError::Unsupported)
- /// on a copy-on-write table, which has no log files to compact.
+ /// Returns [`CoreError::Unsupported`] on a copy-on-write table, which has
no log files to
+ /// compact.
pub async fn compact(&mut self) -> Result<CompactionResult> {
crate::write::compact_table(self).await
}
@@ -502,14 +471,18 @@ impl Table {
crate::write::update_filter(self, filter, updates).await
}
+ /// A copy of the resolved Hudi configs as key-value pairs.
pub fn hudi_options(&self) -> HashMap<String, String> {
self.hudi_configs.as_options()
}
+ /// A copy of the storage options as key-value pairs.
pub fn storage_options(&self) -> HashMap<String, String> {
self.storage_options.as_ref().clone()
}
+ /// Registers the table's object stores with a DataFusion runtime, so
DataFusion can read the
+ /// table's files.
#[cfg(feature = "datafusion")]
pub fn register_storage(
&self,
@@ -523,6 +496,12 @@ impl Table {
.register_object_store(runtime_env.clone());
}
+ /// The table's base path as a URL.
+ ///
+ /// # Panics
+ ///
+ /// Panics if `hoodie.base.path` is missing or not a valid URI, which a
table opened through
+ /// [`builder::TableBuilder`] always has.
pub fn base_url(&self) -> Url {
let err_msg = format!("{:?} is missing or invalid.",
HudiTableConfig::BasePath);
self.hudi_configs
@@ -532,6 +511,12 @@ impl Table {
.expect(&err_msg)
}
+ /// The table name (`hoodie.table.name`).
+ ///
+ /// # Panics
+ ///
+ /// Panics if the config is missing, which table opening rejects unless
config validation is
+ /// skipped.
pub fn table_name(&self) -> String {
let err_msg = format!("{:?} is missing or invalid.",
HudiTableConfig::TableName);
self.hudi_configs
@@ -540,6 +525,12 @@ impl Table {
.into()
}
+ /// The table type: `COPY_ON_WRITE` or `MERGE_ON_READ`.
+ ///
+ /// # Panics
+ ///
+ /// Panics if the config is missing or not a valid table type, which table
opening rejects
+ /// unless config validation is skipped.
pub fn table_type(&self) -> String {
let err_msg = format!("{:?} is missing or invalid.",
HudiTableConfig::TableType);
self.hudi_configs
@@ -548,6 +539,11 @@ impl Table {
.into()
}
+ /// Whether this is a merge-on-read table.
+ ///
+ /// # Panics
+ ///
+ /// Panics under the same condition as [`Self::table_type`].
pub fn is_mor(&self) -> bool {
self.table_type() == TableTypeValue::MergeOnRead.as_ref()
}
@@ -556,6 +552,9 @@ impl Table {
Ok(!self.is_mor() || options.is_read_optimized()?)
}
+ /// The timezone of the timeline's instant times
(`hoodie.table.timeline.timezone`): `utc` or
+ /// `local`, and `utc` when the config is unset. [`Table::create`] records
`local` unless that
+ /// config is passed with [`TableCreateBuilder::with_option`].
pub fn timezone(&self) -> String {
self.hudi_configs
.get_or_default(HudiTableConfig::TimelineTimezone)
diff --git a/crates/core/src/table/partition.rs
b/crates/core/src/table/partition.rs
index b9a6a9de..1539658d 100644
--- a/crates/core/src/table/partition.rs
+++ b/crates/core/src/table/partition.rs
@@ -16,6 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
+//! Partition pruning and partition path helpers.
+
use crate::Result;
use crate::config::HudiConfigs;
use crate::config::table::HudiTableConfig;
@@ -33,7 +35,9 @@ use crate::metadata::meta_field::MetaField;
use std::collections::HashMap;
use std::sync::Arc;
+/// Name of the metadata file in each partition directory.
pub const PARTITION_METAFIELD_PREFIX: &str = ".hoodie_partition_metadata";
+/// The partition path of an unpartitioned table.
pub const EMPTY_PARTITION_PATH: &str = "";
/// Build the partition [Schema] in the order declared by
`partition_field_names`.
@@ -65,6 +69,7 @@ pub(crate) fn project_partition_schema(
Ok(Schema::new(fields))
}
+/// Whether the table has partition fields and does not use a non-partitioned
key generator.
pub fn is_table_partitioned(hudi_configs: &HudiConfigs) -> Result<bool> {
let has_partition_fields = {
let partition_fields: Vec<String> =
hudi_configs.get_or_default(PartitionFields).into();
@@ -102,6 +107,8 @@ pub struct PartitionPruner {
}
impl PartitionPruner {
+ /// Creates a pruner from the filters that apply to `partition_schema`.
Filters on other
+ /// columns, and filters whose value cannot be cast to the column type,
are ignored.
pub fn new(
and_filters: &[Filter],
partition_schema: &Schema,
diff --git a/crates/core/src/timeline/instant.rs
b/crates/core/src/timeline/instant.rs
index 30c19a53..74903eb3 100644
--- a/crates/core/src/timeline/instant.rs
+++ b/crates/core/src/timeline/instant.rs
@@ -267,7 +267,7 @@ impl Instant {
/// 1. Date format: `yyyyMMddHHmmss` (14 chars) or `yyyyMMddHHmmssSSS` (17
chars)
/// 2. Epoch milliseconds: 17-digit number representing milliseconds since
Unix epoch
/// (used by metadata table for timestamps like `00000000000000000`),
or one
- /// of the [`Self::BOOTSTRAP_INSTANT_TIMES`] sentinels.
+ /// of the `Self::BOOTSTRAP_INSTANT_TIMES` sentinels.
///
/// The function tries date format first, then falls back to epoch
milliseconds
/// if the date parsing fails (e.g., invalid month/day like `00`).
diff --git a/crates/core/src/timeline/loader.rs
b/crates/core/src/timeline/loader.rs
index c94613bd..46d12a9c 100644
--- a/crates/core/src/timeline/loader.rs
+++ b/crates/core/src/timeline/loader.rs
@@ -156,7 +156,7 @@ impl TimelineLoader {
/// Every instant time that has a `requested` or `inflight` file and no
/// completed one, for **any** action.
///
- /// Deliberately not built on [`TimelineSelector`], which resolves the
action
+ /// Deliberately not built on `TimelineSelector`, which resolves the action
/// through [`Action`](crate::timeline::instant::Action) and so can only
see
/// the three commit actions the enum names. Java's
`getValidInstantTimestamps`
/// takes its `datasetPendingInstants` from `filterInflightsAndRequested()`
diff --git a/crates/core/src/timeline/mod.rs b/crates/core/src/timeline/mod.rs
index 7b2801b2..a2c40573 100644
--- a/crates/core/src/timeline/mod.rs
+++ b/crates/core/src/timeline/mod.rs
@@ -89,7 +89,7 @@ pub const EARLIEST_START_TIMESTAMP: &str =
"19700101000000000";
/// The actions a read loads from the timeline.
///
-/// Narrowing this is no longer only a listing decision:
[`Timeline::completion_gate_inputs`]
+/// Narrowing this is no longer only a listing decision:
`Timeline::completion_gate_inputs`
/// derives the log scan's committed set from these, so an action that
completes outside this
/// list and wrote data or delete log blocks would have its rows dropped from
a merge-on-read
/// read — silently, since the gate excludes rather than errors. The three
here cover every
@@ -423,8 +423,8 @@ impl Timeline {
/// Get the latest [apache_avro::schema::Schema] as [String] from the
[Timeline].
///
/// ### Note
- /// This API behaves differently from
[crate::table::Table::get_avro_schema],
- /// which additionally looks for [HudiTableConfig::CreateSchema] in the
table config.
+ /// This API behaves differently from
[crate::table::Table::get_schema_in_avro_str],
+ /// which additionally looks for
[HudiTableConfig::CreateSchema](crate::config::table::HudiTableConfig::CreateSchema)
in the table config.
pub async fn get_latest_avro_schema(&self) -> Result<String> {
let commit_metadata = self.get_latest_commit_metadata().await?;
resolve_avro_schema_from_commit_metadata(&commit_metadata)
@@ -434,7 +434,7 @@ impl Timeline {
///
/// ### Note
/// This API behaves differently from [crate::table::Table::get_schema],
- /// which additionally looks for [HudiTableConfig::CreateSchema] in the
table config.
+ /// which additionally looks for
[HudiTableConfig::CreateSchema](crate::config::table::HudiTableConfig::CreateSchema)
in the table config.
pub async fn get_latest_schema(&self) -> Result<Schema> {
let commit_metadata = self.get_latest_commit_metadata().await?;
resolve_data_schema_from_commit_metadata(&commit_metadata,
self.storage.clone()).await
diff --git a/crates/core/src/timeline/selector.rs
b/crates/core/src/timeline/selector.rs
index a9c0e887..7c784771 100644
--- a/crates/core/src/timeline/selector.rs
+++ b/crates/core/src/timeline/selector.rs
@@ -70,7 +70,7 @@ impl InstantRange {
/// A range that admits exactly the instants in `instants`, and nothing
else.
///
- /// See [`Self::explicit_instants`] for why a window cannot stand in for
this.
+ /// See `Self::explicit_instants` for why a window cannot stand in for
this.
pub fn exact_match<I, S>(instants: I, timezone: &str) -> Self
where
I: IntoIterator<Item = S>,
diff --git a/crates/core/src/write/create.rs b/crates/core/src/write/create.rs
index b21ad23f..e0aa68e1 100644
--- a/crates/core/src/write/create.rs
+++ b/crates/core/src/write/create.rs
@@ -92,6 +92,8 @@ pub struct TableCreateBuilder {
}
impl TableCreateBuilder {
+ /// Starts a builder for a table at `base_uri`; same as
+ /// [`Table::create`](crate::table::Table::create).
pub fn new(base_uri: impl Into<String>) -> Self {
Self {
base_uri: base_uri.into(),
@@ -116,31 +118,43 @@ impl TableCreateBuilder {
}
}
+ /// Sets the table name, which is required and must be non-empty.
pub fn with_table_name(mut self, name: impl Into<String>) -> Self {
self.table_name = Some(name.into());
self
}
+ /// Sets the table type. Defaults to [`TableTypeValue::CopyOnWrite`].
pub fn with_table_type(mut self, table_type: TableTypeValue) -> Self {
self.table_type = table_type;
self
}
+ /// Sets the table version: `9` (the default) or `8`; [`Self::create`]
rejects other values.
+ /// A later [`Self::with_metadata_table_enabled`]`(true)` raises a version
below 8 to 8.
pub fn with_table_version(mut self, version: isize) -> Self {
self.table_version = version;
self
}
+ /// Sets the timeline layout version. Defaults to `2`, the only layout
table versions 8 and 9
+ /// accept.
pub fn with_timeline_layout_version(mut self, version: isize) -> Self {
self.timeline_layout_version = version;
self
}
+ /// Whether records carry the `_hoodie_*` meta fields. Defaults to `true`.
pub fn with_populates_meta_fields(mut self, value: bool) -> Self {
self.populates_meta_fields = value;
self
}
+ /// Sets the record key fields. Several fields form a composite key,
stored as `field:value`
+ /// pairs joined by commas. With none, appends generate keys, and
+ /// [`Table::upsert`](crate::table::Table::upsert),
+ /// [`Table::delete_keys`](crate::table::Table::delete_keys), and
+ /// [`Table::update`](crate::table::Table::update) on a merge-on-read
table are rejected.
pub fn with_record_key_fields<I, S>(mut self, fields: I) -> Self
where
I: IntoIterator<Item = S>,
@@ -150,6 +164,7 @@ impl TableCreateBuilder {
self
}
+ /// Sets the columns whose values form the partition path. Unset means
unpartitioned.
pub fn with_partition_fields<I, S>(mut self, fields: I) -> Self
where
I: IntoIterator<Item = S>,
@@ -251,6 +266,7 @@ impl TableCreateBuilder {
self
}
+ /// Sets a storage option, such as a cloud credential, used to create and
then open the table.
pub fn with_storage_option(mut self, key: impl AsRef<str>, value: impl
Into<String>) -> Self {
self.storage_options
.insert(key.as_ref().to_string(), value.into());
@@ -258,6 +274,14 @@ impl TableCreateBuilder {
}
/// Create the table layout on storage and open a [`Table`] handle.
+ ///
+ /// # Errors
+ ///
+ /// Fails, among other cases, when a table already exists at the base URI,
the table name is
+ /// missing or empty, [`Self::with_option`] set a key that a dedicated
builder method owns,
+ /// the version settings are unsupported, or an index is enabled without
what it depends on
+ /// (the record index and `column_stats` need the metadata table,
`partition_stats` needs
+ /// `column_stats`).
pub async fn create(mut self) -> Result<Table> {
// The builder derives the metadata table layout, key generator and
// checksum from these, so an override would persist a table that
diff --git a/crates/core/src/write/lock.rs b/crates/core/src/write/lock.rs
index 70df42f6..87da18d6 100644
--- a/crates/core/src/write/lock.rs
+++ b/crates/core/src/write/lock.rs
@@ -63,6 +63,7 @@ pub struct InProcessLockProvider {
}
impl InProcessLockProvider {
+ /// A provider for the table at `base_path`. Providers for the same path
share one lock.
pub fn new(base_path: impl Into<String>) -> Self {
Self {
base_path: base_path.into(),
diff --git a/crates/core/src/write/mod.rs b/crates/core/src/write/mod.rs
index c81505d7..28421980 100644
--- a/crates/core/src/write/mod.rs
+++ b/crates/core/src/write/mod.rs
@@ -16,7 +16,16 @@
* specific language governing permissions and limitations
* under the License.
*/
-//! Write APIs for Hudi tables (`Table` verbs).
+//! Write APIs for Hudi tables.
+//!
+//! The writes are methods on [`Table`](crate::table::Table): `append`,
`upsert`, `upsert_with`,
+//! `update`, `delete`, `delete_keys`, `overwrite`, and
`dynamic_partition_overwrite`. A write that
+//! changes data is one commit, and updates the metadata table (when enabled)
in the same
+//! transaction. Merge-on-read tables also compact (experimental) with
`compact` and
+//! `await_compaction`. This module holds [`WriteResult`], [`UpsertOptions`],
+//! [`CompactionResult`], the [`TableCreateBuilder`] behind
+//! [`Table::create`](crate::table::Table::create), and the lock provider
types. Every write
+//! returns a [`WriteResult`].
pub(crate) mod append;
pub(crate) mod archival;
diff --git a/crates/datafusion/src/lib.rs b/crates/datafusion/src/lib.rs
index 2900fbd9..7f64c7cb 100644
--- a/crates/datafusion/src/lib.rs
+++ b/crates/datafusion/src/lib.rs
@@ -17,6 +17,21 @@
* under the License.
*/
+//! [Apache DataFusion](https://datafusion.apache.org) integration for Apache
Hudi tables.
+//!
+//! [`HudiDataSource`] is a DataFusion `TableProvider` to register with a
`SessionContext`;
+//! [`HudiTableFactory`] serves `CREATE EXTERNAL TABLE ... STORED AS HUDI`.
Both read through
+//! `hudi-core` and accept the same table options.
+//!
+//! Applications usually reach this crate through the
[`hudi`](https://docs.rs/hudi) crate's
+//! `datafusion` feature. The [Python and Rust quick start guide][quick start
guide] shows
+//! DataFusion queries on a hudi-rs table, and the
[`hudi`](https://docs.rs/hudi) crate docs list
+//! every resource (API references, quick starts, examples, specs).
+//!
+//! [quick start guide]:
https://hudi.apache.org/docs/python-rust-quick-start-guide
+
+#![warn(missing_docs)]
+
pub(crate) mod hudi_exec;
pub(crate) mod util;
@@ -87,8 +102,10 @@ fn filter_field_matches_partition_column(filter_field:
&str, partition_column: &
.is_some_and(|(_, name)| name == partition_column)
}
-/// Create a `HudiDataSource`.
-/// Used for Datafusion to query Hudi tables
+/// A DataFusion table provider for Apache Hudi tables.
+///
+/// The provider reads the table as of its creation, including its schema and
statistics; create
+/// a new provider to see later commits.
///
/// # Examples
///
@@ -116,7 +133,6 @@ fn filter_field_matches_partition_column(filter_field:
&str, partition_column: &
/// Ok(())
/// }
/// ```
-/// A DataFusion table provider for Apache Hudi tables.
#[derive(Clone)]
pub struct HudiDataSource {
table: Arc<HudiTable>,
@@ -172,10 +188,22 @@ impl std::fmt::Debug for HudiDataSource {
}
impl HudiDataSource {
+ /// Creates a provider for the table at `base_uri` with default options.
pub async fn new(base_uri: &str) -> Result<Self> {
Self::new_with_options(base_uri, empty_options()).await
}
+ /// Creates a provider for the table at `base_uri` with options, which are
passed on to
+ /// [`Table::new_with_options`](hudi_core::table::Table::new_with_options).
+ ///
+ /// The provider also reads `hoodie.read.input.partitions` (default `0`,
which defers to
+ /// DataFusion's `target_partitions`),
`hoodie.read.use.read_optimized.mode`, and
+ /// `hoodie.read.file.slice.read.concurrency` from these options.
+ ///
+ /// # Errors
+ ///
+ /// Fails when one of those options has an invalid value, the table cannot
be opened, or the
+ /// table's base file format is HFile.
pub async fn new_with_options<I, K, V>(base_uri: &str, options: I) ->
Result<Self>
where
I: IntoIterator<Item = (K, V)>,
@@ -1014,6 +1042,7 @@ impl TableProvider for HudiDataSource {
pub struct HudiTableFactory {}
impl HudiTableFactory {
+ /// Creates a factory.
pub fn new() -> Self {
Self {}
}
diff --git a/crates/hudi/Cargo.toml b/crates/hudi/Cargo.toml
index 5b72f12f..eca2ec21 100644
--- a/crates/hudi/Cargo.toml
+++ b/crates/hudi/Cargo.toml
@@ -48,4 +48,10 @@ datafusion = [
]
[dev-dependencies]
-tokio = { workspace = true }
+arrow = { workspace = true }
+datafusion = { workspace = true }
+tempfile = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
+
+[package.metadata.docs.rs]
+features = ["datafusion"]
diff --git a/crates/hudi/src/lib.rs b/crates/hudi/src/lib.rs
index 2fa10969..14d19c5c 100644
--- a/crates/hudi/src/lib.rs
+++ b/crates/hudi/src/lib.rs
@@ -16,6 +16,308 @@
* specific language governing permissions and limitations
* under the License.
*/
+//! A native Rust implementation of [Apache Hudi](https://hudi.apache.org).
+//!
+//! This crate creates, writes, and reads Hudi tables on local disk or cloud
object storage
+//! (`s3://`, `az://`, `gs://`), exchanging data as Arrow [`RecordBatch`]es.
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][quick start guide]
on the Hudi website,
+//! and the shorter [README quick start].
+//! - Rust API reference: this crate,
[`hudi-core`](https://docs.rs/hudi-core), and
+//! [`hudi-datafusion`](https://docs.rs/hudi-datafusion).
+//! - Python API reference: the docstrings in the package's type stubs,
+//! [`python/hudi/_internal.pyi`][python stubs].
+//! - Runnable Rust example: [`crates/core/examples/table_writes.rs`][rust
example].
+//! - Reader semantics: [`docs/reader-spec.md`][reader spec]. Writer design:
+//! [`docs/writer-design.md`][writer design].
+//! - Packages: [crates.io](https://crates.io/crates/hudi) and
+//! [PyPI](https://pypi.org/project/hudi/). Source:
+//! [github.com/apache/hudi-rs](https://github.com/apache/hudi-rs).
+//! - Hudi concepts (timeline, table types, query types):
+//! [hudi.apache.org/docs/overview](https://hudi.apache.org/docs/overview).
+//!
+//! [quick start guide]:
https://hudi.apache.org/docs/python-rust-quick-start-guide
+//! [README quick start]: https://github.com/apache/hudi-rs#quick-start
+//! [python stubs]:
https://github.com/apache/hudi-rs/blob/main/python/hudi/_internal.pyi
+//! [rust example]:
https://github.com/apache/hudi-rs/blob/main/crates/core/examples/table_writes.rs
+//! [reader spec]:
https://github.com/apache/hudi-rs/blob/main/docs/reader-spec.md
+//! [writer design]:
https://github.com/apache/hudi-rs/blob/main/docs/writer-design.md
+//!
+//! [`RecordBatch`]:
https://docs.rs/arrow/latest/arrow/array/struct.RecordBatch.html
+//!
+//! # 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`](error::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`](table::Table::append) inserts without an
+//! index lookup; [`Table::upsert`](table::Table::upsert) updates rows whose
record key exists and
+//! inserts the rest; [`Table::delete`](table::Table::delete) and
+//! [`Table::delete_keys`](table::Table::delete_keys) remove rows by predicate
or by key.
+//!
+//! ```
+//! # use std::sync::Arc;
+//! # use arrow::array::{ArrayRef, Int64Array, RecordBatch, StringArray};
+//! use hudi::config::table::TableTypeValue;
+//! use hudi::table::{HoodieKey, Table};
+//!
+//! # fn trips(rows: &[(&str, &str, i64)]) -> RecordBatch {
+//! # let id: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.0)));
+//! # let city: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.1)));
+//! # let ts: ArrayRef =
Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.2)));
+//! # RecordBatch::try_from_iter([("id", id), ("city", city), ("ts",
ts)]).unwrap()
+//! # }
+//! # #[tokio::main]
+//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
+//! # let dir = tempfile::tempdir()?;
+//! # let base_uri = dir.path().to_str().unwrap();
+//! 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);
+//! # Ok(())
+//! # }
+//! ```
+//!
+//! # Compact a merge-on-read table
+//!
+//! On a merge-on-read table, upserts, updates, and deletes write log files.
+//! [`Table::compact`](table::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`](table::Table::await_compaction) waits for.
+//!
+//! ```
+//! # use std::sync::Arc;
+//! # use arrow::array::{ArrayRef, Int64Array, RecordBatch, StringArray};
+//! use hudi::config::table::TableTypeValue;
+//! use hudi::table::Table;
+//!
+//! # fn trips(rows: &[(&str, &str, i64)]) -> RecordBatch {
+//! # let id: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.0)));
+//! # let city: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.1)));
+//! # let ts: ArrayRef =
Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.2)));
+//! # RecordBatch::try_from_iter([("id", id), ("city", city), ("ts",
ts)]).unwrap()
+//! # }
+//! # #[tokio::main]
+//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
+//! # let dir = tempfile::tempdir()?;
+//! # let base_uri = dir.path().to_str().unwrap();
+//! 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());
+//! # Ok(())
+//! # }
+//! ```
+//!
+//! # Snapshot read
+//!
+//! [`Table::new`](table::Table::new) opens an existing table. A read returns
the latest committed
+//! state, narrowed by optional filters and a projection from
[`ReadOptions`](table::ReadOptions).
+//!
+//! ```
+//! # use std::sync::Arc;
+//! # use arrow::array::{ArrayRef, Int64Array, RecordBatch, StringArray};
+//! use hudi::table::{ReadOptions, Table};
+//! # fn trips(rows: &[(&str, &str, i64)]) -> RecordBatch {
+//! # let id: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.0)));
+//! # let city: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.1)));
+//! # let ts: ArrayRef =
Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.2)));
+//! # RecordBatch::try_from_iter([("id", id), ("city", city), ("ts",
ts)]).unwrap()
+//! # }
+//! # #[tokio::main]
+//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
+//! # let dir = tempfile::tempdir()?;
+//! # let base_uri = dir.path().to_str().unwrap();
+//! # let mut writer = Table::create(base_uri)
+//! # .with_table_name("trips")
+//! # .with_record_key_fields(["id"])
+//! # .with_partition_fields(["city"])
+//! # .create()
+//! # .await?;
+//! # writer.append([trips(&[("a", "sf", 1), ("b", "sf", 1), ("c", "nyc",
1)])]).await?;
+//! 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);
+//! # Ok(())
+//! # }
+//! ```
+//!
+//! [`Table::read_stream`](table::Table::read_stream) yields the same snapshot
one batch at a time.
+//!
+//! # Time-travel read
+//!
+//!
[`ReadOptions::with_as_of_timestamp`](table::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 std::sync::Arc;
+//! # use arrow::array::{ArrayRef, Int64Array, RecordBatch, StringArray};
+//! use hudi::table::{ReadOptions, Table};
+//! # fn trips(rows: &[(&str, &str, i64)]) -> RecordBatch {
+//! # let id: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.0)));
+//! # let city: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.1)));
+//! # let ts: ArrayRef =
Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.2)));
+//! # RecordBatch::try_from_iter([("id", id), ("city", city), ("ts",
ts)]).unwrap()
+//! # }
+//! # #[tokio::main]
+//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
+//! # let dir = tempfile::tempdir()?;
+//! # let base_uri = dir.path().to_str().unwrap();
+//! # let mut table = Table::create(base_uri)
+//! # .with_table_name("trips")
+//! # .with_record_key_fields(["id"])
+//! # .with_partition_fields(["city"])
+//! # .create()
+//! # .await?;
+//! 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);
+//! # Ok(())
+//! # }
+//! ```
+//!
+//! # 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`](table::WriteResult::completion_time), not its
`requested_time`.
+//!
+//! ```
+//! # use std::sync::Arc;
+//! # use arrow::array::{ArrayRef, Int64Array, RecordBatch, StringArray};
+//! use hudi::table::{QueryType, ReadOptions, Table};
+//! # fn trips(rows: &[(&str, &str, i64)]) -> RecordBatch {
+//! # let id: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.0)));
+//! # let city: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.1)));
+//! # let ts: ArrayRef =
Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.2)));
+//! # RecordBatch::try_from_iter([("id", id), ("city", city), ("ts",
ts)]).unwrap()
+//! # }
+//! # #[tokio::main]
+//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
+//! # let dir = tempfile::tempdir()?;
+//! # let base_uri = dir.path().to_str().unwrap();
+//! # let mut table = Table::create(base_uri)
+//! # .with_table_name("trips")
+//! # .with_record_key_fields(["id"])
+//! # .with_partition_fields(["city"])
+//! # .create()
+//! # .await?;
+//! 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
+//! # Ok(())
+//! # }
+//! ```
+//!
+//! # Query with DataFusion
+//!
+//! With the `datafusion` feature, register a table with a DataFusion
`SessionContext` and query
+//! it with SQL.
+//!
+//! ```
+//! # use std::sync::Arc;
+//! # use arrow::array::{ArrayRef, Int64Array, RecordBatch, StringArray};
+//! # fn trips(rows: &[(&str, &str, i64)]) -> RecordBatch {
+//! # let id: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.0)));
+//! # let city: ArrayRef =
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.1)));
+//! # let ts: ArrayRef =
Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.2)));
+//! # RecordBatch::try_from_iter([("id", id), ("city", city), ("ts",
ts)]).unwrap()
+//! # }
+//! # #[tokio::main]
+//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
+//! # let dir = tempfile::tempdir()?;
+//! # let base_uri = dir.path().to_str().unwrap();
+//! # let mut table = hudi::table::Table::create(base_uri)
+//! # .with_table_name("trips")
+//! # .with_record_key_fields(["id"])
+//! # .with_partition_fields(["city"])
+//! # .create()
+//! # .await?;
+//! # table.append([trips(&[("a", "sf", 1), ("b", "sf", 1), ("c", "nyc",
1)])]).await?;
+//! # #[cfg(feature = "datafusion")]
+//! # {
+//! 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);
+//! # }
+//! # Ok(())
+//! # }
+//! ```
pub use hudi_core::*;
