You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by xu...@apache.org on 2026/05/02 19:36:31 UTC
(hudi-rs) branch main updated: feat: unify Hudi read APIs around ReadOptions and add Python bindings (#587)
This is an automated email from the ASF dual-hosted git repository.
xushiyan 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 de2982ae feat: unify Hudi read APIs around ReadOptions and add Python bindings (#587)
de2982ae is described below
commit de2982ae7ddd3416310f379108e085fb80a4e8e7
Author: Shiyan Xu <27...@users.noreply.github.com>
AuthorDate: Sat May 2 14:36:25 2026 -0500
feat: unify Hudi read APIs around ReadOptions and add Python bindings (#587)
---
.github/copilot-instructions.md | 13 +-
.github/instructions/python.instructions.md | 2 +-
.github/instructions/rust.instructions.md | 5 +
README.md | 175 +++--
cpp/src/lib.rs | 33 +-
crates/core/src/config/read.rs | 171 ++++-
crates/core/src/expr/filter.rs | 601 +++++++++++-----
crates/core/src/file_group/reader.rs | 356 ++++++----
crates/core/src/keygen/timestamp_based.rs | 32 +-
crates/core/src/metadata/table/mod.rs | 9 +-
crates/core/src/table/file_pruner.rs | 2 +-
crates/core/src/table/mod.rs | 1001 +++++++++++++--------------
crates/core/src/table/partition.rs | 24 +-
crates/core/src/table/read_options.rs | 382 +++++++---
crates/core/src/timeline/mod.rs | 10 +-
crates/core/src/timeline/util.rs | 6 +-
crates/core/src/util/arrow.rs | 60 ++
crates/core/tests/table_read_tests.rs | 663 +++++++++++++++---
crates/datafusion/src/lib.rs | 11 +-
crates/datafusion/src/util/expr.rs | 26 +-
crates/datafusion/tests/plan_tests.rs | 18 +
demo/apps/hudi-file-group-api/cpp/main.cpp | 3 +-
demo/apps/hudi-table-api/python/src/main.py | 2 +-
demo/apps/hudi-table-api/rust/src/main.rs | 4 +-
docs/reader-spec.md | 319 +++++++++
python/hudi/__init__.py | 6 +
python/hudi/_internal.pyi | 261 ++++---
python/src/internal.rs | 521 ++++++++++----
python/src/lib.rs | 8 +-
python/tests/test_config_enums.py | 5 +-
python/tests/test_file_group_read.py | 8 +-
python/tests/test_streaming_read.py | 167 +++++
python/tests/test_table_builder.py | 27 +-
python/tests/test_table_incremental_read.py | 75 +-
python/tests/test_table_misc.py | 124 ++++
python/tests/test_table_read.py | 50 +-
36 files changed, 3700 insertions(+), 1480 deletions(-)
diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md
index 9bf60d9a..8c88cfb2 100644
--- a/.github/copilot-instructions.md
+++ b/.github/copilot-instructions.md
@@ -34,16 +34,9 @@ PR titles must follow [Conventional Commits](https://www.conventionalcommits.org
- `fix(python): handle null partition values correctly`
- `docs: update API documentation for HudiTable`
-## Copilot review behavior
-
-- Focus on the latest commits in updated PRs; don't re-raise issues already fixed.
-- Use the severity tags from [`AGENTS.md` β Code review](../AGENTS.md#code-review-rubric)
- (π΄ / π / π‘ / π¬).
-- For `crates/core` public-API changes, check `crates/datafusion`, `python/`, and `cpp/` for
- downstream impact before approving.
-- Flag `.unwrap()` / `.expect()` / `panic!()` in non-test code as π΄ Critical.
-- Flag blocking I/O in async functions as π΄ Critical.
-- Flag hardcoded credentials or secrets as π΄ Critical.
+For code review behavior, severity tags, and patterns to flag, see the path-scoped
+[`code-review.instructions.md`](./instructions/code-review.instructions.md) (loaded automatically
+for all files during review).
For everything else β build commands, coding conventions, testing, security expectations β see
[`AGENTS.md`](../AGENTS.md).
diff --git a/.github/instructions/python.instructions.md b/.github/instructions/python.instructions.md
index 36577849..3751a283 100644
--- a/.github/instructions/python.instructions.md
+++ b/.github/instructions/python.instructions.md
@@ -16,7 +16,7 @@ applyTo: "python/**"
#[pyfunction]
fn read_table(path: &str) -> PyResult<PyObject> {
let result = hudi_core::read_table(path)
- .map_err(|e| PyRuntimeError::new_err(format!("Failed to read table: {}", e)))?;
+ .map_err(|e| PyRuntimeError::new_err(format!("Failed to read table: {e}")))?;
// ...
}
diff --git a/.github/instructions/rust.instructions.md b/.github/instructions/rust.instructions.md
index 6d357e79..076fce34 100644
--- a/.github/instructions/rust.instructions.md
+++ b/.github/instructions/rust.instructions.md
@@ -35,6 +35,11 @@ let file = File::open(&path).map_err(|e| HudiError::Io(e))?;
- Flag blocking I/O (`std::fs`, `std::thread::sleep`) in async functions
- Use `tokio::task::spawn_blocking` for CPU-intensive work
+## Style
+
+- Use inline format args (Rust 1.88+): `format!("{x}")`, not `format!("{}", x)` β expressions
+ like `path.display()` still require positional args
+
## API Design
### Builder Pattern
diff --git a/README.md b/README.md
index b734efb1..17c0a0e6 100644
--- a/README.md
+++ b/README.md
@@ -62,18 +62,22 @@ Hudi integration in the data ecosystems for a diverse range of users and project
> These examples expect a Hudi table exists at `/tmp/trips_table`, created using
> the [quick start guide](https://hudi.apache.org/docs/quick-start-guide).
+For the full reader API reference (`ReadOptions`, filter expressions, behavioral guarantees), see [docs/reader-spec.md](docs/reader-spec.md).
+
### Snapshot Query
-Snapshot query reads the latest version of the data from the table. The table API also accepts partition filters.
+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
```python
-from hudi import HudiTableBuilder
+from hudi import HudiReadOptions, HudiTableBuilder
import pyarrow as pa
hudi_table = HudiTableBuilder.from_base_uri("/tmp/trips_table").build()
-batches = hudi_table.read_snapshot(filters=[("city", "=", "san_francisco")])
+batches = hudi_table.read(
+ HudiReadOptions(filters=[("city", "=", "san_francisco")])
+)
# convert to PyArrow table
arrow_table = pa.Table.from_batches(batches)
@@ -85,13 +89,15 @@ 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 batches = hudi_table.read_snapshot(&[("city", "=", "san_francisco")]).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 {
@@ -130,23 +136,24 @@ let hudi_table =
### Time-Travel Query
-Time-travel query reads the data at a specific timestamp from the table. The table API also accepts partition filters.
+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
```python
-batches = (
- hudi_table
- .read_snapshot_as_of("20241231123456789", filters=[("city", "=", "san_francisco")])
+batches = hudi_table.read(
+ HudiReadOptions(filters=[("city", "=", "san_francisco")])
+ .with_as_of_timestamp("20241231123456789")
)
```
#### Rust
```rust
-let batches =
- hudi_table
- .read_snapshot_as_of("20241231123456789", &[("city", "=", "san_francisco")]).await?;
+let options = ReadOptions::new()
+ .with_as_of_timestamp("20241231123456789")
+ .with_filters([("city", "=", "san_francisco")])?;
+let batches = hudi_table.read(&options).await?;
```
<details>
@@ -155,14 +162,13 @@ let batches =
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.
-- ISO 8601 format including but not limited to:
+- 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.SSS`
- `yyyy-MM-dd'T'HH:mm:ss+00:00`
- `yyyy-MM-dd'T'HH:mm:ssZ`
- - `yyyy-MM-dd'T'HH:mm:ss`
- - `yyyy-MM-dd`
+
+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.
</details>
### Incremental Query
@@ -172,25 +178,96 @@ Incremental query reads the changed data from the table for a given time range.
#### Python
```python
+from hudi import HudiQueryType
+
# read the records between t1 (exclusive) and t2 (inclusive)
-batches = hudi_table.read_incremental_records(t1, t2)
+batches = hudi_table.read(
+ HudiReadOptions()
+ .with_query_type(HudiQueryType.Incremental)
+ .with_start_timestamp(t1)
+ .with_end_timestamp(t2)
+)
+
+# read the records after t1 (end defaults to the latest commit)
+batches = hudi_table.read(
+ HudiReadOptions()
+ .with_query_type(HudiQueryType.Incremental)
+ .with_start_timestamp(t1)
+)
-# read the records after t1
-batches = hudi_table.read_incremental_records(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
```rust
-// read the records between t1 (exclusive) and t2 (inclusive)
-let batches = hudi_table.read_incremental_records(t1, Some(t2)).await?;
+use hudi::table::QueryType;
-// read the records after t1
-let batches = hudi_table.read_incremental_records(t1, None).await?;
+// read the records between t1 (exclusive) and t2 (inclusive)
+let options = ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(t1)
+ .with_end_timestamp(t2);
+let batches = hudi_table.read(&options).await?;
+
+// read the records after t1 (end defaults to the latest commit)
+let options = ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(t1);
+let batches = hudi_table.read(&options).await?;
+
+// with column filters applied to the changed records
+let options = ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(t1)
+ .with_end_timestamp(t2)
+ .with_filters([("city", "=", "san_francisco")])?;
+let batches = hudi_table.read(&options).await?;
```
*Incremental queries support the same timestamp formats as time-travel queries.*
+### Streaming Read
+
+Streaming reads yield `RecordBatch`es one at a time without loading the full result into memory.
+The same `ReadOptions` knobs apply, plus `batch_size` and `projection`.
+
+#### Python
+
+```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
+
+```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
@@ -202,23 +279,30 @@ engines, where the plan provides file paths.
from hudi import HudiFileGroupReader
reader = HudiFileGroupReader(
- "/table/base/path", {"hoodie.read.file_group.start_timestamp": "0"})
+ "/table/base/path", {"hoodie.read.start.timestamp": "0"})
# Returns a PyArrow RecordBatch
-record_batch = reader.read_file_slice_by_base_file_path("relative/path.parquet")
+record_batch = reader.read_file_slice_from_paths("relative/path.parquet", [])
```
#### Rust
```rust,ignore
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.file_group.start_timestamp", "0")]).await?;
+ "/table/base/path", [("hoodie.read.start.timestamp", "0")]).await?;
// Returns an Arrow RecordBatch
-let record_batch = reader.read_file_slice_by_base_file_path("relative/path.parquet").await?;
+let record_batch = reader
+ .read_file_slice_from_paths(
+ "relative/path.parquet",
+ Vec::<&str>::new(),
+ &ReadOptions::new(),
+ )
+ .await?;
```
#### C++
@@ -230,10 +314,11 @@ let record_batch = reader.read_file_slice_by_base_file_path("relative/path.parqu
// Functions may throw rust::Error on failure
auto reader = new_file_group_reader_with_options(
- "/table/base/path", {"hoodie.read.file_group.start_timestamp=0"});
+ "/table/base/path", {"hoodie.read.start.timestamp=0"});
// Returns an ArrowArrayStream pointer
-ArrowArrayStream* stream_ptr = reader->read_file_slice_by_base_file_path("relative/path.parquet");
+std::vector<std::string> log_file_paths{};
+ArrowArrayStream* stream_ptr = reader->read_file_slice_from_paths("relative/path.parquet", log_file_paths);
```
## Query Engine Integration
@@ -244,23 +329,33 @@ Hudi-rs provides APIs to support integration with query engines. The sections be
Create a Hudi table instance using its constructor or the `TableBuilder` API.
-| Stage | API | Description |
-|-----------------|-------------------------------------------|--------------------------------------------------------------------------------|
-| Query planning | `get_file_slices()` | For snapshot query, get a list of file slices. |
-| | `get_file_slices_splits()` | For snapshot query, get a list of file slices in splits. |
-| | `get_file_slices_as_of()` | For time-travel query, get a list of file slices at a given time. |
-| | `get_file_slices_splits_as_of()` | For time-travel query, get a list of file slices in splits at a given time. |
-| | `get_file_slices_between()` | For incremental query, get a list of changed file slices between a time range. |
-| Query execution | `create_file_group_reader_with_options()` | Create a file group reader instance with the table instance's configs. |
+All read APIs accept a `ReadOptions` (Rust) / `HudiReadOptions` (Python) value. It stores three fields β `filters`, `projection`, and `hudi_options` β and exposes chainable `with_*` builders for the rest. The available knobs:
+
+- `query_type` (`with_query_type`) β `Snapshot` (default) or `Incremental`. Drives dispatch in `read`, `read_stream`, and `get_file_slices`.
+- `filters` β column filters as `(field, op, value)` tuples. The field can be any column (partition or data). Used for partition pruning, file-level stats pruning (snapshot only), and row-level filtering.
+- `projection` β columns to return. Streaming pushes the projection down to the parquet reader; eager reads project after merging.
+- `batch_size` (`with_batch_size`) β rows per batch (streaming only; eager reads return one batch per file slice).
+- `as_of_timestamp` (`with_as_of_timestamp`) β snapshot/time-travel timestamp (defaults to latest commit).
+- `start_timestamp` / `end_timestamp` (`with_start_timestamp` / `with_end_timestamp`) β incremental range (defaults to earliestβ¦latest).
+- `hudi_options` β per-read Hudi configs that override table-level defaults (e.g. `hoodie.read.use.read_optimized.mode`).
+
+| 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()` | Estimated `(num_rows, byte_size)` for scan planning. Returns `None` when the metadata table is disabled. |
+| Query execution | `create_file_group_reader_with_options(read_options, extras)` | Create a file group reader with the table's configs. Both args are optional; `read_options.hudi_options` overrides table-level defaults (excluding the four `Table`-owned read keys), and `extras` always wins. |
+| | `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`. |
### 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_by_base_file_path()` | Read records from a given base file path; log files will be ignored |
+| 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()`. Yields true streaming batches when the slice is base-file-only or read-optimized; for MOR slices with log files, falls back to a single merged batch. |
+| | `read_file_slice_from_paths_stream()` | Streaming version of `read_file_slice_from_paths()`. |
### Apache DataFusion
diff --git a/cpp/src/lib.rs b/cpp/src/lib.rs
index eddf6776..08c14658 100644
--- a/cpp/src/lib.rs
+++ b/cpp/src/lib.rs
@@ -23,6 +23,7 @@ use cxx::{CxxString, CxxVector};
use hudi::file_group::FileGroup;
use hudi::file_group::file_slice::FileSlice;
use hudi::file_group::reader::FileGroupReader;
+use hudi::table::ReadOptions;
#[cxx::bridge]
mod ffi {
@@ -46,9 +47,10 @@ mod ffi {
log_file_names: &CxxVector<CxxString>,
) -> Result<Box<HudiFileSlice>>;
- fn read_file_slice_by_base_file_path(
+ fn read_file_slice_from_paths(
self: &HudiFileGroupReader,
- relative_path: &CxxString,
+ base_file_path: &CxxString,
+ log_file_paths: &CxxVector<CxxString>,
) -> Result<*mut ArrowArrayStream>;
fn read_file_slice(
@@ -92,17 +94,31 @@ pub fn new_file_group_reader_with_options(
}
impl HudiFileGroupReader {
- pub fn read_file_slice_by_base_file_path(
+ pub fn read_file_slice_from_paths(
&self,
- relative_path: &CxxString,
+ base_file_path: &CxxString,
+ log_file_paths: &CxxVector<CxxString>,
) -> std::result::Result<*mut ffi::ArrowArrayStream, String> {
- let relative_path = relative_path
+ let base_file_path = base_file_path
.to_str()
.map_err(|e| format!("Failed to convert CxxString to str: {e}"))?;
+ let log_file_paths = log_file_paths
+ .iter()
+ .map(|p| {
+ p.to_str()
+ .map(String::from)
+ .map_err(|e| format!("Failed to convert CxxString to str: {e}"))
+ })
+ .collect::<std::result::Result<Vec<_>, _>>()?;
+
let record_batch = self
.rt
- .block_on(self.inner.read_file_slice_by_base_file_path(relative_path))
+ .block_on(self.inner.read_file_slice_from_paths(
+ base_file_path,
+ log_file_paths,
+ &ReadOptions::new(),
+ ))
.map_err(|e| format!("Failed to read file batch: {e}"))?;
let schema = record_batch.schema();
@@ -118,7 +134,10 @@ impl HudiFileGroupReader {
) -> std::result::Result<*mut ffi::ArrowArrayStream, String> {
let record_batch = self
.rt
- .block_on(self.inner.read_file_slice(&file_slice.inner))
+ .block_on(
+ self.inner
+ .read_file_slice(&file_slice.inner, &ReadOptions::new()),
+ )
.map_err(|e| format!("Failed to read file slice: {e}"))?;
let schema = record_batch.schema();
diff --git a/crates/core/src/config/read.rs b/crates/core/src/config/read.rs
index 823bcfd7..4f8ea787 100644
--- a/crates/core/src/config/read.rs
+++ b/crates/core/src/config/read.rs
@@ -22,12 +22,45 @@ use std::collections::HashMap;
use std::fmt::Display;
use std::str::FromStr;
-use strum_macros::{EnumIter, IntoStaticStr};
+use strum_macros::{AsRefStr, EnumIter, IntoStaticStr};
use crate::config::Result;
-use crate::config::error::ConfigError::{NotFound, ParseBool, ParseInt};
+use crate::config::error::ConfigError;
+use crate::config::error::ConfigError::{InvalidValue, NotFound, ParseBool, ParseInt};
use crate::config::{ConfigParser, HudiConfigValue};
+/// Config value for [`HudiReadConfig::QueryType`]. Canonical strings are
+/// `snapshot` and `incremental`; [`FromStr`] accepts case-insensitive forms.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, AsRefStr)]
+pub enum QueryType {
+ /// Latest table state at one commit (the latest by default; an explicit
+ /// `as_of_timestamp` for time-travel).
+ #[default]
+ #[strum(serialize = "snapshot")]
+ Snapshot,
+ /// Records changed in the half-open range (`start_timestamp`, `end_timestamp`].
+ #[strum(serialize = "incremental")]
+ Incremental,
+}
+
+impl Display for QueryType {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.write_str(self.as_ref())
+ }
+}
+
+impl FromStr for QueryType {
+ type Err = ConfigError;
+
+ fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
+ match s.to_ascii_lowercase().as_str() {
+ "snapshot" => Ok(Self::Snapshot),
+ "incremental" => Ok(Self::Incremental),
+ v => Err(InvalidValue(v.to_string())),
+ }
+ }
+}
+
/// Configurations for reading Hudi tables.
///
/// **Example**
@@ -46,11 +79,18 @@ use crate::config::{ConfigParser, HudiConfigValue};
#[derive(Clone, Debug, PartialEq, Eq, Hash, EnumIter, IntoStaticStr)]
pub enum HudiReadConfig {
- /// Start timestamp (exclusive) for [FileGroup] to filter records.
- FileGroupStartTimestamp,
+ /// Selects the read semantic. Accepted values: `snapshot` (default), `incremental`.
+ /// See [`crate::table::QueryType`].
+ QueryType,
+
+ /// Snapshot/time-travel timestamp. Reads return the table state at this commit.
+ AsOfTimestamp,
- /// End timestamp (inclusive) for [FileGroup] to filter records.
- FileGroupEndTimestamp,
+ /// Start timestamp (exclusive) used by file-group readers to filter records.
+ StartTimestamp,
+
+ /// End timestamp (inclusive) used by file-group readers to filter records.
+ EndTimestamp,
/// Number of input partitions to read the data in parallel.
///
@@ -60,7 +100,7 @@ pub enum HudiReadConfig {
/// Parallelism for listing files on storage.
ListingParallelism,
- /// When set to true, only [BaseFile]s will be read for optimized reads.
+ /// When set to true, only base files will be read for optimized reads.
/// This is only applicable to Merge-On-Read (MOR) tables.
UseReadOptimizedMode,
@@ -69,11 +109,16 @@ pub enum HudiReadConfig {
StreamBatchSize,
}
-impl AsRef<str> for HudiReadConfig {
- fn as_ref(&self) -> &str {
+impl HudiReadConfig {
+ /// `&'static str` form of the config key. `const fn` so callers can use it in
+ /// `const` contexts (e.g. building static lookup tables of `hoodie.*` keys
+ /// without duplicating the literal strings).
+ pub const fn key_str(&self) -> &'static str {
match self {
- Self::FileGroupStartTimestamp => "hoodie.read.file_group.start_timestamp",
- Self::FileGroupEndTimestamp => "hoodie.read.file_group.end_timestamp",
+ Self::QueryType => "hoodie.read.query.type",
+ Self::AsOfTimestamp => "hoodie.read.as.of.timestamp",
+ Self::StartTimestamp => "hoodie.read.start.timestamp",
+ Self::EndTimestamp => "hoodie.read.end.timestamp",
Self::InputPartitions => "hoodie.read.input.partitions",
Self::ListingParallelism => "hoodie.read.listing.parallelism",
Self::UseReadOptimizedMode => "hoodie.read.use.read_optimized.mode",
@@ -82,6 +127,12 @@ impl AsRef<str> for HudiReadConfig {
}
}
+impl AsRef<str> for HudiReadConfig {
+ fn as_ref(&self) -> &str {
+ self.key_str()
+ }
+}
+
impl Display for HudiReadConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_ref())
@@ -93,6 +144,9 @@ impl ConfigParser for HudiReadConfig {
fn default_value(&self) -> Option<HudiConfigValue> {
match self {
+ HudiReadConfig::QueryType => Some(HudiConfigValue::String(
+ QueryType::default().as_ref().to_string(),
+ )),
HudiReadConfig::InputPartitions => Some(HudiConfigValue::UInteger(0usize)),
HudiReadConfig::ListingParallelism => Some(HudiConfigValue::UInteger(10usize)),
HudiReadConfig::UseReadOptimizedMode => Some(HudiConfigValue::Boolean(false)),
@@ -108,12 +162,12 @@ impl ConfigParser for HudiReadConfig {
.ok_or(NotFound(self.key()));
match self {
- Self::FileGroupStartTimestamp => {
- get_result.map(|v| HudiConfigValue::String(v.to_string()))
- }
- Self::FileGroupEndTimestamp => {
- get_result.map(|v| HudiConfigValue::String(v.to_string()))
- }
+ Self::QueryType => get_result
+ .and_then(QueryType::from_str)
+ .map(|v| HudiConfigValue::String(v.as_ref().to_string())),
+ Self::AsOfTimestamp => get_result.map(|v| HudiConfigValue::String(v.to_string())),
+ Self::StartTimestamp => get_result.map(|v| HudiConfigValue::String(v.to_string())),
+ Self::EndTimestamp => get_result.map(|v| HudiConfigValue::String(v.to_string())),
Self::InputPartitions => get_result
.and_then(|v| {
usize::from_str(v).map_err(|e| ParseInt(self.key(), v.to_string(), e))
@@ -131,7 +185,13 @@ impl ConfigParser for HudiReadConfig {
.map(HudiConfigValue::Boolean),
Self::StreamBatchSize => get_result
.and_then(|v| {
- usize::from_str(v).map_err(|e| ParseInt(self.key(), v.to_string(), e))
+ let key = self.key();
+ let parsed =
+ usize::from_str(v).map_err(|e| ParseInt(key.clone(), v.to_string(), e))?;
+ if parsed == 0 {
+ return Err(InvalidValue(format!("{key}=0 (must be > 0)")));
+ }
+ Ok(parsed)
})
.map(HudiConfigValue::UInteger),
}
@@ -142,34 +202,58 @@ impl ConfigParser for HudiReadConfig {
mod tests {
use super::*;
use crate::config::read::HudiReadConfig::{
- InputPartitions, ListingParallelism, UseReadOptimizedMode,
+ AsOfTimestamp, EndTimestamp, InputPartitions, ListingParallelism,
+ QueryType as QueryTypeKey, StartTimestamp, StreamBatchSize, UseReadOptimizedMode,
};
#[test]
fn parse_valid_config_value() {
let options = HashMap::from([
+ (QueryTypeKey.as_ref().to_string(), "Incremental".to_string()),
+ (AsOfTimestamp.as_ref().to_string(), "20240101".to_string()),
+ (StartTimestamp.as_ref().to_string(), "20240102".to_string()),
+ (EndTimestamp.as_ref().to_string(), "20240103".to_string()),
(InputPartitions.as_ref().to_string(), "100".to_string()),
(ListingParallelism.as_ref().to_string(), "100".to_string()),
(
UseReadOptimizedMode.as_ref().to_string(),
"true".to_string(),
),
+ (StreamBatchSize.as_ref().to_string(), "2048".to_string()),
]);
+ let actual: String = QueryTypeKey.parse_value(&options).unwrap().into();
+ assert_eq!(actual, "incremental");
+ let actual: String = AsOfTimestamp.parse_value(&options).unwrap().into();
+ assert_eq!(actual, "20240101");
+ let actual: String = StartTimestamp.parse_value(&options).unwrap().into();
+ assert_eq!(actual, "20240102");
+ let actual: String = EndTimestamp.parse_value(&options).unwrap().into();
+ assert_eq!(actual, "20240103");
let actual: usize = InputPartitions.parse_value(&options).unwrap().into();
assert_eq!(actual, 100);
let actual: usize = ListingParallelism.parse_value(&options).unwrap().into();
assert_eq!(actual, 100);
let actual: bool = UseReadOptimizedMode.parse_value(&options).unwrap().into();
assert!(actual);
+ let actual: usize = StreamBatchSize.parse_value(&options).unwrap().into();
+ assert_eq!(actual, 2048);
}
#[test]
fn parse_invalid_config_value() {
let options = HashMap::from([
+ (QueryTypeKey.as_ref().to_string(), "bogus".to_string()),
(InputPartitions.as_ref().to_string(), "foo".to_string()),
(ListingParallelism.as_ref().to_string(), "_100".to_string()),
(UseReadOptimizedMode.as_ref().to_string(), "1".to_string()),
+ (StreamBatchSize.as_ref().to_string(), "abc".to_string()),
]);
+ assert!(matches!(
+ QueryTypeKey.parse_value(&options).unwrap_err(),
+ InvalidValue(_)
+ ));
+ let actual: String = QueryTypeKey.parse_value_or_default(&options).into();
+ assert_eq!(actual, "snapshot");
assert!(matches!(
InputPartitions.parse_value(&options).unwrap_err(),
ParseInt(_, _, _)
@@ -187,6 +271,53 @@ mod tests {
ParseBool(_, _, _)
));
let actual: bool = UseReadOptimizedMode.parse_value_or_default(&options).into();
- assert!(!actual)
+ assert!(!actual);
+ assert!(matches!(
+ StreamBatchSize.parse_value(&options).unwrap_err(),
+ ParseInt(_, _, _)
+ ));
+ let actual: usize = StreamBatchSize.parse_value_or_default(&options).into();
+ assert_eq!(actual, 1024);
+ }
+
+ #[test]
+ fn timestamp_keys_have_no_default_value() {
+ assert!(AsOfTimestamp.default_value().is_none());
+ assert!(StartTimestamp.default_value().is_none());
+ assert!(EndTimestamp.default_value().is_none());
+ }
+
+ #[test]
+ fn query_type_from_str_accepts_case_insensitive_and_rejects_invalid() {
+ assert_eq!(
+ QueryType::from_str("snapshot").unwrap(),
+ QueryType::Snapshot
+ );
+ assert_eq!(
+ QueryType::from_str("SNAPSHOT").unwrap(),
+ QueryType::Snapshot
+ );
+ assert_eq!(
+ QueryType::from_str("Incremental").unwrap(),
+ QueryType::Incremental
+ );
+ assert!(matches!(
+ QueryType::from_str("bogus").unwrap_err(),
+ InvalidValue(_)
+ ));
+ }
+
+ #[test]
+ fn display_impls_match_canonical_keys() {
+ assert_eq!(format!("{}", QueryType::Snapshot), "snapshot");
+ assert_eq!(format!("{}", QueryType::Incremental), "incremental");
+ assert_eq!(
+ format!("{}", HudiReadConfig::StreamBatchSize),
+ "hoodie.read.stream.batch_size"
+ );
+ assert_eq!(
+ format!("{}", HudiReadConfig::QueryType),
+ "hoodie.read.query.type"
+ );
}
}
diff --git a/crates/core/src/expr/filter.rs b/crates/core/src/expr/filter.rs
index 791a5564..506305e0 100644
--- a/crates/core/src/expr/filter.rs
+++ b/crates/core/src/expr/filter.rs
@@ -29,27 +29,27 @@ use std::str::FromStr;
#[derive(Debug, Clone)]
pub struct Filter {
- pub field_name: String,
+ pub field: String,
pub operator: ExprOperator,
pub values: Vec<String>,
}
impl Filter {
- pub fn new(field_name: String, operator: ExprOperator, values: Vec<String>) -> Result<Self> {
+ pub fn new(field: String, operator: ExprOperator, values: Vec<String>) -> Result<Self> {
if operator.is_multi_value() {
if values.is_empty() {
return Err(CoreError::Schema(format!(
- "{operator} operator requires at least one value for field '{field_name}'"
+ "{operator} operator requires at least one value for field '{field}'"
)));
}
} else if values.len() != 1 {
return Err(CoreError::Schema(format!(
- "Operator {operator} requires exactly one value for field '{field_name}', got {}",
+ "Operator {operator} requires exactly one value for field '{field}', got {}",
values.len()
)));
}
Ok(Self {
- field_name,
+ field,
operator,
values,
})
@@ -67,8 +67,17 @@ impl Filter {
impl From<Filter> for (String, String, String) {
fn from(filter: Filter) -> Self {
- let value_str = filter.values.join(",");
- (filter.field_name, filter.operator.to_string(), value_str)
+ let value_str = if filter.operator.is_multi_value() {
+ filter
+ .values
+ .iter()
+ .map(|v| escape_in_value(v))
+ .collect::<Vec<_>>()
+ .join(",")
+ } else {
+ filter.values.first().cloned().unwrap_or_default()
+ };
+ (filter.field, filter.operator.to_string(), value_str)
}
}
@@ -76,20 +85,72 @@ impl TryFrom<(&str, &str, &str)> for Filter {
type Error = CoreError;
fn try_from(binary_expr_tuple: (&str, &str, &str)) -> Result<Self, Self::Error> {
- let (field_name, operator_str, field_value) = binary_expr_tuple;
- let field_name = field_name.to_string();
+ let (field, operator_str, field_value) = binary_expr_tuple;
+ let field = field.to_string();
let operator = ExprOperator::from_str(operator_str)?;
let values = if operator.is_multi_value() {
- field_value
- .split(',')
- .map(|s| s.trim().to_string())
- .filter(|s| !s.is_empty())
- .collect()
+ split_in_values(field_value)
} else {
vec![field_value.to_string()]
};
- Filter::new(field_name, operator, values)
+ Filter::new(field, operator, values)
+ }
+}
+
+/// Split a multi-value (`IN` / `NOT IN`) value string on unescaped commas, then
+/// trim each segment. `\,` is a literal comma; `\\` is a literal backslash; any
+/// other backslash is preserved as-is so callers that don't use escapes are not
+/// surprised. Empty segments after trimming are dropped β the cardinality check
+/// in [`Filter::new`] enforces at least one value.
+///
+/// Symmetric with [`escape_in_value`] used by `From<Filter> for (String, String, String)`,
+/// so a round-trip preserves values that contain commas or backslashes.
+fn split_in_values(s: &str) -> Vec<String> {
+ let mut out: Vec<String> = Vec::new();
+ let mut current = String::new();
+ let mut chars = s.chars().peekable();
+ while let Some(c) = chars.next() {
+ if c == '\\' {
+ match chars.peek() {
+ Some(&',') => {
+ chars.next();
+ current.push(',');
+ }
+ Some(&'\\') => {
+ chars.next();
+ current.push('\\');
+ }
+ _ => current.push('\\'),
+ }
+ } else if c == ',' {
+ let trimmed = current.trim().to_string();
+ if !trimmed.is_empty() {
+ out.push(trimmed);
+ }
+ current.clear();
+ } else {
+ current.push(c);
+ }
+ }
+ let trimmed = current.trim().to_string();
+ if !trimmed.is_empty() {
+ out.push(trimmed);
+ }
+ out
+}
+
+/// Escape backslash and comma so a value can survive a round trip through the
+/// `(String, String, String)` tuple form used for `IN` / `NOT IN`.
+fn escape_in_value(v: &str) -> String {
+ let mut out = String::with_capacity(v.len());
+ for c in v.chars() {
+ match c {
+ '\\' => out.push_str("\\\\"),
+ ',' => out.push_str("\\,"),
+ other => out.push(other),
+ }
}
+ out
}
pub fn from_str_tuples<I, S>(tuples: I) -> Result<Vec<Filter>>
@@ -103,6 +164,63 @@ where
.collect()
}
+/// Evaluate a slice of [`Filter`]s against a [`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
+/// skipped silently. Callers that want strict-on-unknown-column behavior must call
+/// [`validate_fields_against_schemas`] before invoking this function (this is what
+/// the file-group reader paths do).
+///
+/// All applicable filters are evaluated as row-level predicates and ANDed together.
+pub fn filters_to_row_mask(
+ filters: &[Filter],
+ batch: &arrow_array::RecordBatch,
+) -> Result<BooleanArray> {
+ let num_rows = batch.num_rows();
+ let mut mask: Option<BooleanArray> = None;
+ for filter in filters {
+ let Some(column) = batch.column_by_name(&filter.field) else {
+ continue;
+ };
+ let schemable = SchemableFilter::try_from((filter.clone(), batch.schema().as_ref()))?;
+ let column_mask = schemable.apply_comparison(column)?;
+ mask = Some(match mask {
+ Some(prev) => boolean::and(&prev, &column_mask)?,
+ None => column_mask,
+ });
+ }
+ Ok(mask.unwrap_or_else(|| BooleanArray::from(vec![true; num_rows])))
+}
+
+/// Error if any [`Filter`] targets a column not present in any of the provided schemas.
+///
+/// This is the strict counterpart to [`filters_to_row_mask`], which silently skips
+/// missing columns. Pass one schema for the file-group-reader case (validate against
+/// the read batch). Pass multiple β e.g. data schema + partition schema β at the
+/// table layer, where partition columns may be valid filter targets even when not
+/// physically present in the data schema. Failing fast prevents typos like
+/// `("rder_id", "=", "x")` from becoming silent no-ops.
+pub fn validate_fields_against_schemas<'a, I>(filters: &[Filter], schemas: I) -> Result<()>
+where
+ I: IntoIterator<Item = &'a Schema>,
+{
+ use std::collections::HashSet;
+ let mut valid: HashSet<&str> = HashSet::new();
+ for schema in schemas {
+ valid.extend(schema.fields().iter().map(|f| f.name().as_str()));
+ }
+ for filter in filters {
+ if !valid.contains(filter.field.as_str()) {
+ return Err(CoreError::Schema(format!(
+ "Filter field '{}' not found in schema",
+ filter.field
+ )));
+ }
+ }
+ Ok(())
+}
+
pub struct FilterField {
pub name: String,
}
@@ -118,7 +236,7 @@ impl FilterField {
pub fn eq(&self, value: impl Into<String>) -> Filter {
Filter {
- field_name: self.name.clone(),
+ field: self.name.clone(),
operator: ExprOperator::Eq,
values: vec![value.into()],
}
@@ -126,7 +244,7 @@ impl FilterField {
pub fn ne(&self, value: impl Into<String>) -> Filter {
Filter {
- field_name: self.name.clone(),
+ field: self.name.clone(),
operator: ExprOperator::Ne,
values: vec![value.into()],
}
@@ -134,7 +252,7 @@ impl FilterField {
pub fn lt(&self, value: impl Into<String>) -> Filter {
Filter {
- field_name: self.name.clone(),
+ field: self.name.clone(),
operator: ExprOperator::Lt,
values: vec![value.into()],
}
@@ -142,7 +260,7 @@ impl FilterField {
pub fn lte(&self, value: impl Into<String>) -> Filter {
Filter {
- field_name: self.name.clone(),
+ field: self.name.clone(),
operator: ExprOperator::Lte,
values: vec![value.into()],
}
@@ -150,7 +268,7 @@ impl FilterField {
pub fn gt(&self, value: impl Into<String>) -> Filter {
Filter {
- field_name: self.name.clone(),
+ field: self.name.clone(),
operator: ExprOperator::Gt,
values: vec![value.into()],
}
@@ -158,7 +276,7 @@ impl FilterField {
pub fn gte(&self, value: impl Into<String>) -> Filter {
Filter {
- field_name: self.name.clone(),
+ field: self.name.clone(),
operator: ExprOperator::Gte,
values: vec![value.into()],
}
@@ -170,7 +288,7 @@ impl FilterField {
S: Into<String>,
{
Filter {
- field_name: self.name.clone(),
+ field: self.name.clone(),
operator: ExprOperator::In,
values: values.into_iter().map(|v| v.into()).collect(),
}
@@ -182,7 +300,7 @@ impl FilterField {
S: Into<String>,
{
Filter {
- field_name: self.name.clone(),
+ field: self.name.clone(),
operator: ExprOperator::NotIn,
values: values.into_iter().map(|v| v.into()).collect(),
}
@@ -204,25 +322,21 @@ impl TryFrom<(Filter, &Schema)> for SchemableFilter {
type Error = CoreError;
fn try_from((filter, schema): (Filter, &Schema)) -> Result<Self, Self::Error> {
- let field_name = filter.field_name.clone();
- let field: &Field = schema.field_with_name(&field_name).map_err(|e| {
+ let field_name = filter.field.as_str();
+ let arrow_field = schema.field_with_name(field_name).map_err(|e| {
CoreError::Schema(format!("Field {field_name} not found in schema: {e:?}"))
})?;
- let operator = filter.operator;
-
let values: Result<Vec<_>> = filter
.values
.iter()
- .map(|v| Self::cast_value(&[v.as_str()], field.data_type()))
+ .map(|v| Self::cast_value(&[v.as_str()], arrow_field.data_type()))
.collect();
- let values = values?;
- let field = field.clone();
Ok(SchemableFilter {
- field,
- operator,
- values,
+ field: arrow_field.clone(),
+ operator: filter.operator,
+ values: values?,
})
}
}
@@ -276,8 +390,9 @@ impl SchemableFilter {
#[cfg(test)]
mod tests {
use super::*;
- use arrow_array::{Int64Array, StringArray};
+ use arrow_array::{ArrayRef, Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
+ use std::sync::Arc;
fn create_test_schema() -> Schema {
Schema::new(vec![
@@ -287,41 +402,129 @@ mod tests {
}
#[test]
- fn test_schemable_filter_try_from() -> Result<()> {
+ fn test_filters_to_row_mask_combines_filters_with_and() -> Result<()> {
+ let schema = Arc::new(create_test_schema());
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(vec!["a", "b", "a"])) as ArrayRef,
+ Arc::new(Int64Array::from(vec![1, 2, 3])) as ArrayRef,
+ ],
+ )
+ .unwrap();
+ let filters = vec![
+ Filter::try_from(("string_col", "=", "a"))?,
+ Filter::try_from(("int_col", ">", "1"))?,
+ ];
+ let mask = filters_to_row_mask(&filters, &batch)?;
+ assert_eq!(mask, BooleanArray::from(vec![false, false, true]));
+ Ok(())
+ }
+
+ #[test]
+ fn test_filters_to_row_mask_skips_missing_columns() -> Result<()> {
+ let schema = Arc::new(create_test_schema());
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(vec!["a", "b"])) as ArrayRef,
+ Arc::new(Int64Array::from(vec![1, 2])) as ArrayRef,
+ ],
+ )
+ .unwrap();
+ let filters = vec![
+ Filter::try_from(("not_in_batch", "=", "x"))?,
+ Filter::try_from(("string_col", "=", "a"))?,
+ ];
+ let mask = filters_to_row_mask(&filters, &batch)?;
+ assert_eq!(mask, BooleanArray::from(vec![true, false]));
+ Ok(())
+ }
+
+ #[test]
+ fn test_validate_fields_against_schemas() -> Result<()> {
let schema = create_test_schema();
- // Test string column filter creation
- let string_filter = Filter {
- field_name: "string_col".to_string(),
- operator: ExprOperator::Eq,
- values: vec!["test_value".to_string()],
- };
+ // Empty filter list passes (single-schema form).
+ assert!(validate_fields_against_schemas(&[], [&schema]).is_ok());
- let schemable = SchemableFilter::try_from((string_filter, &schema))?;
- assert_eq!(schemable.field.name(), "string_col");
- assert_eq!(schemable.field.data_type(), &DataType::Utf8);
- assert_eq!(schemable.operator, ExprOperator::Eq);
+ // All-known fields pass.
+ let valid = vec![
+ Filter::try_from(("string_col", "=", "x"))?,
+ Filter::try_from(("int_col", ">", "1"))?,
+ ];
+ assert!(validate_fields_against_schemas(&valid, [&schema]).is_ok());
- // Test integer column filter creation
- let int_filter = Filter {
- field_name: "int_col".to_string(),
- operator: ExprOperator::Gt,
- values: vec!["42".to_string()],
- };
+ // Unknown field errors with a Schema error naming the bad column.
+ let invalid = vec![
+ Filter::try_from(("string_col", "=", "x"))?,
+ Filter::try_from(("typo_col", "=", "y"))?,
+ ];
+ let err = validate_fields_against_schemas(&invalid, [&schema]).unwrap_err();
+ assert!(matches!(err, CoreError::Schema(_)));
+ assert!(err.to_string().contains("typo_col"));
+
+ // Multi-schema: a filter on a column present in only one of the schemas
+ // passes when the union is considered. This mirrors the Table-layer use case
+ // where partition fields may not be in the data schema but are still valid
+ // filter targets.
+ let partition_schema = Schema::new(vec![Field::new("city", DataType::Utf8, false)]);
+ let cross_schema_filters = vec![
+ Filter::try_from(("string_col", "=", "x"))?, // in data schema only
+ Filter::try_from(("city", "=", "sf"))?, // in partition schema only
+ ];
+ assert!(
+ validate_fields_against_schemas(&cross_schema_filters, [&schema, &partition_schema])
+ .is_ok()
+ );
- let schemable = SchemableFilter::try_from((int_filter, &schema))?;
- assert_eq!(schemable.field.name(), "int_col");
- assert_eq!(schemable.field.data_type(), &DataType::Int64);
- assert_eq!(schemable.operator, ExprOperator::Gt);
+ // A filter on a column missing from BOTH schemas still errors.
+ let still_invalid = vec![Filter::try_from(("nope", "=", "x"))?];
+ let err = validate_fields_against_schemas(&still_invalid, [&schema, &partition_schema])
+ .unwrap_err();
+ assert!(err.to_string().contains("nope"));
- // Test error case - non-existent column
- let invalid_filter = Filter {
- field_name: "non_existent".to_string(),
- operator: ExprOperator::Eq,
- values: vec!["value".to_string()],
- };
+ Ok(())
+ }
+
+ #[test]
+ fn test_filters_to_row_mask_empty_returns_all_true() -> Result<()> {
+ let schema = Arc::new(create_test_schema());
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(vec!["a", "b"])) as ArrayRef,
+ Arc::new(Int64Array::from(vec![1, 2])) as ArrayRef,
+ ],
+ )
+ .unwrap();
+ let mask = filters_to_row_mask(&[], &batch)?;
+ assert_eq!(mask, BooleanArray::from(vec![true, true]));
+ Ok(())
+ }
+
+ #[test]
+ fn test_schemable_filter_try_from_schema_lookup() -> Result<()> {
+ let schema = create_test_schema();
+
+ // Happy path: looks up the field, copies operator, casts values.
+ let filter = Filter::new(
+ "string_col".to_string(),
+ ExprOperator::Eq,
+ vec!["test_value".to_string()],
+ )?;
+ let schemable = SchemableFilter::try_from((filter, &schema))?;
+ assert_eq!(schemable.field.name(), "string_col");
+ assert_eq!(schemable.field.data_type(), &DataType::Utf8);
+ assert_eq!(schemable.operator, ExprOperator::Eq);
- assert!(SchemableFilter::try_from((invalid_filter, &schema)).is_err());
+ // Error path: unknown column.
+ let invalid = Filter::new(
+ "non_existent".to_string(),
+ ExprOperator::Eq,
+ vec!["value".to_string()],
+ )?;
+ assert!(SchemableFilter::try_from((invalid, &schema)).is_err());
Ok(())
}
@@ -369,126 +572,115 @@ mod tests {
}
#[test]
- fn test_schemable_filter_apply_comparison() -> Result<()> {
+ fn test_schemable_filter_apply_comparison_all_operators() -> Result<()> {
+ // Single parameterized test covering all 8 operators across both a string
+ // column ("a", "b", "a") and an int column (40, 50, 60). Verifies the
+ // dispatch in `apply_comparison` and that the per-operator kernels are
+ // wired correctly. Single-value ops use `Vec::from([v])`; set ops use
+ // multi-value vectors.
let schema = create_test_schema();
-
- // Test string equality comparison
- let eq_filter = Filter {
- field_name: "string_col".to_string(),
- operator: ExprOperator::Eq,
- values: vec!["test".to_string()],
- };
- let schemable = SchemableFilter::try_from((eq_filter, &schema))?;
-
- let test_array = StringArray::from(vec!["test", "other", "test"]);
- let result = schemable.apply_comparison(&test_array)?;
- assert_eq!(result, BooleanArray::from(vec![true, false, true]));
-
- // Test integer greater than comparison
- let gt_filter = Filter {
- field_name: "int_col".to_string(),
- operator: ExprOperator::Gt,
- values: vec!["50".to_string()],
- };
- let schemable = SchemableFilter::try_from((gt_filter, &schema))?;
-
- let test_array = Int64Array::from(vec![40, 50, 60]);
- let result = schemable.apply_comparison(&test_array)?;
- assert_eq!(result, BooleanArray::from(vec![false, false, true]));
-
- Ok(())
- }
-
- #[test]
- fn test_schemable_filter_all_operators() -> Result<()> {
- let schema = create_test_schema();
- let test_array = Int64Array::from(vec![40, 50, 60]);
-
- let test_cases = vec![
- (ExprOperator::Eq, "50", vec![false, true, false]),
- (ExprOperator::Ne, "50", vec![true, false, true]),
- (ExprOperator::Lt, "50", vec![true, false, false]),
- (ExprOperator::Lte, "50", vec![true, true, false]),
- (ExprOperator::Gt, "50", vec![false, false, true]),
- (ExprOperator::Gte, "50", vec![false, true, true]),
+ let string_array = StringArray::from(vec!["a", "b", "a"]);
+ let int_array = Int64Array::from(vec![40, 50, 60]);
+
+ let cases: Vec<(&str, Vec<&str>, ExprOperator, Vec<bool>)> = vec![
+ // String column
+ (
+ "string_col",
+ vec!["a"],
+ ExprOperator::Eq,
+ vec![true, false, true],
+ ),
+ (
+ "string_col",
+ vec!["a"],
+ ExprOperator::Ne,
+ vec![false, true, false],
+ ),
+ (
+ "string_col",
+ vec!["a", "b"],
+ ExprOperator::In,
+ vec![true, true, true],
+ ),
+ (
+ "string_col",
+ vec!["a", "b"],
+ ExprOperator::NotIn,
+ vec![false, false, false],
+ ),
+ // Int column β covers all 6 binary operators + 2 set operators
+ (
+ "int_col",
+ vec!["50"],
+ ExprOperator::Eq,
+ vec![false, true, false],
+ ),
+ (
+ "int_col",
+ vec!["50"],
+ ExprOperator::Ne,
+ vec![true, false, true],
+ ),
+ (
+ "int_col",
+ vec!["50"],
+ ExprOperator::Lt,
+ vec![true, false, false],
+ ),
+ (
+ "int_col",
+ vec!["50"],
+ ExprOperator::Lte,
+ vec![true, true, false],
+ ),
+ (
+ "int_col",
+ vec!["50"],
+ ExprOperator::Gt,
+ vec![false, false, true],
+ ),
+ (
+ "int_col",
+ vec!["50"],
+ ExprOperator::Gte,
+ vec![false, true, true],
+ ),
+ (
+ "int_col",
+ vec!["40", "60"],
+ ExprOperator::In,
+ vec![true, false, true],
+ ),
+ (
+ "int_col",
+ vec!["40", "60"],
+ ExprOperator::NotIn,
+ vec![false, true, false],
+ ),
];
- for (operator, value, expected) in test_cases {
- let filter = Filter {
- field_name: "int_col".to_string(),
+ for (column, values, operator, expected) in cases {
+ let filter = Filter::new(
+ column.to_string(),
operator,
- values: vec![value.to_string()],
- };
-
+ values.iter().map(|s| s.to_string()).collect(),
+ )?;
let schemable = SchemableFilter::try_from((filter, &schema))?;
- let result = schemable.apply_comparison(&test_array)?;
+ let result = match column {
+ "string_col" => schemable.apply_comparison(&string_array)?,
+ "int_col" => schemable.apply_comparison(&int_array)?,
+ other => unreachable!("unexpected column {other}"),
+ };
assert_eq!(
result,
BooleanArray::from(expected),
- "Failed for operator {operator:?} with value {value}"
+ "operator {operator:?} on {column} with {values:?}"
);
}
Ok(())
}
- #[test]
- fn test_schemable_filter_in() -> Result<()> {
- let schema = create_test_schema();
-
- let in_filter = Filter::new(
- "string_col".to_string(),
- ExprOperator::In,
- vec!["foo".to_string(), "bar".to_string()],
- )
- .unwrap();
-
- let schemable = SchemableFilter::try_from((in_filter, &schema))?;
- let test_array = StringArray::from(vec!["foo", "baz", "bar", "qux"]);
- let result = schemable.apply_comparison(&test_array)?;
- assert_eq!(result, BooleanArray::from(vec![true, false, true, false]));
-
- Ok(())
- }
-
- #[test]
- fn test_schemable_filter_not_in() -> Result<()> {
- let schema = create_test_schema();
-
- let not_in_filter = Filter::new(
- "string_col".to_string(),
- ExprOperator::NotIn,
- vec!["foo".to_string(), "bar".to_string()],
- )
- .unwrap();
-
- let schemable = SchemableFilter::try_from((not_in_filter, &schema))?;
- let test_array = StringArray::from(vec!["foo", "baz", "bar", "qux"]);
- let result = schemable.apply_comparison(&test_array)?;
- assert_eq!(result, BooleanArray::from(vec![false, true, false, true]));
-
- Ok(())
- }
-
- #[test]
- fn test_schemable_filter_in_with_integers() -> Result<()> {
- let schema = create_test_schema();
-
- let in_filter = Filter::new(
- "int_col".to_string(),
- ExprOperator::In,
- vec!["40".to_string(), "60".to_string()],
- )
- .unwrap();
-
- let schemable = SchemableFilter::try_from((in_filter, &schema))?;
- let test_array = Int64Array::from(vec![40, 50, 60]);
- let result = schemable.apply_comparison(&test_array)?;
- assert_eq!(result, BooleanArray::from(vec![true, false, true]));
-
- Ok(())
- }
-
#[test]
fn test_filter_try_from_tuple_in_operator() -> Result<()> {
// TryFrom should parse comma-separated values for IN
@@ -510,7 +702,7 @@ mod tests {
#[test]
fn test_filter_roundtrip_in_operator() -> Result<()> {
- // Round-trip: Filter -> (String, String, String) -> Filter
+ // Round-trip plain values (no escape needed).
let original = Filter::new(
"col".to_string(),
ExprOperator::In,
@@ -530,4 +722,75 @@ mod tests {
Ok(())
}
+
+ #[test]
+ fn test_filter_in_value_comma_escape_parses_literal_comma() -> Result<()> {
+ // `\,` is a literal comma; `,` is a separator.
+ let filter = Filter::try_from(("name", "IN", "Smith\\, John,Jane"))?;
+ assert_eq!(filter.operator, ExprOperator::In);
+ assert_eq!(filter.values, vec!["Smith, John", "Jane"]);
+
+ // `\\` is a literal backslash; non-special escapes are preserved as `\`.
+ let filter = Filter::try_from(("name", "IN", "a\\\\b,c\\d"))?;
+ assert_eq!(filter.values, vec!["a\\b", "c\\d"]);
+
+ // Trailing escape with no following char is preserved as a literal `\`.
+ let filter = Filter::try_from(("name", "IN", "x\\"))?;
+ assert_eq!(filter.values, vec!["x\\"]);
+
+ // Empty segments after trim are dropped; a fully empty value still errors via cardinality.
+ let filter = Filter::try_from(("name", "IN", "a,,b"))?;
+ assert_eq!(filter.values, vec!["a", "b"]);
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_filter_roundtrip_in_value_with_special_chars() -> Result<()> {
+ // Filter values containing commas and backslashes must round-trip
+ // through the (String, String, String) tuple form.
+ let original = Filter::new(
+ "name".to_string(),
+ ExprOperator::In,
+ vec!["Smith, John".to_string(), "back\\slash".to_string()],
+ )?;
+ let tuple: (String, String, String) = original.into();
+ assert_eq!(
+ tuple,
+ (
+ "name".to_string(),
+ "IN".to_string(),
+ "Smith\\, John,back\\\\slash".to_string()
+ )
+ );
+ let restored = Filter::try_from((tuple.0.as_str(), tuple.1.as_str(), tuple.2.as_str()))?;
+ assert_eq!(restored.values, vec!["Smith, John", "back\\slash"]);
+ Ok(())
+ }
+
+ #[test]
+ fn test_filter_roundtrip_scalar_with_special_chars() -> Result<()> {
+ // Scalar operators must not escape commas or backslashes because the
+ // parser does not unescape them for non-IN operators.
+ for op in [
+ ExprOperator::Eq,
+ ExprOperator::Ne,
+ ExprOperator::Lt,
+ ExprOperator::Lte,
+ ExprOperator::Gt,
+ ExprOperator::Gte,
+ ] {
+ let original = Filter::new("city".to_string(), op, vec!["a,b\\c".to_string()])?;
+ let tuple: (String, String, String) = original.into();
+ assert_eq!(tuple.2, "a,b\\c", "operator {op:?} should not escape");
+ let restored =
+ Filter::try_from((tuple.0.as_str(), tuple.1.as_str(), tuple.2.as_str()))?;
+ assert_eq!(
+ restored.values,
+ vec!["a,b\\c"],
+ "operator {op:?} round-trip"
+ );
+ }
+ Ok(())
+ }
}
diff --git a/crates/core/src/file_group/reader.rs b/crates/core/src/file_group/reader.rs
index 9a683cce..751f14a2 100644
--- a/crates/core/src/file_group/reader.rs
+++ b/crates/core/src/file_group/reader.rs
@@ -20,10 +20,11 @@ use crate::Result;
use crate::config::HudiConfigs;
use crate::config::read::HudiReadConfig;
use crate::config::table::HudiTableConfig;
-use crate::config::util::split_hudi_options_from_others;
use crate::error::CoreError;
use crate::error::CoreError::ReadFileSliceError;
-use crate::expr::filter::{Filter, SchemableFilter};
+use crate::expr::filter::{
+ Filter, SchemableFilter, filters_to_row_mask, validate_fields_against_schemas,
+};
use crate::file_group::file_slice::FileSlice;
use crate::file_group::log_file::scanner::{LogFileScanner, ScanResult};
use crate::file_group::record_batches::RecordBatches;
@@ -36,6 +37,7 @@ use crate::storage::{ParquetReadOptions, Storage};
use crate::table::ReadOptions;
use crate::table::builder::OptionResolver;
use crate::timeline::selector::InstantRange;
+use crate::util::arrow::project_batch_by_names;
use arrow::compute::and;
use arrow::compute::filter_record_batch;
use arrow_array::{BooleanArray, RecordBatch};
@@ -53,25 +55,25 @@ pub struct FileGroupReader {
}
impl FileGroupReader {
- /// Creates a new reader with the given Hudi configurations and overwriting options.
+ /// Creates a new reader from base Hudi configs plus pre-split per-call
+ /// overrides β Hudi configs and storage options live in separate maps so
+ /// callers can't accidentally cross the streams.
///
- /// # Notes
- /// This API does **not** use [`OptionResolver`] that loads table properties from storage to resolve options.
- pub(crate) fn new_with_configs_and_overwriting_options<I, K, V>(
+ /// `extra_hudi_opts` extends `hudi_configs` (last-writer-wins). `storage_opts`
+ /// is the full storage option set for this reader (table-level + any overrides
+ /// the caller has already merged in).
+ ///
+ /// This API does **not** use [`OptionResolver`] that loads table properties
+ /// from storage to resolve options β callers supply final configs.
+ pub(crate) fn new_with_overrides(
hudi_configs: Arc<HudiConfigs>,
- overwriting_options: I,
- ) -> Result<Self>
- where
- I: IntoIterator<Item = (K, V)>,
- K: AsRef<str>,
- V: Into<String>,
- {
- let (hudi_opts, others) = split_hudi_options_from_others(overwriting_options);
-
+ extra_hudi_opts: HashMap<String, String>,
+ storage_opts: HashMap<String, String>,
+ ) -> Result<Self> {
let mut final_opts = hudi_configs.as_options();
- final_opts.extend(hudi_opts);
+ final_opts.extend(extra_hudi_opts);
let hudi_configs = Arc::new(HudiConfigs::new(final_opts));
- let storage = Storage::new(Arc::new(others), hudi_configs.clone())?;
+ let storage = Storage::new(Arc::new(storage_opts), hudi_configs.clone())?;
Ok(Self {
hudi_configs,
@@ -104,23 +106,15 @@ impl FileGroupReader {
})
}
- /// Reads the data from the base file at the given relative path.
- ///
- /// # Arguments
- /// * `relative_path` - The relative path to the base file.
- ///
- /// # Returns
- /// A record batch read from the base file.
- pub async fn read_file_slice_by_base_file_path(
- &self,
- relative_path: &str,
- ) -> Result<RecordBatch> {
+ /// Internal: read base file + apply commit-time filter, no [`ReadOptions`] applied.
+ /// Used by the merge path so options aren't applied prematurely before merging
+ /// with log files.
+ async fn read_base_file_eager(&self, relative_path: &str) -> Result<RecordBatch> {
let records: RecordBatch = self
.storage
.get_parquet_file_data(relative_path)
.map_err(|e| ReadFileSliceError(format!("Failed to read path {relative_path}: {e:?}")))
.await?;
-
apply_commit_time_filter(&self.hudi_configs, records)
}
@@ -131,23 +125,23 @@ impl FileGroupReader {
.into();
let start_timestamp = self
.hudi_configs
- .try_get(HudiReadConfig::FileGroupStartTimestamp)
+ .try_get(HudiReadConfig::StartTimestamp)
.map(|v| -> String { v.into() });
let end_timestamp = self
.hudi_configs
- .try_get(HudiReadConfig::FileGroupEndTimestamp)
+ .try_get(HudiReadConfig::EndTimestamp)
.map(|v| -> String { v.into() });
InstantRange::new(timezone, start_timestamp, end_timestamp, false, true)
}
/// Reads the data from the given file slice.
///
- /// # Arguments
- /// * `file_slice` - The file slice to read.
- ///
- /// # Returns
- /// A record batch read from the file slice.
- pub async fn read_file_slice(&self, file_slice: &FileSlice) -> Result<RecordBatch> {
+ /// See [`Self::read_file_slice_from_paths`] for how `options` is applied.
+ pub async fn read_file_slice(
+ &self,
+ file_slice: &FileSlice,
+ options: &ReadOptions,
+ ) -> Result<RecordBatch> {
let base_file_path = file_slice.base_file_relative_path()?;
let log_file_paths = if file_slice.has_log_file() {
file_slice
@@ -158,22 +152,21 @@ impl FileGroupReader {
} else {
vec![]
};
- self.read_file_slice_from_paths(&base_file_path, log_file_paths)
+ self.read_file_slice_from_paths(&base_file_path, log_file_paths, options)
.await
}
/// Reads a file slice from a base file and a list of log files.
///
- /// # Arguments
- /// * `base_file_path` - The relative path to the base file.
- /// * `log_file_paths` - An iterator of relative paths to log files.
- ///
- /// # Returns
- /// A record batch read from the base file merged with log files.
+ /// `options.filters` are applied as a row-level mask after reading;
+ /// `options.projection` selects columns. Both apply to the merged result.
+ /// Other fields (`as_of_timestamp`, `start_timestamp`, `end_timestamp`, `batch_size`)
+ /// are not meaningful for eager reads and are ignored.
pub async fn read_file_slice_from_paths<I, S>(
&self,
base_file_path: &str,
log_file_paths: I,
+ options: &ReadOptions,
) -> Result<RecordBatch>
where
I: IntoIterator<Item = S>,
@@ -189,8 +182,8 @@ impl FileGroupReader {
.into();
let base_file_only = log_file_paths.is_empty() || use_read_optimized;
- if base_file_only {
- self.read_file_slice_by_base_file_path(base_file_path).await
+ let merged = if base_file_only {
+ self.read_base_file_eager(base_file_path).await?
} else {
let instant_range = self.create_instant_range_for_log_file_scan();
let scan_result = LogFileScanner::new(self.hudi_configs.clone(), self.storage.clone())
@@ -207,9 +200,7 @@ impl FileGroupReader {
}
};
- let base_batch = self
- .read_file_slice_by_base_file_path(base_file_path)
- .await?;
+ let base_batch = self.read_base_file_eager(base_file_path).await?;
let schema = base_batch.schema();
let num_data_batches = log_batches.num_data_batches() + 1;
let num_delete_batches = log_batches.num_delete_batches();
@@ -219,8 +210,10 @@ impl FileGroupReader {
all_batches.extend(log_batches);
let merger = RecordMerger::new(schema.clone(), self.hudi_configs.clone());
- merger.merge_record_batches(all_batches)
- }
+ merger.merge_record_batches(all_batches)?
+ };
+
+ apply_eager_options(options, merged)
}
// =========================================================================
@@ -240,11 +233,6 @@ impl FileGroupReader {
/// and yields the merged result as a single batch. This limitation exists because
/// streaming merge of base files with log files is not yet implemented.
///
- /// # Limitations
- ///
- /// - The `projection` and `row_predicate` fields in [ReadOptions] are not yet
- /// implemented for streaming reads. Only `batch_size` is currently supported.
- ///
/// # Arguments
/// * `file_slice` - The file slice to read.
/// * `options` - Read options for configuring the read operation.
@@ -324,7 +312,7 @@ impl FileGroupReader {
} else {
// Fallback: collect + merge, then yield as single-item stream
let batch = self
- .read_file_slice_from_paths(base_file_path, log_file_paths)
+ .read_file_slice_from_paths(base_file_path, log_file_paths, options)
.await?;
Ok(Box::pin(futures::stream::once(async { Ok(batch) })))
}
@@ -335,7 +323,8 @@ impl FileGroupReader {
/// Supports the following [ReadOptions]:
/// - `batch_size`: Controls the number of rows per batch
/// - `projection`: Pushes column selection to the parquet reader level
- /// - `row_predicate`: Filters rows after reading each batch
+ /// - `filters`: Applied as a row-level mask after reading each batch (in addition to
+ /// any pruning that already happened upstream)
async fn read_base_file_stream(
&self,
relative_path: &str,
@@ -345,18 +334,75 @@ impl FileGroupReader {
.hudi_configs
.get_or_default(HudiReadConfig::StreamBatchSize)
.into();
- let batch_size = options.batch_size.unwrap_or(default_batch_size);
+ let batch_size = options.batch_size()?.unwrap_or(default_batch_size);
let mut parquet_options = ParquetReadOptions::new().with_batch_size(batch_size);
- // Add projection pushdown using column names (converted to indices internally
- // by get_parquet_file_stream using the same schema the projection is applied to)
- if let Some(ref projection_names) = options.projection {
- parquet_options = parquet_options.with_projection(projection_names.clone());
+ // If projection is set, widen the parquet read to also include any columns
+ // we need post-read but the user didn't request:
+ // - filter fields, so the row-level mask can evaluate them
+ // - `_hoodie_commit_time`, when commit-time filtering is active
+ // (PopulatesMetaFields + StartTimestamp)
+ // The widened columns are dropped by the final projection step below.
+ //
+ // We only exclude partition column filter fields from widening when
+ // `hoodie.datasource.write.drop.partition.columns` is enabled β otherwise
+ // partition columns are still present in parquet (e.g. with timestamp-based
+ // keygen, the source data column is also configured as a partition field).
+ // Excluding them unconditionally would silently drop legitimate row filters
+ // on those columns.
+ let drops_partition_columns: bool = self
+ .hudi_configs
+ .get_or_default(HudiTableConfig::DropsPartitionFields)
+ .into();
+ let dropped_partition_columns: Vec<String> = if drops_partition_columns {
+ self.hudi_configs
+ .get_or_default(HudiTableConfig::PartitionFields)
+ .into()
+ } else {
+ Vec::new()
+ };
+ let needs_commit_time_col: bool = {
+ let populates_meta_fields: bool = self
+ .hudi_configs
+ .get_or_default(HudiTableConfig::PopulatesMetaFields)
+ .into();
+ let has_start_ts = self
+ .hudi_configs
+ .try_get(HudiReadConfig::StartTimestamp)
+ .is_some();
+ populates_meta_fields && has_start_ts
+ };
+ let final_projection = options.projection.clone();
+ let read_projection = options.projection.as_ref().map(|proj| {
+ let mut combined: Vec<String> = proj.clone();
+ for filter in &options.filters {
+ let field = filter.field.as_str();
+ if dropped_partition_columns.iter().any(|p| p == field) {
+ continue;
+ }
+ if !combined.iter().any(|c| c == field) {
+ combined.push(field.to_string());
+ }
+ }
+ if needs_commit_time_col {
+ let commit_time = MetaField::CommitTime.as_ref().to_string();
+ if !combined.iter().any(|c| c == &commit_time) {
+ combined.push(commit_time);
+ }
+ }
+ combined
+ });
+ if let Some(ref cols) = read_projection {
+ parquet_options = parquet_options.with_projection(cols.clone());
}
let hudi_configs = self.hudi_configs.clone();
let path = relative_path.to_string();
- let row_predicate = options.row_predicate.clone();
+ let filters = Arc::new(options.filters.clone());
+ let final_projection = Arc::new(final_projection);
+ // Validate once on first batch so typoed filter columns surface as errors
+ // rather than silent no-ops in `filters_to_row_mask`.
+ let validated = Arc::new(std::sync::atomic::AtomicBool::new(false));
let parquet_stream = self
.storage
@@ -364,35 +410,44 @@ impl FileGroupReader {
.map_err(|e| ReadFileSliceError(format!("Failed to read path {path}: {e:?}")))
.await?;
- // Apply filtering: commit time filter first, then row predicate
+ // Apply filtering: commit time β structured filters β final projection.
let stream = parquet_stream.into_stream().filter_map(move |result| {
let hudi_configs = hudi_configs.clone();
- let row_predicate = row_predicate.clone();
+ let filters = filters.clone();
+ let final_projection = final_projection.clone();
+ let validated = validated.clone();
async move {
match result {
Err(e) => Some(Err(ReadFileSliceError(format!(
"Failed to read batch: {e:?}"
)))),
Ok(batch) => {
- // Apply commit time filter
- let filtered = match apply_commit_time_filter(&hudi_configs, batch) {
+ if !validated.load(std::sync::atomic::Ordering::Relaxed) {
+ if let Err(e) =
+ validate_fields_against_schemas(&filters, [batch.schema().as_ref()])
+ {
+ return Some(Err(e));
+ }
+ validated.store(true, std::sync::atomic::Ordering::Relaxed);
+ }
+ let batch = match apply_commit_time_filter(&hudi_configs, batch) {
Err(e) => return Some(Err(e)),
Ok(b) if b.num_rows() == 0 => return None,
Ok(b) => b,
};
-
- // Apply row predicate if present
- let final_batch = if let Some(ref predicate) = row_predicate {
- match apply_row_predicate(predicate.as_ref(), filtered) {
- Err(e) => return Some(Err(e)),
- Ok(b) if b.num_rows() == 0 => return None,
- Ok(b) => b,
- }
- } else {
- filtered
+ let batch = match apply_filter_mask(&filters, batch) {
+ Err(e) => return Some(Err(e)),
+ Ok(b) if b.num_rows() == 0 => return None,
+ Ok(b) => b,
};
-
- Some(Ok(final_batch))
+ // Project down to the user's requested columns (no-op if we
+ // didn't have to widen the read projection).
+ let batch = match project_batch_by_names(batch, final_projection.as_deref())
+ {
+ Err(e) => return Some(Err(e)),
+ Ok(b) => b,
+ };
+ Some(Ok(batch))
}
}
}
@@ -516,7 +571,7 @@ fn create_commit_time_filter_mask(
}
let start_ts = hudi_configs
- .try_get(HudiReadConfig::FileGroupStartTimestamp)
+ .try_get(HudiReadConfig::StartTimestamp)
.map(|v| -> String { v.into() });
if start_ts.is_none() {
// If start timestamp is not provided, the query is snapshot or time-travel
@@ -532,7 +587,7 @@ fn create_commit_time_filter_mask(
}
if let Some(end) = hudi_configs
- .try_get(HudiReadConfig::FileGroupEndTimestamp)
+ .try_get(HudiReadConfig::EndTimestamp)
.map(|v| -> String { v.into() })
{
let filter = Filter::try_from((MetaField::CommitTime.as_ref(), "<=", end.as_str()))?;
@@ -556,6 +611,19 @@ fn create_commit_time_filter_mask(
Ok(Some(mask))
}
+/// Apply structured filters and projection to an eager [`RecordBatch`].
+///
+/// All `options.filters` must target columns present in the batch β at file-group
+/// level no upstream partition pruning has happened, so a filter on a column that
+/// isn't in the batch can never apply and is rejected with a schema error. Callers
+/// going through `Table` strip filters on dropped partition columns before reaching
+/// here; direct `FileGroupReader` callers must not pass such filters.
+fn apply_eager_options(options: &ReadOptions, batch: RecordBatch) -> Result<RecordBatch> {
+ validate_fields_against_schemas(&options.filters, [batch.schema().as_ref()])?;
+ let batch = apply_filter_mask(&options.filters, batch)?;
+ project_batch_by_names(batch, options.projection.as_deref())
+}
+
/// Apply commit time filtering to a record batch.
fn apply_commit_time_filter(hudi_configs: &HudiConfigs, batch: RecordBatch) -> Result<RecordBatch> {
match create_commit_time_filter_mask(hudi_configs, &batch)? {
@@ -565,14 +633,17 @@ fn apply_commit_time_filter(hudi_configs: &HudiConfigs, batch: RecordBatch) -> R
}
}
-/// Apply a row predicate to filter records in a batch.
-fn apply_row_predicate(
- predicate: &(dyn Fn(&RecordBatch) -> Result<BooleanArray> + Send + Sync),
- batch: RecordBatch,
-) -> Result<RecordBatch> {
- let mask = predicate(&batch)?;
+/// Apply structured filters as a row mask on the batch.
+///
+/// Filters whose field is not present in the batch (e.g., partition columns already
+/// pruned upstream) are skipped β see [`crate::expr::filter::filters_to_row_mask`].
+fn apply_filter_mask(filters: &[Filter], batch: RecordBatch) -> Result<RecordBatch> {
+ if filters.is_empty() {
+ return Ok(batch);
+ }
+ let mask = filters_to_row_mask(filters, &batch)?;
filter_record_batch(&batch, &mask)
- .map_err(|e| ReadFileSliceError(format!("Failed to apply row predicate: {e:?}")))
+ .map_err(|e| ReadFileSliceError(format!("Failed to apply filter mask: {e:?}")))
}
#[cfg(test)]
@@ -656,6 +727,38 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_new_with_options_resolves_table_properties_from_storage() {
+ // The minimum-props fixture's hoodie.properties carries TableType,
+ // TableName, and TableVersion. With empty user options, the
+ // OptionResolver must read them off storage and seed hudi_configs β
+ // otherwise downstream commit-time / merge logic would fall back to
+ // defaults and silently misbehave on real tables.
+ let base_uri = get_base_uri_with_valid_props_minimum();
+ let reader = FileGroupReader::new_with_options(&base_uri, empty_options())
+ .await
+ .unwrap();
+
+ let table_type: String = reader
+ .hudi_configs
+ .get(HudiTableConfig::TableType)
+ .unwrap()
+ .into();
+ assert_eq!(table_type, "COPY_ON_WRITE");
+ let table_name: String = reader
+ .hudi_configs
+ .get(HudiTableConfig::TableName)
+ .unwrap()
+ .into();
+ assert_eq!(table_name, "trips");
+ let table_version: isize = reader
+ .hudi_configs
+ .get(HudiTableConfig::TableVersion)
+ .unwrap()
+ .into();
+ assert_eq!(table_version, 6);
+ }
+
#[tokio::test]
async fn test_new_with_options_invalid_base_uri_or_invalid_props() {
let base_uri = get_non_existent_base_uri();
@@ -694,7 +797,7 @@ mod tests {
&base_uri,
[
(HudiTableConfig::PopulatesMetaFields.as_ref(), "false"),
- (HudiReadConfig::FileGroupStartTimestamp.as_ref(), "2"),
+ (HudiReadConfig::StartTimestamp.as_ref(), "2"),
],
)
.await?;
@@ -707,11 +810,9 @@ mod tests {
assert_eq!(mask, None);
// Test case 3: Filtering commit time > '2'
- let reader = FileGroupReader::new_with_options(
- &base_uri,
- [(HudiReadConfig::FileGroupStartTimestamp, "2")],
- )
- .await?;
+ let reader =
+ FileGroupReader::new_with_options(&base_uri, [(HudiReadConfig::StartTimestamp, "2")])
+ .await?;
let mask = create_commit_time_filter_mask(&reader.hudi_configs, &records)?;
assert_eq!(
mask,
@@ -720,11 +821,9 @@ mod tests {
);
// Test case 4: Filtering commit time <= '4'
- let reader = FileGroupReader::new_with_options(
- &base_uri,
- [(HudiReadConfig::FileGroupEndTimestamp, "4")],
- )
- .await?;
+ let reader =
+ FileGroupReader::new_with_options(&base_uri, [(HudiReadConfig::EndTimestamp, "4")])
+ .await?;
let mask = create_commit_time_filter_mask(&reader.hudi_configs, &records)?;
assert_eq!(mask, None, "Commit time filtering should not be needed");
@@ -732,8 +831,8 @@ mod tests {
let reader = FileGroupReader::new_with_options(
&base_uri,
[
- (HudiReadConfig::FileGroupStartTimestamp, "2"),
- (HudiReadConfig::FileGroupEndTimestamp, "4"),
+ (HudiReadConfig::StartTimestamp, "2"),
+ (HudiReadConfig::EndTimestamp, "4"),
],
)
.await?;
@@ -757,7 +856,7 @@ mod tests {
let log_file_paths: Vec<&str> = vec![];
let result = reader
- .read_file_slice_from_paths(base_file_path, log_file_paths)
+ .read_file_slice_from_paths(base_file_path, log_file_paths, &ReadOptions::new())
.await;
match result {
@@ -788,11 +887,11 @@ mod tests {
let log_file_paths = vec![TEST_SAMPLE_LOG_FILE.to_string()];
let result = reader
- .read_file_slice_from_paths(base_file_path, log_file_paths)
+ .read_file_slice_from_paths(base_file_path, log_file_paths, &ReadOptions::new())
.await;
// In read-optimized mode, log files should be ignored
- // This should behave the same as read_file_slice_by_base_file_path
+ // This should behave the same as a base-file-only read
match result {
Ok(_) => {
// Test passes if we get a result - the method correctly ignored log files
@@ -819,7 +918,7 @@ mod tests {
let log_file_paths = vec![TEST_SAMPLE_LOG_FILE.to_string()];
let result = reader
- .read_file_slice_from_paths(base_file_path, log_file_paths)
+ .read_file_slice_from_paths(base_file_path, log_file_paths, &ReadOptions::new())
.await;
// The actual file reading might fail due to missing test data, which is expected
@@ -850,7 +949,7 @@ mod tests {
let log_file_paths: Vec<&str> = vec![];
let result = reader
- .read_file_slice_from_paths(base_file_path, log_file_paths)
+ .read_file_slice_from_paths(base_file_path, log_file_paths, &ReadOptions::new())
.await;
assert!(result.is_err(), "Should return error for non-existent file");
@@ -876,7 +975,9 @@ mod tests {
let file_slice = FileSlice::new(base_file, String::new()); // empty partition path
// Call read_file_slice
- let result = reader.read_file_slice(&file_slice).await;
+ let result = reader
+ .read_file_slice(&file_slice, &ReadOptions::new())
+ .await;
match result {
Ok(batch) => {
@@ -907,16 +1008,18 @@ mod tests {
/// Helper to create a FileGroupReader without using block_on (safe for async tests).
fn create_test_reader(base_uri: &str) -> Result<FileGroupReader> {
let hudi_configs = Arc::new(HudiConfigs::new([(HudiTableConfig::BasePath, base_uri)]));
- FileGroupReader::new_with_configs_and_overwriting_options(hudi_configs, empty_options())
+ FileGroupReader::new_with_overrides(hudi_configs, HashMap::new(), HashMap::new())
}
/// Helper to create a FileGroupReader with read-optimized mode.
fn create_test_reader_read_optimized(base_uri: &str) -> Result<FileGroupReader> {
let hudi_configs = Arc::new(HudiConfigs::new([(HudiTableConfig::BasePath, base_uri)]));
- FileGroupReader::new_with_configs_and_overwriting_options(
- hudi_configs,
- [(HudiReadConfig::UseReadOptimizedMode.as_ref(), "true")],
- )
+ let mut hudi_opts = HashMap::new();
+ hudi_opts.insert(
+ HudiReadConfig::UseReadOptimizedMode.as_ref().to_string(),
+ "true".to_string(),
+ );
+ FileGroupReader::new_with_overrides(hudi_configs, hudi_opts, HashMap::new())
}
#[tokio::test]
@@ -1092,13 +1195,7 @@ mod tests {
let file_slice = FileSlice::new(base_file, String::new());
// Use very small batch size
- let options = ReadOptions {
- partition_filters: vec![],
- projection: None,
- row_predicate: None,
- batch_size: Some(1),
- as_of_timestamp: None,
- };
+ let options = ReadOptions::new().with_batch_size(1)?;
let result = reader.read_file_slice_stream(&file_slice, &options).await;
@@ -1162,13 +1259,16 @@ mod tests {
/// Helper to create a FileGroupReader with commit time filtering options.
fn create_test_reader_with_commit_time_filter(base_uri: &str) -> Result<FileGroupReader> {
let hudi_configs = Arc::new(HudiConfigs::new([(HudiTableConfig::BasePath, base_uri)]));
- FileGroupReader::new_with_configs_and_overwriting_options(
- hudi_configs,
- [
- (HudiReadConfig::FileGroupStartTimestamp.as_ref(), "2"),
- (HudiReadConfig::FileGroupEndTimestamp.as_ref(), "4"),
- ],
- )
+ let mut hudi_opts = HashMap::new();
+ hudi_opts.insert(
+ HudiReadConfig::StartTimestamp.as_ref().to_string(),
+ "2".to_string(),
+ );
+ hudi_opts.insert(
+ HudiReadConfig::EndTimestamp.as_ref().to_string(),
+ "4".to_string(),
+ );
+ FileGroupReader::new_with_overrides(hudi_configs, hudi_opts, HashMap::new())
}
#[tokio::test]
@@ -1239,7 +1339,7 @@ mod tests {
HudiTableConfig::BasePath,
metadata_table_uri.as_str(),
)]));
- FileGroupReader::new_with_configs_and_overwriting_options(hudi_configs, empty_options())
+ FileGroupReader::new_with_overrides(hudi_configs, HashMap::new(), HashMap::new())
}
#[tokio::test]
diff --git a/crates/core/src/keygen/timestamp_based.rs b/crates/core/src/keygen/timestamp_based.rs
index 35247396..5d124ffc 100644
--- a/crates/core/src/keygen/timestamp_based.rs
+++ b/crates/core/src/keygen/timestamp_based.rs
@@ -502,18 +502,18 @@ impl KeyGeneratorFilterTransformer for TimestampBasedKeyGenerator {
}
fn transform_filter(&self, filter: &Filter) -> Result<Vec<Filter>> {
- if filter.field_name != self.source_field {
+ if filter.field != self.source_field {
return Ok(vec![filter.clone()]);
}
- let field_name = MetaField::PartitionPath.as_ref().to_string();
+ let field = MetaField::PartitionPath.as_ref().to_string();
match filter.operator {
ExprOperator::Eq | ExprOperator::Ne => {
let dt = self.parse_timestamp(&filter.values[0])?;
let path = self.format_partition_path(&dt);
Ok(vec![Filter {
- field_name,
+ field,
operator: filter.operator,
values: vec![path],
}])
@@ -528,7 +528,7 @@ impl KeyGeneratorFilterTransformer for TimestampBasedKeyGenerator {
})
.collect();
Ok(vec![Filter {
- field_name,
+ field,
operator: filter.operator,
values: paths?,
}])
@@ -547,7 +547,7 @@ impl KeyGeneratorFilterTransformer for TimestampBasedKeyGenerator {
other => other,
};
Ok(vec![Filter {
- field_name,
+ field,
operator: op,
values: vec![path],
}])
@@ -882,13 +882,13 @@ mod tests {
TimestampBasedKeyGenerator::from_configs(&create_test_configs_date_string()).unwrap();
let filter = Filter {
- field_name: "ts_str".to_string(),
+ field: "ts_str".to_string(),
operator: ExprOperator::Eq,
values: vec!["2023-04-01T12:01:00.123Z".to_string()],
};
let transformed = keygen.transform_filter(&filter).unwrap();
assert_eq!(transformed.len(), 1);
- assert_eq!(transformed[0].field_name, "_hoodie_partition_path");
+ assert_eq!(transformed[0].field, "_hoodie_partition_path");
assert_eq!(transformed[0].operator, ExprOperator::Eq);
assert_eq!(
transformed[0].values[0],
@@ -901,7 +901,7 @@ mod tests {
.unwrap();
// 2024-01-25 00:00:00 UTC = 1706140800 seconds
let filter = Filter {
- field_name: "event_timestamp".to_string(),
+ field: "event_timestamp".to_string(),
operator: ExprOperator::Eq,
values: vec!["1706140800".to_string()],
};
@@ -913,7 +913,7 @@ mod tests {
let keygen =
TimestampBasedKeyGenerator::from_configs(&create_test_configs_date_string()).unwrap();
let filter = Filter {
- field_name: "ts_str".to_string(),
+ field: "ts_str".to_string(),
operator: ExprOperator::Ne,
values: vec!["2023-04-01T12:01:00.123Z".to_string()],
};
@@ -936,13 +936,13 @@ mod tests {
(ExprOperator::Lte, ExprOperator::Lte),
] {
let filter = Filter {
- field_name: "event_timestamp".to_string(),
+ field: "event_timestamp".to_string(),
operator: input_op,
values: vec!["1706140800".to_string()],
};
let transformed = keygen.transform_filter(&filter).unwrap();
assert_eq!(transformed.len(), 1, "Expected 1 filter for {input_op:?}");
- assert_eq!(transformed[0].field_name, "_hoodie_partition_path");
+ assert_eq!(transformed[0].field, "_hoodie_partition_path");
assert_eq!(transformed[0].values[0], "2024/01/25");
assert_eq!(
transformed[0].operator, expected_op,
@@ -954,20 +954,20 @@ mod tests {
let keygen =
TimestampBasedKeyGenerator::from_configs(&create_test_configs_date_string()).unwrap();
let filter = Filter {
- field_name: "other_field".to_string(),
+ field: "other_field".to_string(),
operator: ExprOperator::Eq,
values: vec!["value".to_string()],
};
let transformed = keygen.transform_filter(&filter).unwrap();
assert_eq!(transformed.len(), 1);
- assert_eq!(transformed[0].field_name, "other_field");
+ assert_eq!(transformed[0].field, "other_field");
// Invalid timestamp value produces error, not panic
let unix_keygen =
TimestampBasedKeyGenerator::from_configs(&create_test_configs_unix_timestamp())
.unwrap();
let filter = Filter {
- field_name: "event_timestamp".to_string(),
+ field: "event_timestamp".to_string(),
operator: ExprOperator::Eq,
values: vec!["not_a_number".to_string()],
};
@@ -1025,7 +1025,7 @@ mod tests {
// Range ops should return empty (skip pruning)
let filter = Filter {
- field_name: "ts".to_string(),
+ field: "ts".to_string(),
operator: ExprOperator::Gte,
values: vec!["1706140800".to_string()],
};
@@ -1033,7 +1033,7 @@ mod tests {
// But Eq still works
let filter = Filter {
- field_name: "ts".to_string(),
+ field: "ts".to_string(),
operator: ExprOperator::Eq,
values: vec!["1706140800".to_string()],
};
diff --git a/crates/core/src/metadata/table/mod.rs b/crates/core/src/metadata/table/mod.rs
index 3b4b497f..fd167d49 100644
--- a/crates/core/src/metadata/table/mod.rs
+++ b/crates/core/src/metadata/table/mod.rs
@@ -258,10 +258,11 @@ impl Table {
}
let file_slice = file_slices.into_iter().next().unwrap();
- let fg_reader = self.create_file_group_reader_with_options([(
- HudiReadConfig::FileGroupEndTimestamp,
- timestamp,
- )])?;
+ let fg_reader = self.create_file_group_reader_with_options(
+ None,
+ [(HudiReadConfig::EndTimestamp, timestamp)],
+ std::iter::empty::<(&str, &str)>(),
+ )?;
fg_reader
.read_metadata_table_files_partition(&file_slice, keys)
diff --git a/crates/core/src/table/file_pruner.rs b/crates/core/src/table/file_pruner.rs
index 50ef0a88..a7030e70 100644
--- a/crates/core/src/table/file_pruner.rs
+++ b/crates/core/src/table/file_pruner.rs
@@ -64,7 +64,7 @@ impl FilePruner {
// Only keep filters on non-partition columns that exist in the table schema
let and_filters: Vec<SchemableFilter> = and_filters
.iter()
- .filter(|filter| !partition_columns.contains(filter.field_name.as_str()))
+ .filter(|filter| !partition_columns.contains(filter.field.as_str()))
.filter_map(|filter| SchemableFilter::try_from((filter.clone(), table_schema)).ok())
.collect();
diff --git a/crates/core/src/table/mod.rs b/crates/core/src/table/mod.rs
index 5ece2037..50259006 100644
--- a/crates/core/src/table/mod.rs
+++ b/crates/core/src/table/mod.rs
@@ -47,30 +47,29 @@
//! 3. read hudi table
//! ```rust
//! use url::Url;
-//! use hudi_core::config::util::empty_filters;
-//! use hudi_core::table::Table;
+//! 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_snapshot(empty_filters()).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::config::util::empty_filters;
-//! use hudi_core::table::Table;
+//! 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 file_slices = hudi_table
-//! .get_file_slices_splits(2, empty_filters())
+//! 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 {
@@ -95,14 +94,15 @@ pub mod partition;
mod read_options;
mod validation;
-pub use read_options::ReadOptions;
+pub use read_options::{QueryType, ReadOptions};
use crate::Result;
use crate::config::HudiConfigs;
use crate::config::read::HudiReadConfig;
use crate::config::table::HudiTableConfig::PartitionFields;
use crate::config::table::{BaseFileFormatValue, HudiTableConfig, TableTypeValue};
-use crate::expr::filter::{Filter, from_str_tuples};
+use crate::error::CoreError;
+use crate::expr::filter::{Filter, validate_fields_against_schemas};
use crate::file_group::file_slice::FileSlice;
use crate::file_group::reader::FileGroupReader;
use crate::keygen::is_timestamp_based_keygen;
@@ -119,7 +119,6 @@ use crate::table::fs_view::FileSystemView;
use crate::table::partition::{PartitionPruner, project_partition_schema};
use crate::timeline::util::format_timestamp;
use crate::timeline::{EARLIEST_START_TIMESTAMP, Timeline};
-use crate::util::collection::split_into_chunks;
use arrow::record_batch::RecordBatch;
use arrow_schema::{Field, Schema};
use std::collections::HashMap;
@@ -336,16 +335,16 @@ impl Table {
/// 1. Timeline commit metadata.
/// 2. `hoodie.properties` file's [HudiTableConfig::CreateSchema].
pub async fn get_schema_in_avro_str(&self) -> Result<String> {
- self.get_schema_in_avro_str_internal(false).await
+ self.get_schema_in_avro_str_inner(false).await
}
/// Get the latest Avro schema string of the table, with Hudi meta fields (`_hoodie_*`)
/// prepended.
pub async fn get_schema_in_avro_str_with_meta_fields(&self) -> Result<String> {
- self.get_schema_in_avro_str_internal(true).await
+ self.get_schema_in_avro_str_inner(true).await
}
- async fn get_schema_in_avro_str_internal(&self, includes_meta_fields: bool) -> Result<String> {
+ async fn get_schema_in_avro_str_inner(&self, includes_meta_fields: bool) -> Result<String> {
if includes_meta_fields {
resolve_avro_schema_with_meta_fields(self).await
} else {
@@ -361,16 +360,16 @@ impl Table {
/// 2. Base file schema.
/// 3. `hoodie.properties` file's [HudiTableConfig::CreateSchema].
pub async fn get_schema(&self) -> Result<Schema> {
- self.get_schema_internal(false).await
+ self.get_schema_inner(false).await
}
/// Get the latest [arrow_schema::Schema] of the table, with Hudi meta fields (`_hoodie_*`)
/// prepended.
pub async fn get_schema_with_meta_fields(&self) -> Result<Schema> {
- self.get_schema_internal(true).await
+ self.get_schema_inner(true).await
}
- async fn get_schema_internal(&self, includes_meta_fields: bool) -> Result<Schema> {
+ async fn get_schema_inner(&self, includes_meta_fields: bool) -> Result<Schema> {
if includes_meta_fields {
resolve_schema(self).await
} else {
@@ -417,104 +416,45 @@ impl Table {
&self.timeline
}
- /// Get all the [FileSlice]s in splits from the table.
+ /// Get the [FileSlice]s the read targets, dispatching on `options.query_type`.
///
- /// # Arguments
- /// * `num_splits` - The number of chunks to split the file slices into.
- /// * `filters` - Partition filters to apply.
- pub async fn get_file_slices_splits<I, S>(
- &self,
- num_splits: usize,
- filters: I,
- ) -> Result<Vec<Vec<FileSlice>>>
- where
- I: IntoIterator<Item = (S, S, S)>,
- S: AsRef<str>,
- {
- if let Some(timestamp) = self.timeline.get_latest_commit_timestamp_as_option() {
- let filters = from_str_tuples(filters)?;
- self.get_file_slices_splits_internal(num_splits, timestamp, &filters)
- .await
- } else {
- Ok(Vec::new())
+ /// - [`QueryType::Snapshot`]: returns slices visible at `options.as_of_timestamp`,
+ /// defaulting to the latest commit. `options.filters` drive both partition
+ /// pruning and file-level stats pruning (when min/max stats are available).
+ /// - [`QueryType::Incremental`]: returns slices changed in
+ /// (`options.start_timestamp`, `options.end_timestamp`], defaulting to earliest
+ /// and latest respectively. `options.filters` drive partition pruning only;
+ /// data-column filters do not prune files at planning time.
+ ///
+ /// Returns an empty vector when the table has no commits.
+ ///
+ /// To bucket the result for parallel reads, use
+ /// [`crate::util::collection::split_into_chunks`] or your engine's preferred
+ /// partitioning policy.
+ pub async fn get_file_slices(&self, options: &ReadOptions) -> Result<Vec<FileSlice>> {
+ match options.query_type()? {
+ QueryType::Snapshot => self.get_snapshot_file_slices(options).await,
+ QueryType::Incremental => self.get_incremental_file_slices(options).await,
}
}
- /// Get all the [FileSlice]s in splits from the table at a given timestamp.
- ///
- /// # Arguments
- /// * `num_splits` - The number of chunks to split the file slices into.
- /// * `timestamp` - The timestamp which file slices associated with.
- /// * `filters` - Partition filters to apply.
- pub async fn get_file_slices_splits_as_of<I, S>(
- &self,
- num_splits: usize,
- timestamp: &str,
- filters: I,
- ) -> Result<Vec<Vec<FileSlice>>>
- where
- I: IntoIterator<Item = (S, S, S)>,
- S: AsRef<str>,
- {
- let timestamp = format_timestamp(timestamp, &self.timezone())?;
- let filters = from_str_tuples(filters)?;
- self.get_file_slices_splits_internal(num_splits, ×tamp, &filters)
+ async fn get_snapshot_file_slices(&self, options: &ReadOptions) -> Result<Vec<FileSlice>> {
+ let Some(timestamp) = self.resolve_snapshot_timestamp(options)? else {
+ return Ok(Vec::new());
+ };
+ self.get_file_slices_inner(×tamp, &options.filters)
.await
}
- async fn get_file_slices_splits_internal(
- &self,
- num_splits: usize,
- timestamp: &str,
- filters: &[Filter],
- ) -> Result<Vec<Vec<FileSlice>>> {
- let file_slices = self.get_file_slices_internal(timestamp, filters).await?;
- Ok(split_into_chunks(file_slices, num_splits))
- }
-
- /// Get all the [FileSlice]s in the table.
- ///
- /// # Arguments
- /// * `filters` - Partition filters to apply.
- ///
- /// # Notes
- /// * This API is useful for implementing snapshot query.
- pub async fn get_file_slices<I, S>(&self, filters: I) -> Result<Vec<FileSlice>>
- where
- I: IntoIterator<Item = (S, S, S)>,
- S: AsRef<str>,
- {
- if let Some(timestamp) = self.timeline.get_latest_commit_timestamp_as_option() {
- let filters = from_str_tuples(filters)?;
- self.get_file_slices_internal(timestamp, &filters).await
- } else {
- Ok(Vec::new())
- }
- }
-
- /// Get all the [FileSlice]s in the table at a given timestamp.
- ///
- /// # Arguments
- /// * `timestamp` - The timestamp which file slices associated with.
- /// * `filters` - Partition filters to apply.
- ///
- /// # Notes
- /// * This API is useful for implementing time travel query.
- pub async fn get_file_slices_as_of<I, S>(
- &self,
- timestamp: &str,
- filters: I,
- ) -> Result<Vec<FileSlice>>
- where
- I: IntoIterator<Item = (S, S, S)>,
- S: AsRef<str>,
- {
- let timestamp = format_timestamp(timestamp, &self.timezone())?;
- let filters = from_str_tuples(filters)?;
- self.get_file_slices_internal(×tamp, &filters).await
+ async fn get_incremental_file_slices(&self, options: &ReadOptions) -> Result<Vec<FileSlice>> {
+ let Some((start, end)) = self.resolve_incremental_range(options)? else {
+ return Ok(Vec::new());
+ };
+ self.get_file_slices_between_inner(&start, &end, &options.filters)
+ .await
}
- async fn get_file_slices_internal(
+ async fn get_file_slices_inner(
&self,
timestamp: &str,
filters: &[Filter],
@@ -522,12 +462,17 @@ impl Table {
let timeline_view = self.timeline.create_view_as_of(timestamp).await?;
let partition_schema = self.get_partition_schema().await?;
+ // Validate against the meta-inclusive schema so filters on Hudi meta fields
+ // (e.g. `_hoodie_record_key`) are accepted β those columns are present in
+ // returned batches and the row-level mask applies them. The pruners are
+ // tolerant of meta-field filters: PartitionPruner ignores non-partition
+ // columns, and FilePruner skips columns without stats.
+ let table_schema = self.get_schema_with_meta_fields().await?;
+ validate_fields_against_schemas(filters, [&table_schema, &partition_schema])?;
+
let partition_pruner =
PartitionPruner::new(filters, &partition_schema, self.hudi_configs.as_ref())?;
- // Get table schema for file pruning
- let table_schema = self.get_schema().await?;
-
// Create file pruner with filters on non-partition columns
let file_pruner = FilePruner::new(filters, &table_schema, &partition_schema)?;
@@ -562,80 +507,11 @@ impl Table {
.await
}
- /// Get all the changed [FileSlice]s in the table between the given timestamps.
- ///
- /// # Arguments
- /// * `start_timestamp` - If provided, only file slices that were changed after this timestamp will be returned.
- /// * `end_timestamp` - If provided, only file slices that were changed before or at this timestamp will be returned.
- ///
- /// # Notes
- /// * This API is useful for implementing incremental query.
- pub async fn get_file_slices_between(
- &self,
- start_timestamp: Option<&str>,
- end_timestamp: Option<&str>,
- ) -> Result<Vec<FileSlice>> {
- // If the end timestamp is not provided, use the latest file slice timestamp.
- // This is the request timestamp (commit_timestamp) for both v6 and v8+ tables.
- let Some(end) =
- end_timestamp.or_else(|| self.timeline.get_latest_commit_timestamp_as_option())
- else {
- // No latest commit timestamp means the table is empty.
- return Ok(Vec::new());
- };
-
- let start = start_timestamp.unwrap_or(EARLIEST_START_TIMESTAMP);
-
- self.get_file_slices_between_internal(start, end).await
- }
-
- /// Get all the changed [FileSlice]s in splits from the table between the given timestamps.
- ///
- /// # Arguments
- /// * `num_splits` - The number of chunks to split the file slices into.
- /// * `start_timestamp` - If provided, only file slices that were changed after this timestamp will be returned.
- /// * `end_timestamp` - If provided, only file slices that were changed before or at this timestamp will be returned.
- ///
- /// # Notes
- /// * This API is useful for implementing incremental query with read parallelism.
- /// * Uses the same splitting flow as the time-travel API to respect read parallelism config.
- pub async fn get_file_slices_splits_between(
- &self,
- num_splits: usize,
- start_timestamp: Option<&str>,
- end_timestamp: Option<&str>,
- ) -> Result<Vec<Vec<FileSlice>>> {
- // If the end timestamp is not provided, use the latest file slice timestamp.
- // This is the request timestamp (commit_timestamp) for both v6 and v8+ tables.
- let Some(end) =
- end_timestamp.or_else(|| self.timeline.get_latest_commit_timestamp_as_option())
- else {
- // No latest commit timestamp means the table is empty.
- return Ok(Vec::new());
- };
-
- let start = start_timestamp.unwrap_or(EARLIEST_START_TIMESTAMP);
-
- self.get_file_slices_splits_between_internal(num_splits, start, end)
- .await
- }
-
- async fn get_file_slices_splits_between_internal(
- &self,
- num_splits: usize,
- start_timestamp: &str,
- end_timestamp: &str,
- ) -> Result<Vec<Vec<FileSlice>>> {
- let file_slices = self
- .get_file_slices_between_internal(start_timestamp, end_timestamp)
- .await?;
- Ok(split_into_chunks(file_slices, num_splits))
- }
-
- async fn get_file_slices_between_internal(
+ async fn get_file_slices_between_inner(
&self,
start_timestamp: &str,
end_timestamp: &str,
+ filters: &[Filter],
) -> Result<Vec<FileSlice>> {
// Seed the cached estimator from a sample base file at or before
// end_timestamp so the file group builder can populate FileMetadata
@@ -647,8 +523,29 @@ impl Table {
.get_file_groups_between(Some(start_timestamp), Some(end_timestamp), estimator)
.await?;
+ // Skip schema fetch and pruner construction when there are no filters.
+ let partition_pruner = if filters.is_empty() {
+ None
+ } else {
+ let partition_schema = self.get_partition_schema().await?;
+ // See `get_file_slices_inner` for why validation uses the
+ // meta-inclusive schema.
+ let table_schema = self.get_schema_with_meta_fields().await?;
+ validate_fields_against_schemas(filters, [&table_schema, &partition_schema])?;
+ Some(PartitionPruner::new(
+ filters,
+ &partition_schema,
+ self.hudi_configs.as_ref(),
+ )?)
+ };
+
let mut file_slices: Vec<FileSlice> = Vec::new();
for file_group in file_groups {
+ if let Some(ref pruner) = partition_pruner
+ && !pruner.should_include(&file_group.partition_path)
+ {
+ continue;
+ }
if let Some(file_slice) = file_group.get_file_slice_as_of(end_timestamp) {
file_slices.push(file_slice.clone());
}
@@ -657,266 +554,302 @@ impl Table {
Ok(file_slices)
}
- /// Create a [FileGroupReader] using the [Table]'s Hudi configs, and overwriting options.
- pub fn create_file_group_reader_with_options<I, K, V>(
+ /// Create a [FileGroupReader] using the [Table]'s Hudi configs.
+ ///
+ /// Two override channels keep Hudi configs and storage credentials cleanly
+ /// separated β a `hoodie.*` key can't be misclassified as storage, and a
+ /// stray storage option can't be silently picked up as a Hudi config.
+ ///
+ /// **Hudi configs** (last-writer-wins):
+ /// 1. Table-level Hudi configs (constant for this `Table` instance).
+ /// 2. `read_options.hudi_options` when `read_options` is `Some`. The four
+ /// `Table`-owned read keys ([`HudiReadConfig::QueryType`],
+ /// [`HudiReadConfig::AsOfTimestamp`], [`HudiReadConfig::StartTimestamp`],
+ /// [`HudiReadConfig::EndTimestamp`]) are stripped here: the `Table` layer
+ /// interprets them for dispatch and snapshot resolution, and a stale value
+ /// (e.g. an incremental `StartTimestamp` left in the bag) would silently
+ /// activate commit-time filtering at the FG reader.
+ /// 3. `extra_hudi_overrides` β caller-supplied resolved Hudi configs;
+ /// always win. Use these to inject the resolved timestamps for the
+ /// current read.
+ ///
+ /// **Storage options** (last-writer-wins):
+ /// 1. Table-level storage options (cloud credentials, endpoints, etc).
+ /// 2. `extra_storage_overrides` β caller-supplied per-path storage overrides.
+ pub fn create_file_group_reader_with_options<H, S, K1, V1, K2, V2>(
&self,
- options: I,
+ read_options: Option<&ReadOptions>,
+ extra_hudi_overrides: H,
+ extra_storage_overrides: S,
) -> Result<FileGroupReader>
where
- I: IntoIterator<Item = (K, V)>,
- K: AsRef<str>,
- V: Into<String>,
+ H: IntoIterator<Item = (K1, V1)>,
+ K1: AsRef<str>,
+ V1: Into<String>,
+ S: IntoIterator<Item = (K2, V2)>,
+ K2: AsRef<str>,
+ V2: Into<String>,
{
- let mut overwriting_options = HashMap::with_capacity(self.storage_options.len());
+ let mut hudi_opts: HashMap<String, String> = HashMap::new();
+ if let Some(opts) = read_options {
+ for (k, v) in &opts.hudi_options {
+ if Self::TABLE_OWNED_READ_KEYS.contains(&k.as_str()) {
+ continue;
+ }
+ hudi_opts.insert(k.clone(), v.clone());
+ }
+ }
+ for (k, v) in extra_hudi_overrides {
+ hudi_opts.insert(k.as_ref().to_string(), v.into());
+ }
+
+ let mut storage_opts: HashMap<String, String> =
+ HashMap::with_capacity(self.storage_options.len());
for (k, v) in self.storage_options.iter() {
- overwriting_options.insert(k.clone(), v.clone());
+ storage_opts.insert(k.clone(), v.clone());
}
- for (k, v) in options {
- overwriting_options.insert(k.as_ref().to_string(), v.into());
+ for (k, v) in extra_storage_overrides {
+ storage_opts.insert(k.as_ref().to_string(), v.into());
}
- FileGroupReader::new_with_configs_and_overwriting_options(
- self.hudi_configs.clone(),
- overwriting_options,
- )
+
+ FileGroupReader::new_with_overrides(self.hudi_configs.clone(), hudi_opts, storage_opts)
}
- /// Get all the latest records in the table.
+ /// Read-option keys the `Table` layer interprets directly. These are
+ /// excluded from the per-read overrides forwarded to the FG reader so a
+ /// stale value can't change physical-read behavior β see
+ /// [`Self::create_file_group_reader_with_options`].
+ const TABLE_OWNED_READ_KEYS: [&'static str; 4] = [
+ HudiReadConfig::QueryType.key_str(),
+ HudiReadConfig::AsOfTimestamp.key_str(),
+ HudiReadConfig::StartTimestamp.key_str(),
+ HudiReadConfig::EndTimestamp.key_str(),
+ ];
+
+ /// Read records, dispatching on `options.query_type`.
///
- /// # Arguments
- /// * `filters` - Partition filters to apply.
- pub async fn read_snapshot<I, S>(&self, filters: I) -> Result<Vec<RecordBatch>>
- where
- I: IntoIterator<Item = (S, S, S)>,
- S: AsRef<str>,
- {
- if let Some(timestamp) = self.timeline.get_latest_commit_timestamp_as_option() {
- let filters = from_str_tuples(filters)?;
- self.read_snapshot_internal(timestamp, &filters).await
- } else {
- Ok(Vec::new())
+ /// - [`QueryType::Snapshot`] reads at `options.as_of_timestamp` or the latest commit.
+ /// - [`QueryType::Incremental`] reads the change range
+ /// (`options.start_timestamp`, `options.end_timestamp`].
+ ///
+ /// `options.filters` drive partition pruning, file-level stats pruning (snapshot
+ /// only), and a row-level mask on every returned batch β see [`ReadOptions::filters`]
+ /// for the full breakdown. `options.hudi_options` override table-level Hudi configs
+ /// for this single read.
+ pub async fn read(&self, options: &ReadOptions) -> Result<Vec<RecordBatch>> {
+ match options.query_type()? {
+ QueryType::Snapshot => self.read_snapshot_inner(options).await,
+ QueryType::Incremental => self.read_incremental_inner(options).await,
}
}
- /// Get all the records in the table at a given timestamp.
- ///
- /// # Arguments
- /// * `timestamp` - The timestamp which records associated with.
- /// * `filters` - Partition filters to apply.
- pub async fn read_snapshot_as_of<I, S>(
- &self,
- timestamp: &str,
- filters: I,
- ) -> Result<Vec<RecordBatch>>
- where
- I: IntoIterator<Item = (S, S, S)>,
- S: AsRef<str>,
- {
- let timestamp = format_timestamp(timestamp, &self.timezone())?;
- let filters = from_str_tuples(filters)?;
- self.read_snapshot_internal(×tamp, &filters).await
+ async fn read_snapshot_inner(&self, options: &ReadOptions) -> Result<Vec<RecordBatch>> {
+ let Some(timestamp) = self.resolve_snapshot_timestamp(options)? else {
+ return Ok(Vec::new());
+ };
+ let file_slices = self
+ .get_file_slices_inner(×tamp, &options.filters)
+ .await?;
+ let fg_reader = self.create_file_group_reader_for_snapshot(options, ×tamp)?;
+ let fg_options = self.options_for_file_group(options);
+ let batches = futures::future::try_join_all(
+ file_slices
+ .iter()
+ .map(|f| fg_reader.read_file_slice(f, &fg_options)),
+ )
+ .await?;
+ Ok(batches)
}
- async fn read_snapshot_internal(
+ /// Build a `FileGroupReader` for a snapshot read with the resolved snapshot
+ /// bound injected as `EndTimestamp`. The Table-owned read keys in
+ /// `options.hudi_options` (notably any stray incremental `StartTimestamp`)
+ /// are dropped by [`Self::create_file_group_reader_with_options`].
+ fn create_file_group_reader_for_snapshot(
&self,
- timestamp: &str,
- filters: &[Filter],
- ) -> Result<Vec<RecordBatch>> {
- let file_slices = self.get_file_slices_internal(timestamp, filters).await?;
- let fg_reader = self.create_file_group_reader_with_options([(
- HudiReadConfig::FileGroupEndTimestamp,
- timestamp,
- )])?;
- let batches =
- futures::future::try_join_all(file_slices.iter().map(|f| fg_reader.read_file_slice(f)))
- .await?;
- Ok(batches)
+ options: &ReadOptions,
+ snapshot_timestamp: &str,
+ ) -> Result<FileGroupReader> {
+ self.create_file_group_reader_with_options(
+ Some(options),
+ [(HudiReadConfig::EndTimestamp, snapshot_timestamp.to_string())],
+ std::iter::empty::<(&str, &str)>(),
+ )
}
- /// Get records that were inserted or updated between the given timestamps.
- ///
- /// Records that were updated multiple times should have their latest states within
- /// the time span being returned.
- ///
- /// # Arguments
- /// * `start_timestamp` - Only records that were inserted or updated after this timestamp will be returned.
- /// * `end_timestamp` - If provided, only records that were inserted or updated before or at this timestamp will be returned.
- pub async fn read_incremental_records(
- &self,
- start_timestamp: &str,
- end_timestamp: Option<&str>,
- ) -> Result<Vec<RecordBatch>> {
- // If the end timestamp is not provided, use the latest file slice timestamp.
- // This is the request timestamp (commit_timestamp) for both v6 and v8+ tables.
- let Some(end_ts) =
- end_timestamp.or_else(|| self.timeline.get_latest_commit_timestamp_as_option())
- else {
+ async fn read_incremental_inner(&self, options: &ReadOptions) -> Result<Vec<RecordBatch>> {
+ let Some((start, end)) = self.resolve_incremental_range(options)? else {
return Ok(Vec::new());
};
-
- let timezone = self.timezone();
- let start_ts = format_timestamp(start_timestamp, &timezone)?;
- let end_ts = format_timestamp(end_ts, &timezone)?;
-
- // Use incremental API that reads from timeline commit metadata
let file_slices = self
- .get_file_slices_between_internal(&start_ts, &end_ts)
+ .get_file_slices_between_inner(&start, &end, &options.filters)
.await?;
- let fg_reader = self.create_file_group_reader_with_options([
- (HudiReadConfig::FileGroupStartTimestamp, start_ts.clone()),
- (HudiReadConfig::FileGroupEndTimestamp, end_ts),
- ])?;
+ let fg_reader = self.create_file_group_reader_with_options(
+ Some(options),
+ [
+ (HudiReadConfig::StartTimestamp, start),
+ (HudiReadConfig::EndTimestamp, end),
+ ],
+ std::iter::empty::<(&str, &str)>(),
+ )?;
+ let fg_options = self.options_for_file_group(options);
- let batches =
- futures::future::try_join_all(file_slices.iter().map(|f| fg_reader.read_file_slice(f)))
- .await?;
+ let batches = futures::future::try_join_all(
+ file_slices
+ .iter()
+ .map(|f| fg_reader.read_file_slice(f, &fg_options)),
+ )
+ .await?;
Ok(batches)
}
- /// Get the change-data-capture (CDC) records between the given timestamps.
+ /// Build the [`ReadOptions`] passed to `FileGroupReader` for a per-slice read,
+ /// stripping filters that target a partition column dropped from data files.
///
- /// The CDC records should reflect the records that were inserted, updated, and deleted
- /// between the timestamps.
- #[allow(dead_code)]
- async fn read_incremental_changes(
- &self,
- _start_timestamp: &str,
- _end_timestamp: Option<&str>,
- ) -> Result<Vec<RecordBatch>> {
- todo!("read_incremental_changes")
+ /// `FileGroupReader` validates filter columns strictly against the read batch
+ /// schema; when `hoodie.datasource.write.drop.partition.columns` is enabled,
+ /// partition columns aren't in parquet, so a partition filter would surface as
+ /// an error there. The partition pruner has already used those filters at
+ /// table level, so dropping them here is safe and avoids the false-positive.
+ fn options_for_file_group(&self, options: &ReadOptions) -> ReadOptions {
+ let drops: bool = self
+ .hudi_configs
+ .get_or_default(HudiTableConfig::DropsPartitionFields)
+ .into();
+ if !drops || options.filters.is_empty() {
+ return options.clone();
+ }
+ let partition_columns: Vec<String> = self
+ .hudi_configs
+ .get_or_default(HudiTableConfig::PartitionFields)
+ .into();
+ let mut applicable = options.clone();
+ applicable.filters = options
+ .filters
+ .iter()
+ .filter(|filter| !partition_columns.iter().any(|p| p == &filter.field))
+ .cloned()
+ .collect();
+ applicable
+ }
+
+ /// Resolve the snapshot timestamp from `options`: explicit `as_of_timestamp` if set,
+ /// otherwise the table's latest commit. Returns `None` only when the table has no
+ /// commits and no explicit timestamp was given.
+ fn resolve_snapshot_timestamp(&self, options: &ReadOptions) -> Result<Option<String>> {
+ if let Some(ts) = options.as_of_timestamp() {
+ return Ok(Some(format_timestamp(ts, &self.timezone())?));
+ }
+ Ok(self
+ .timeline
+ .get_latest_commit_timestamp_as_option()
+ .map(|s| s.to_string()))
+ }
+
+ /// Resolve the incremental change range `(start, end]` from `options`. `start`
+ /// defaults to [`EARLIEST_START_TIMESTAMP`]; `end` defaults to the latest commit.
+ ///
+ /// Returns `Ok(None)` only when no `end_timestamp` is provided AND the table has
+ /// no commits. Invalid timestamp strings propagate as `Err`.
+ fn resolve_incremental_range(&self, options: &ReadOptions) -> Result<Option<(String, String)>> {
+ let timezone = self.timezone();
+ let Some(end) = options
+ .end_timestamp()
+ .or_else(|| self.timeline.get_latest_commit_timestamp_as_option())
+ else {
+ return Ok(None);
+ };
+ let end = format_timestamp(end, &timezone)?;
+ let start = options
+ .start_timestamp()
+ .unwrap_or(EARLIEST_START_TIMESTAMP);
+ let start = format_timestamp(start, &timezone)?;
+ Ok(Some((start, end)))
}
// =========================================================================
// Streaming Read APIs
// =========================================================================
- /// Reads a file slice as a stream of record batches.
+ /// Streaming read; dispatches on `options.query_type`.
///
- /// This is the streaming version of reading a single file slice. It returns a stream
- /// that yields record batches as they are read, without loading all data into memory.
+ /// Snapshot streams batches as they are read from each file slice. Incremental
+ /// streaming is not yet supported and returns an `Unsupported` error.
///
- /// # Arguments
- /// * `file_slice` - The file slice to read.
- /// * `options` - Read options for configuring the read operation.
- ///
- /// # Returns
- /// A stream of record batches.
+ /// For MOR tables with log files, streaming falls back to a collect-and-merge that
+ /// yields the merged result as a single batch.
///
/// # Example
/// ```ignore
/// use futures::StreamExt;
/// use hudi::table::ReadOptions;
///
- /// let file_slices = table.get_file_slices(empty_filters()).await?;
- /// let options = ReadOptions::new().with_batch_size(4096);
- ///
- /// for file_slice in &file_slices {
- /// let mut stream = table.read_file_slice_stream(file_slice, &options).await?;
- /// while let Some(result) = stream.next().await {
- /// let batch = result?;
- /// // Process batch...
- /// }
+ /// let options = ReadOptions::new()
+ /// .with_filters([("city", "=", "san_francisco")])?
+ /// .with_batch_size(4096)?;
+ /// let mut stream = table.read_stream(&options).await?;
+ /// while let Some(result) = stream.next().await {
+ /// println!("Read {} rows", result?.num_rows());
/// }
/// ```
- pub async fn read_file_slice_stream(
+ pub async fn read_stream(
&self,
- file_slice: &FileSlice,
options: &ReadOptions,
) -> Result<futures::stream::BoxStream<'static, Result<RecordBatch>>> {
- use futures::stream;
-
- let timestamp = options
- .as_of_timestamp
- .as_deref()
- .or_else(|| self.timeline.get_latest_commit_timestamp_as_option());
-
- let Some(timestamp) = timestamp else {
- // No commit timestamp means empty table - return empty stream (consistent with read_snapshot)
- return Ok(Box::pin(stream::empty()));
- };
-
- let fg_reader = self.create_file_group_reader_with_options([(
- HudiReadConfig::FileGroupEndTimestamp,
- timestamp,
- )])?;
-
- fg_reader.read_file_slice_stream(file_slice, options).await
+ match options.query_type()? {
+ QueryType::Snapshot => self.read_snapshot_stream_inner(options).await,
+ QueryType::Incremental => Err(CoreError::Unsupported(
+ "Streaming for incremental queries is not yet supported".to_string(),
+ )),
+ }
}
- /// Reads the table snapshot as a stream of record batches.
- ///
- /// This is the streaming version of [Table::read_snapshot]. Instead of returning
- /// all batches at once, it returns a stream that yields record batches as they
- /// are read from the underlying file slices.
- ///
- /// # Arguments
- /// * `options` - Read options including partition filters, batch size, and projection.
- ///
- /// # Returns
- /// A stream of record batches from all file slices. Errors from individual file
- /// slices are propagated through the stream.
- ///
- /// # Limitations
- ///
- /// - The `row_predicate` field in [ReadOptions] is not yet implemented for streaming reads.
- /// - For MOR tables with log files, streaming falls back to the collect-and-merge approach.
- ///
- /// # Example
- /// ```ignore
- /// use futures::StreamExt;
- /// use hudi::table::ReadOptions;
- ///
- /// let options = ReadOptions::new()
- /// .with_filters([("city", "=", "san_francisco")])
- /// .with_batch_size(4096);
- /// let mut stream = table.read_snapshot_stream(&options).await?;
- ///
- /// while let Some(result) = stream.next().await {
- /// let batch = result?;
- /// println!("Read {} rows", batch.num_rows());
- /// }
- /// ```
- pub async fn read_snapshot_stream(
+ async fn read_snapshot_stream_inner(
&self,
options: &ReadOptions,
) -> Result<futures::stream::BoxStream<'static, Result<RecordBatch>>> {
use futures::stream::{self, StreamExt};
- let Some(timestamp) = self.timeline.get_latest_commit_timestamp_as_option() else {
+ let Some(timestamp) = self.resolve_snapshot_timestamp(options)? else {
return Ok(Box::pin(stream::empty()));
};
- let filters: Vec<Filter> = options
- .partition_filters
- .iter()
- .map(|(f, o, v)| Filter::try_from((f.as_str(), o.as_str(), v.as_str())))
- .collect::<Result<Vec<_>>>()?;
- let file_slices = self.get_file_slices_internal(timestamp, &filters).await?;
+ let file_slices = self
+ .get_file_slices_inner(×tamp, &options.filters)
+ .await?;
if file_slices.is_empty() {
return Ok(Box::pin(stream::empty()));
}
- let fg_reader = self.create_file_group_reader_with_options([(
- HudiReadConfig::FileGroupEndTimestamp,
- timestamp,
- )])?;
-
- // Extract options to pass to each file slice read.
- let batch_size = options.batch_size;
- let projection = options.projection.clone();
- let row_predicate = options.row_predicate.clone();
+ let fg_reader = self.create_file_group_reader_for_snapshot(options, ×tamp)?;
+
+ // Extract per-batch options. Keep `filters` so they apply at row-level too β
+ // the upstream pruning already used them at file/partition level; applying at
+ // row-level closes the gap for non-partition column filters. Strip filters
+ // on dropped partition columns so they don't trigger FGR validation errors.
+ let fg_options_template = self.options_for_file_group(options);
+ let projection = fg_options_template.projection.clone();
+ let row_filters = fg_options_template.filters.clone();
+ // Carry batch_size in hudi_options if set; everything else (timestamps,
+ // query_type) is irrelevant to the per-slice FG-reader read.
+ let mut per_slice_hudi_options: HashMap<String, String> = HashMap::new();
+ if let Some(bs) = fg_options_template.batch_size()? {
+ per_slice_hudi_options.insert(
+ HudiReadConfig::StreamBatchSize.as_ref().to_string(),
+ bs.to_string(),
+ );
+ }
let streams_iter = file_slices.into_iter().map(move |file_slice| {
let fg_reader = fg_reader.clone();
let projection = projection.clone();
- let row_predicate = row_predicate.clone();
+ let row_filters = row_filters.clone();
let options = ReadOptions {
- partition_filters: vec![],
+ filters: row_filters,
projection,
- row_predicate,
- batch_size,
- as_of_timestamp: None,
+ hudi_options: per_slice_hudi_options.clone(),
};
async move {
fg_reader
@@ -1008,9 +941,8 @@ mod tests {
PrecombineField, RecordKeyFields, TableName, TableType, TableVersion,
TimelineLayoutVersion, TimelineTimezone,
};
- use crate::config::util::{empty_filters, empty_options};
+ use crate::config::util::empty_options;
use crate::error::CoreError;
- use crate::file_group::FileGroup;
use crate::metadata::meta_field::MetaField;
use crate::storage::Storage;
use crate::storage::util::join_url_segments;
@@ -1047,7 +979,8 @@ mod tests {
) -> Result<Vec<String>> {
let mut file_paths = Vec::new();
let base_url = table.base_url();
- for f in table.get_file_slices(filters.to_vec()).await? {
+ let options = ReadOptions::new().with_filters(filters.iter().copied())?;
+ for f in table.get_file_slices(&options).await? {
let relative_path = f.base_file_relative_path()?;
let file_url = join_url_segments(&base_url, &[relative_path.as_str()])?;
file_paths.push(file_url.to_string());
@@ -1435,10 +1368,12 @@ mod tests {
let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
let hudi_table = Table::new(base_url.path()).await.unwrap();
let batches = hudi_table
- .create_file_group_reader_with_options(empty_options())
+ .create_file_group_reader_with_options(None, empty_options(), empty_options())
.unwrap()
- .read_file_slice_by_base_file_path(
+ .read_file_slice_from_paths(
"a079bdb3-731c-4894-b855-abfcd6921007-0_0-203-274_20240418173551906.parquet",
+ Vec::<&str>::new(),
+ &ReadOptions::new(),
)
.await
.unwrap();
@@ -1452,7 +1387,7 @@ mod tests {
let hudi_table = Table::new(base_url.path()).await.unwrap();
let latest_timestamp = hudi_table.timeline.get_latest_commit_timestamp().unwrap();
- let snapshot_batches = hudi_table.read_snapshot(empty_filters()).await.unwrap();
+ let snapshot_batches = hudi_table.read(&ReadOptions::new()).await.unwrap();
assert!(!snapshot_batches.is_empty());
let snapshot_rows = snapshot_batches
.iter()
@@ -1461,7 +1396,7 @@ mod tests {
assert!(snapshot_rows > 0);
let as_of_batches = hudi_table
- .read_snapshot_as_of(&latest_timestamp, empty_filters())
+ .read(&ReadOptions::new().with_as_of_timestamp(&latest_timestamp))
.await
.unwrap();
assert!(!as_of_batches.is_empty());
@@ -1474,168 +1409,211 @@ mod tests {
let base_url = SampleTable::V6Empty.url_to_cow();
let hudi_table = Table::new(base_url.path()).await.unwrap();
- let snapshot_batches = hudi_table.read_snapshot(empty_filters()).await.unwrap();
+ let snapshot_batches = hudi_table.read(&ReadOptions::new()).await.unwrap();
assert!(snapshot_batches.is_empty());
let incremental_batches = hudi_table
- .read_incremental_records(EARLIEST_START_TIMESTAMP, None)
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(EARLIEST_START_TIMESTAMP),
+ )
.await
.unwrap();
assert!(incremental_batches.is_empty());
- let mut snapshot_stream = hudi_table
- .read_snapshot_stream(&ReadOptions::new())
- .await
- .unwrap();
+ let mut snapshot_stream = hudi_table.read_stream(&ReadOptions::new()).await.unwrap();
assert!(snapshot_stream.next().await.is_none());
-
- let file_group = FileGroup::new_with_base_file_name(
- "a079bdb3-731c-4894-b855-abfcd6921007-0_0-203-274_20240418173551906.parquet",
- "",
- )
- .unwrap();
- let file_slice = file_group
- .get_file_slice_as_of("20240418173551906")
- .unwrap()
- .clone();
- let mut file_slice_stream = hudi_table
- .read_file_slice_stream(&file_slice, &ReadOptions::new())
- .await
- .unwrap();
- assert!(file_slice_stream.next().await.is_none());
- }
-
- #[tokio::test]
- async fn hudi_table_read_file_slice_stream_reads_batches() {
- use futures::TryStreamExt;
-
- let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
- let hudi_table = Table::new(base_url.path()).await.unwrap();
- let file_slices = hudi_table.get_file_slices(empty_filters()).await.unwrap();
- let file_slice = file_slices.first().unwrap().clone();
- let options = ReadOptions::new().with_batch_size(2);
-
- let stream = hudi_table
- .read_file_slice_stream(&file_slice, &options)
- .await
- .unwrap();
- let batches = stream.try_collect::<Vec<_>>().await.unwrap();
- assert!(!batches.is_empty());
}
#[tokio::test]
- async fn hudi_table_read_snapshot_stream_returns_batches_with_options() {
+ async fn hudi_table_read_snapshot_stream_returns_batches_with_options() -> Result<()> {
use futures::TryStreamExt;
let base_url = SampleTable::V6SimplekeygenNonhivestyle.url_to_cow();
let hudi_table = Table::new(base_url.path()).await.unwrap();
let options = ReadOptions::new()
- .with_filters([("byteField", ">=", "10")])
+ .with_filters([("byteField", ">=", "10")])?
.with_projection(["id"])
- .with_batch_size(2);
+ .with_batch_size(2)?;
- let stream = hudi_table.read_snapshot_stream(&options).await.unwrap();
+ let stream = hudi_table.read_stream(&options).await.unwrap();
let batches = stream.try_collect::<Vec<_>>().await.unwrap();
assert!(!batches.is_empty());
assert!(batches[0].column_by_name("id").is_some());
+ Ok(())
}
#[tokio::test]
- async fn hudi_table_read_snapshot_stream_returns_empty_when_no_file_slices_match_filters() {
+ async fn hudi_table_read_snapshot_stream_returns_empty_when_no_file_slices_match_filters()
+ -> Result<()> {
use futures::StreamExt;
let base_url = SampleTable::V6SimplekeygenNonhivestyle.url_to_cow();
let hudi_table = Table::new(base_url.path()).await.unwrap();
- let options = ReadOptions::new().with_filters([("byteField", "=", "999")]);
+ let options = ReadOptions::new().with_filters([("byteField", "=", "999")])?;
- let mut stream = hudi_table.read_snapshot_stream(&options).await.unwrap();
+ let mut stream = hudi_table.read_stream(&options).await.unwrap();
assert!(stream.next().await.is_none());
+ Ok(())
}
#[tokio::test]
- async fn hudi_table_read_incremental_records() {
+ async fn hudi_table_read_with_incremental_query_type() {
let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
let hudi_table = Table::new(base_url.path()).await.unwrap();
let batches = hudi_table
- .read_incremental_records(EARLIEST_START_TIMESTAMP, None)
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(EARLIEST_START_TIMESTAMP),
+ )
.await
.unwrap();
assert!(!batches.is_empty());
}
#[tokio::test]
- async fn empty_hudi_table_get_file_slices_splits() {
- let base_url = SampleTable::V6Empty.url_to_cow();
-
+ async fn hudi_table_read_dispatches_on_query_type() {
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
let hudi_table = Table::new(base_url.path()).await.unwrap();
- let file_slices_splits = hudi_table
- .get_file_slices_splits(2, empty_filters())
+
+ // Default: Snapshot.
+ let snapshot = hudi_table.read(&ReadOptions::new()).await.unwrap();
+ assert!(!snapshot.is_empty());
+
+ // Explicit Incremental returns the same shape as the dedicated shortcut.
+ let incremental = hudi_table
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(EARLIEST_START_TIMESTAMP),
+ )
+ .await
+ .unwrap();
+ assert!(!incremental.is_empty());
+
+ // Shortcuts override query_type set on the input options.
+ let forced_snapshot = hudi_table
+ .read(&ReadOptions::new().with_query_type(QueryType::Incremental))
.await
.unwrap();
- assert!(file_slices_splits.is_empty());
+ assert!(!forced_snapshot.is_empty());
}
#[tokio::test]
- async fn hudi_table_get_file_slices_splits() {
- let base_url = SampleTable::V6SimplekeygenNonhivestyle.url_to_cow();
+ async fn hudi_table_read_stream_errors_on_incremental() {
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await.unwrap();
+ let result = hudi_table
+ .read_stream(&ReadOptions::new().with_query_type(QueryType::Incremental))
+ .await;
+ match result {
+ Ok(_) => panic!("incremental streaming must error"),
+ Err(e) => {
+ assert!(matches!(e, CoreError::Unsupported(_)));
+ assert!(e.to_string().contains("not yet supported"));
+ }
+ }
+ }
+ #[tokio::test]
+ async fn hudi_table_get_file_slices_dispatches_on_query_type() {
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
let hudi_table = Table::new(base_url.path()).await.unwrap();
- let file_slices_splits = hudi_table
- .get_file_slices_splits(2, empty_filters())
+
+ let snapshot_slices = hudi_table
+ .get_file_slices(&ReadOptions::new())
+ .await
+ .unwrap();
+ let incremental_slices = hudi_table
+ .get_file_slices(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(EARLIEST_START_TIMESTAMP),
+ )
.await
.unwrap();
- assert_eq!(file_slices_splits.len(), 2);
- assert_eq!(file_slices_splits[0].len(), 2);
- assert_eq!(file_slices_splits[1].len(), 1);
+ assert!(!snapshot_slices.is_empty());
+ assert!(!incremental_slices.is_empty());
+ }
+
+ #[tokio::test]
+ async fn read_with_invalid_as_of_timestamp_errors() {
+ // `as_of_timestamp` is parsed via `format_timestamp` before any IO. A
+ // malformed value must surface as `TimestampParsingError` rather than
+ // silently being treated as "latest" or sliced into a no-op result.
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await.unwrap();
+ let options = ReadOptions::new().with_as_of_timestamp("not-a-timestamp");
+
+ let snapshot_err = hudi_table.read(&options).await.unwrap_err();
+ assert!(
+ matches!(snapshot_err, CoreError::TimestampParsingError(_)),
+ "expected TimestampParsingError, got: {snapshot_err}"
+ );
+
+ let stream_result = hudi_table.read_stream(&options).await;
+ match stream_result {
+ Ok(_) => panic!("read_stream must propagate the parse error synchronously"),
+ Err(e) => assert!(
+ matches!(e, CoreError::TimestampParsingError(_)),
+ "expected TimestampParsingError on stream path, got: {e}"
+ ),
+ }
}
#[tokio::test]
- async fn hudi_table_get_file_slices_splits_as_of_timestamps() {
+ async fn hudi_table_read_options_hudi_options_plumbed_to_reader() {
+ // Per-read hudi_options should override table-level configs without
+ // mutating the table. Here we set the per-read StartTimestamp
+ // via hudi_options on a snapshot read; the read should still succeed.
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await.unwrap();
+ let options =
+ ReadOptions::new().with_hudi_option(HudiReadConfig::StartTimestamp.as_ref(), "0");
+ let batches = hudi_table.read(&options).await.unwrap();
+ assert!(!batches.is_empty());
+ }
+
+ #[tokio::test]
+ async fn hudi_table_get_file_slices_as_of_replacecommit() {
+ // Insert-overwrite-table replacecommit: as_of before vs at the replacecommit
+ // returns different slice sets. Splitter behavior is covered by
+ // util::collection::split_into_chunks tests.
let base_url = SampleTable::V6SimplekeygenNonhivestyleOverwritetable.url_to_mor_parquet();
let hudi_table = Table::new(base_url.path()).await.unwrap();
- // before replacecommit (insert overwrite table)
+ // before replacecommit
let second_latest_timestamp = "20250121000656060";
- let file_slices_splits = hudi_table
- .get_file_slices_splits_as_of(2, second_latest_timestamp, empty_filters())
+ let file_slices = hudi_table
+ .get_file_slices(&ReadOptions::new().with_as_of_timestamp(second_latest_timestamp))
.await
.unwrap();
- assert_eq!(file_slices_splits.len(), 2);
- assert_eq!(file_slices_splits[0].len(), 2);
- assert_eq!(file_slices_splits[1].len(), 1);
- let file_slices = file_slices_splits
+ assert_eq!(file_slices.len(), 3);
+ let p10: Vec<_> = file_slices
.iter()
- .flatten()
.filter(|f| f.partition_path == "10")
- .collect::<Vec<_>>();
- assert_eq!(
- file_slices.len(),
- 1,
- "Partition 10 should have 1 file slice"
- );
- let file_slice = file_slices[0];
+ .collect();
+ assert_eq!(p10.len(), 1, "Partition 10 should have 1 file slice");
+ let file_slice = p10[0];
assert_eq!(
file_slice.base_file.file_name(),
"92e64357-e4d1-4639-a9d3-c3535829d0aa-0_1-53-79_20250121000647668.parquet"
);
- assert_eq!(
- file_slice.log_files.len(),
- 1,
- "File slice should have 1 log file"
- );
+ assert_eq!(file_slice.log_files.len(), 1);
assert_eq!(
file_slice.log_files.iter().next().unwrap().file_name(),
".92e64357-e4d1-4639-a9d3-c3535829d0aa-0_20250121000647668.log.1_0-73-101"
);
- // as of replacecommit (insert overwrite table)
+ // as of replacecommit
let latest_timestamp = "20250121000702475";
- let file_slices_splits = hudi_table
- .get_file_slices_splits_as_of(2, latest_timestamp, empty_filters())
+ let file_slices = hudi_table
+ .get_file_slices(&ReadOptions::new().with_as_of_timestamp(latest_timestamp))
.await
.unwrap();
- assert_eq!(file_slices_splits.len(), 1);
- assert_eq!(file_slices_splits[0].len(), 1);
+ assert_eq!(file_slices.len(), 1);
}
#[tokio::test]
@@ -1643,7 +1621,10 @@ mod tests {
let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
let hudi_table = Table::new(base_url.path()).await.unwrap();
- let file_slices = hudi_table.get_file_slices(empty_filters()).await.unwrap();
+ let file_slices = hudi_table
+ .get_file_slices(&ReadOptions::new())
+ .await
+ .unwrap();
assert_eq!(
file_slices
.iter()
@@ -1655,7 +1636,7 @@ mod tests {
// as of the latest timestamp
let hudi_table = Table::new(base_url.path()).await.unwrap();
let file_slices = hudi_table
- .get_file_slices_as_of("20240418173551906", empty_filters())
+ .get_file_slices(&ReadOptions::new().with_as_of_timestamp("20240418173551906"))
.await
.unwrap();
assert_eq!(
@@ -1671,7 +1652,7 @@ mod tests {
.await
.unwrap();
let file_slices = hudi_table
- .get_file_slices_as_of("20240418173551905", empty_filters())
+ .get_file_slices(&ReadOptions::new().with_as_of_timestamp("20240418173551905"))
.await
.unwrap();
assert_eq!(
@@ -1687,7 +1668,7 @@ mod tests {
.await
.unwrap();
let file_slices = hudi_table
- .get_file_slices_as_of("19700101000000", empty_filters())
+ .get_file_slices(&ReadOptions::new().with_as_of_timestamp("19700101000000"))
.await
.unwrap();
assert_eq!(
@@ -1700,22 +1681,30 @@ mod tests {
}
#[tokio::test]
- async fn empty_hudi_table_get_file_slices_between_timestamps() {
+ async fn empty_hudi_table_get_file_slices_incremental() {
let base_url = SampleTable::V6Empty.url_to_cow();
let hudi_table = Table::new(base_url.path()).await.unwrap();
let file_slices = hudi_table
- .get_file_slices_between(Some(EARLIEST_START_TIMESTAMP), None)
+ .get_file_slices(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(EARLIEST_START_TIMESTAMP),
+ )
.await
.unwrap();
assert!(file_slices.is_empty())
}
#[tokio::test]
- async fn hudi_table_get_file_slices_between_timestamps() {
+ async fn hudi_table_get_file_slices_incremental() {
let base_url = SampleTable::V6SimplekeygenNonhivestyleOverwritetable.url_to_mor_parquet();
let hudi_table = Table::new(base_url.path()).await.unwrap();
let mut file_slices = hudi_table
- .get_file_slices_between(None, Some("20250121000656060"))
+ .get_file_slices(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_end_timestamp("20250121000656060"),
+ )
.await
.unwrap();
assert_eq!(file_slices.len(), 3);
@@ -1766,62 +1755,6 @@ mod tests {
assert_eq!(m2.num_records, 458);
}
- #[tokio::test]
- async fn empty_hudi_table_get_file_slices_splits_between() {
- let base_url = SampleTable::V6Empty.url_to_cow();
- let hudi_table = Table::new(base_url.path()).await.unwrap();
- let file_slices_splits = hudi_table
- .get_file_slices_splits_between(2, Some(EARLIEST_START_TIMESTAMP), None)
- .await
- .unwrap();
- assert!(file_slices_splits.is_empty())
- }
-
- #[tokio::test]
- async fn hudi_table_get_file_slices_splits_between() {
- let base_url = SampleTable::V6SimplekeygenNonhivestyleOverwritetable.url_to_mor_parquet();
- let hudi_table = Table::new(base_url.path()).await.unwrap();
- let file_slices_splits = hudi_table
- .get_file_slices_splits_between(2, None, Some("20250121000656060"))
- .await
- .unwrap();
-
- assert_eq!(file_slices_splits.len(), 2);
- let total_file_slices: usize = file_slices_splits.iter().map(|split| split.len()).sum();
- assert_eq!(total_file_slices, 3);
- assert_eq!(file_slices_splits[0].len(), 2);
- assert_eq!(file_slices_splits[1].len(), 1);
- }
-
- #[tokio::test]
- async fn hudi_table_get_file_slices_splits_between_with_single_split() {
- let base_url = SampleTable::V6SimplekeygenNonhivestyleOverwritetable.url_to_mor_parquet();
- let hudi_table = Table::new(base_url.path()).await.unwrap();
- let file_slices_splits = hudi_table
- .get_file_slices_splits_between(1, None, Some("20250121000656060"))
- .await
- .unwrap();
-
- // Should have 1 split with all 3 file slices
- assert_eq!(file_slices_splits.len(), 1);
- assert_eq!(file_slices_splits[0].len(), 3);
- }
-
- #[tokio::test]
- async fn hudi_table_get_file_slices_splits_between_with_many_splits() {
- let base_url = SampleTable::V6SimplekeygenNonhivestyleOverwritetable.url_to_mor_parquet();
- let hudi_table = Table::new(base_url.path()).await.unwrap();
- let file_slices_splits = hudi_table
- .get_file_slices_splits_between(10, None, Some("20250121000656060"))
- .await
- .unwrap();
-
- assert_eq!(file_slices_splits.len(), 3);
- for split in &file_slices_splits {
- assert_eq!(split.len(), 1);
- }
- }
-
#[tokio::test]
async fn hudi_table_get_file_paths_for_simple_keygen_non_hive_style() {
let base_url = SampleTable::V6SimplekeygenNonhivestyle.url_to_cow();
@@ -2018,7 +1951,7 @@ mod tests {
assert_eq!(cloned.table_type(), table.table_type());
// Clone shares the cached metadata table via Arc<OnceCell>
- let file_slices = cloned.get_file_slices(empty_filters()).await.unwrap();
+ let file_slices = cloned.get_file_slices(&ReadOptions::new()).await.unwrap();
assert!(!file_slices.is_empty());
// compute_table_stats works on cloned table
@@ -2037,7 +1970,7 @@ mod tests {
.unwrap();
assert!(table.is_metadata_table_enabled());
- let file_slices = table.get_file_slices(empty_filters()).await.unwrap();
+ let file_slices = table.get_file_slices(&ReadOptions::new()).await.unwrap();
assert!(!file_slices.is_empty());
}
@@ -2047,10 +1980,10 @@ mod tests {
let table = Table::new(base_url.path()).await.unwrap();
assert!(table.is_metadata_table_enabled());
- // This exercises the MDT code path in get_file_slices_internal:
+ // This exercises the MDT code path in get_file_slices_inner:
// metadata table init, fetch_files_partition_records, and
// fs_view's load_file_groups with estimator
- let file_slices = table.get_file_slices(empty_filters()).await.unwrap();
+ let file_slices = table.get_file_slices(&ReadOptions::new()).await.unwrap();
assert!(!file_slices.is_empty());
// Verify file metadata is populated from MDT with estimated stats
@@ -2078,7 +2011,7 @@ mod tests {
.unwrap();
assert!(!records.is_empty());
- let file_slices = table.get_file_slices(empty_filters()).await.unwrap();
+ let file_slices = table.get_file_slices(&ReadOptions::new()).await.unwrap();
assert!(!file_slices.is_empty());
}
}
diff --git a/crates/core/src/table/partition.rs b/crates/core/src/table/partition.rs
index 0445eac8..190f136f 100644
--- a/crates/core/src/table/partition.rs
+++ b/crates/core/src/table/partition.rs
@@ -451,7 +451,7 @@ mod tests {
fn test_partition_filter_try_from_valid() {
let schema = create_test_schema();
let filter = Filter {
- field_name: "date".to_string(),
+ field: "date".to_string(),
operator: ExprOperator::Eq,
values: vec!["2023-01-01".to_string()],
};
@@ -472,7 +472,7 @@ mod tests {
fn test_partition_filter_try_from_invalid_field() {
let schema = create_test_schema();
let filter = Filter {
- field_name: "invalid_field".to_string(),
+ field: "invalid_field".to_string(),
operator: ExprOperator::Eq,
values: vec!["2023-01-01".to_string()],
};
@@ -490,7 +490,7 @@ mod tests {
fn test_partition_filter_try_from_invalid_value() {
let schema = create_test_schema();
let filter = Filter {
- field_name: "count".to_string(),
+ field: "count".to_string(),
operator: ExprOperator::Eq,
values: vec!["not_a_number".to_string()],
};
@@ -535,7 +535,7 @@ mod tests {
]);
let user_filter = Filter {
- field_name: "ts_str".to_string(),
+ field: "ts_str".to_string(),
operator: ExprOperator::Gte,
values: vec!["2023-04-15T12:00:00.000Z".to_string()],
};
@@ -548,7 +548,7 @@ mod tests {
.unwrap();
assert_eq!(transformed.len(), 1);
- assert_eq!(transformed[0].field_name, MetaField::PartitionPath.as_ref());
+ assert_eq!(transformed[0].field, MetaField::PartitionPath.as_ref());
assert_eq!(transformed[0].operator, ExprOperator::Gte);
assert_eq!(transformed[0].values[0], "year=2023/month=04/day=15");
@@ -566,7 +566,7 @@ mod tests {
// 2024-01-25 00:00:00 UTC = 1706140800 seconds
let user_filter = Filter {
- field_name: "event_time".to_string(),
+ field: "event_time".to_string(),
operator: ExprOperator::Eq,
values: vec!["1706140800".to_string()],
};
@@ -579,7 +579,7 @@ mod tests {
.unwrap();
assert_eq!(transformed.len(), 1);
- assert_eq!(transformed[0].field_name, MetaField::PartitionPath.as_ref());
+ assert_eq!(transformed[0].field, MetaField::PartitionPath.as_ref());
assert_eq!(transformed[0].values[0], "2024/01/25");
// v8 detection via keygenerator.type=TIMESTAMP (no keygenerator.class)
@@ -596,7 +596,7 @@ mod tests {
]);
let user_filter = Filter {
- field_name: "ts_str".to_string(),
+ field: "ts_str".to_string(),
operator: ExprOperator::Eq,
values: vec!["2023-04-15T12:00:00.000Z".to_string()],
};
@@ -626,7 +626,7 @@ mod tests {
let partition_schema = Schema::new(vec![Field::new("region", DataType::Utf8, false)]);
let user_filter = Filter {
- field_name: "region".to_string(),
+ field: "region".to_string(),
operator: ExprOperator::Eq,
values: vec!["us-west".to_string()],
};
@@ -639,7 +639,7 @@ mod tests {
.unwrap();
assert_eq!(transformed.len(), 1);
- assert_eq!(transformed[0].field_name, user_filter.field_name);
+ assert_eq!(transformed[0].field, user_filter.field);
assert_eq!(transformed[0].values[0], user_filter.values[0]);
}
@@ -668,7 +668,7 @@ mod tests {
)]);
let user_filter = Filter {
- field_name: "ts".to_string(),
+ field: "ts".to_string(),
operator: ExprOperator::Gte,
values: vec!["2024-01-15T00:00:00Z".to_string()],
};
@@ -703,7 +703,7 @@ mod tests {
]);
let user_filter = Filter {
- field_name: "ts".to_string(),
+ field: "ts".to_string(),
operator: ExprOperator::Eq,
values: vec!["2024-01-15T00:00:00Z".to_string()],
};
diff --git a/crates/core/src/table/read_options.rs b/crates/core/src/table/read_options.rs
index f2ca99df..46296b0b 100644
--- a/crates/core/src/table/read_options.rs
+++ b/crates/core/src/table/read_options.rs
@@ -16,65 +16,75 @@
* specific language governing permissions and limitations
* under the License.
*/
-//! Read options for streaming reads.
+//! Read options shared by all read APIs (eager and streaming).
-use std::sync::Arc;
+use std::collections::HashMap;
+use std::str::FromStr;
-use arrow_array::{BooleanArray, RecordBatch};
+use crate::config::error::ConfigError;
+use crate::config::read::HudiReadConfig;
+pub use crate::config::read::QueryType;
+use crate::expr::filter::{Filter, from_str_tuples};
-/// A row-level predicate function for filtering records.
+/// Options for all Hudi read APIs (snapshot, time-travel, incremental, etc).
///
-/// Uses `Arc` instead of `Box` to allow cloning for async streaming contexts.
-pub type RowPredicate = Arc<dyn Fn(&RecordBatch) -> crate::Result<BooleanArray> + Send + Sync>;
-
-/// A partition filter tuple: (field_name, operator, value).
-/// Example: ("city", "=", "san_francisco")
-pub type PartitionFilter = (String, String, String);
-
-/// Options for reading file slices with streaming APIs.
-///
-/// This struct provides configuration for:
-/// - Partition filters (filtering partitions)
-/// - Column projection (which columns to read)
-/// - Row-level predicates (filtering rows)
-/// - Batch size control (rows per batch)
-/// - Time travel (as-of timestamp)
+/// `filters` and `projection` are typed because they don't have a Hudi config equivalent.
+/// Every other knob β query type, timestamps, batch size, ad-hoc Hudi configs β is stored
+/// in `hudi_options` under its `HudiReadConfig` key. Builder methods like
+/// [`Self::with_query_type`], [`Self::with_as_of_timestamp`], [`Self::with_start_timestamp`],
+/// [`Self::with_end_timestamp`], and [`Self::with_batch_size`] are convenience wrappers that
+/// insert into `hudi_options` under the right key. Typed accessors
+/// ([`Self::query_type`], [`Self::as_of_timestamp`], etc.) read from the bag.
///
-/// # Streaming Support
-///
-/// All options are supported in streaming APIs:
-/// - `batch_size` controls the number of rows per batch
-/// - `partition_filters` prunes partitions before reading
-/// - `projection` pushes column selection to the parquet reader level
-/// - `row_predicate` filters rows after reading each batch
-/// - `as_of_timestamp` enables time travel queries
+/// One source of truth: the bag. Builders + accessors are the ergonomic API on top.
///
/// # Example
///
/// ```ignore
-/// use hudi::table::ReadOptions;
+/// use hudi::table::{ReadOptions, QueryType};
///
/// let options = ReadOptions::new()
-/// .with_filters([("city", "=", "san_francisco")])
+/// .with_filters([("city", "=", "san_francisco")])?
/// .with_projection(["id", "name", "city"])
-/// .with_batch_size(4096);
+/// .with_batch_size(4096)?;
+///
+/// // Time-travel snapshot
+/// let options = ReadOptions::new().with_as_of_timestamp("20240101000000000");
+///
+/// // Incremental read between two commits
+/// let options = ReadOptions::new()
+/// .with_query_type(QueryType::Incremental)
+/// .with_start_timestamp("20240101000000000")
+/// .with_end_timestamp("20240201000000000");
+///
+/// // Read-optimized snapshot (skip log files)
+/// let options = ReadOptions::new()
+/// .with_hudi_option("hoodie.read.use.read_optimized.mode", "true");
/// ```
-#[derive(Default)]
+#[derive(Clone, Debug, Default)]
pub struct ReadOptions {
- /// Partition filters. Each filter is a tuple of (field, operator, value).
- pub partition_filters: Vec<PartitionFilter>,
+ /// Column filters, parsed and cardinality-validated at construction.
+ /// `field` is any column name (partition or data).
+ ///
+ /// Filters drive both **pruning** and **row-level filtering**:
+ /// - When the field is a **partition column**, the filter prunes whole partitions
+ /// for both snapshot and incremental queries.
+ /// - When the field is a **data column** with column statistics (min/max) in the
+ /// metadata table, the filter prunes whole files **for snapshot queries only**.
+ /// Incremental file planning currently does partition pruning only β data-column
+ /// filters apply at the row-level mask but do not prune files.
+ /// - All filters are applied as a row-level mask after reading, so callers
+ /// always get only rows that match regardless of the planning path.
+ pub filters: Vec<Filter>,
/// Column names to project (select). If None, all columns are read.
pub projection: Option<Vec<String>>,
- /// Row-level filter predicate. Applied after reading each batch.
- pub row_predicate: Option<RowPredicate>,
-
- /// Target number of rows per batch.
- pub batch_size: Option<usize>,
-
- /// Timestamp for time travel queries (as-of).
- pub as_of_timestamp: Option<String>,
+ /// Resolved bag of Hudi configs for this read. Populated by typed builders
+ /// (e.g. [`Self::with_query_type`], [`Self::with_start_timestamp`]) and by
+ /// [`Self::with_hudi_option`] for any keys that don't have a typed builder.
+ /// Keys are `HudiReadConfig` strings (`hoodie.*`).
+ pub hudi_options: HashMap<String, String>,
}
impl ReadOptions {
@@ -83,28 +93,86 @@ impl ReadOptions {
Self::default()
}
- /// Sets partition filters.
- ///
- /// # Arguments
- /// * `filters` - Partition filters as tuples of (field, operator, value)
- pub fn with_filters<I, S1, S2, S3>(mut self, filters: I) -> Self
+ // ---- typed builders (insert into hudi_options under the right key) ----
+
+ /// Sets the query type. Stored as [`HudiReadConfig::QueryType`].
+ pub fn with_query_type(mut self, query_type: QueryType) -> Self {
+ self.hudi_options.insert(
+ HudiReadConfig::QueryType.as_ref().to_string(),
+ query_type.as_ref().to_string(),
+ );
+ self
+ }
+
+ /// Sets the as-of timestamp for snapshot/time-travel queries.
+ /// Stored as [`HudiReadConfig::AsOfTimestamp`].
+ pub fn with_as_of_timestamp<S: AsRef<str>>(mut self, timestamp: S) -> Self {
+ self.hudi_options.insert(
+ HudiReadConfig::AsOfTimestamp.as_ref().to_string(),
+ timestamp.as_ref().to_string(),
+ );
+ self
+ }
+
+ /// Sets the lower-bound timestamp (exclusive) for incremental queries.
+ /// Stored as [`HudiReadConfig::StartTimestamp`].
+ pub fn with_start_timestamp<S: AsRef<str>>(mut self, timestamp: S) -> Self {
+ self.hudi_options.insert(
+ HudiReadConfig::StartTimestamp.as_ref().to_string(),
+ timestamp.as_ref().to_string(),
+ );
+ self
+ }
+
+ /// Sets the upper-bound timestamp (inclusive) for incremental queries.
+ /// Stored as [`HudiReadConfig::EndTimestamp`].
+ pub fn with_end_timestamp<S: AsRef<str>>(mut self, timestamp: S) -> Self {
+ self.hudi_options.insert(
+ HudiReadConfig::EndTimestamp.as_ref().to_string(),
+ timestamp.as_ref().to_string(),
+ );
+ self
+ }
+
+ /// Sets the target batch size (rows per batch).
+ /// Stored as [`HudiReadConfig::StreamBatchSize`]. Errors if `size == 0`
+ /// β a zero-row batch yields no batches at the parquet stream reader and
+ /// is almost certainly a caller mistake.
+ pub fn with_batch_size(mut self, size: usize) -> crate::Result<Self> {
+ if size == 0 {
+ let key = HudiReadConfig::StreamBatchSize.as_ref();
+ return Err(ConfigError::InvalidValue(format!("{key} must be > 0, got 0")).into());
+ }
+ self.hudi_options.insert(
+ HudiReadConfig::StreamBatchSize.as_ref().to_string(),
+ size.to_string(),
+ );
+ Ok(self)
+ }
+
+ /// Sets column filters from `(field, op, value)` tuples. Parses and
+ /// validates operator + cardinality at build time; an unrecognized
+ /// operator or empty `IN`/`NOT IN` value list errors here rather than
+ /// at read time. Schema-level validation (column existence, value
+ /// castability) still happens at read time when the schema is known.
+ pub fn with_filters<I, S1, S2, S3>(mut self, filters: I) -> crate::Result<Self>
where
I: IntoIterator<Item = (S1, S2, S3)>,
- S1: Into<String>,
- S2: Into<String>,
- S3: Into<String>,
+ S1: AsRef<str>,
+ S2: AsRef<str>,
+ S3: AsRef<str>,
{
- self.partition_filters = filters
- .into_iter()
- .map(|(f, o, v)| (f.into(), o.into(), v.into()))
- .collect();
- self
+ self.filters = from_str_tuples(filters.into_iter().map(|(f, o, v)| {
+ (
+ f.as_ref().to_string(),
+ o.as_ref().to_string(),
+ v.as_ref().to_string(),
+ )
+ }))?;
+ Ok(self)
}
/// Sets the column projection (which columns to read).
- ///
- /// # Arguments
- /// * `columns` - Column names to project
pub fn with_projection<I, S>(mut self, columns: I) -> Self
where
I: IntoIterator<Item = S>,
@@ -114,52 +182,81 @@ impl ReadOptions {
self
}
- /// Sets the row-level predicate for filtering records.
- ///
- /// The predicate function receives each `RecordBatch` and returns a `BooleanArray`
- /// mask indicating which rows to keep. Rows where the mask is `true` are retained.
- ///
- /// # Arguments
- /// * `predicate` - A function that takes a RecordBatch and returns a BooleanArray mask
- pub fn with_row_predicate<F>(mut self, predicate: F) -> Self
+ /// Sets a single Hudi config that applies to this read only.
+ pub fn with_hudi_option<K, V>(mut self, key: K, value: V) -> Self
where
- F: Fn(&RecordBatch) -> crate::Result<BooleanArray> + Send + Sync + 'static,
+ K: Into<String>,
+ V: Into<String>,
{
- self.row_predicate = Some(Arc::new(predicate));
+ self.hudi_options.insert(key.into(), value.into());
self
}
- /// Sets the target batch size (rows per batch).
- ///
- /// # Arguments
- /// * `size` - Target number of rows per batch
- pub fn with_batch_size(mut self, size: usize) -> Self {
- self.batch_size = Some(size);
+ /// Sets a batch of Hudi configs that apply to this read only.
+ pub fn with_hudi_options<I, K, V>(mut self, opts: I) -> Self
+ where
+ I: IntoIterator<Item = (K, V)>,
+ K: Into<String>,
+ V: Into<String>,
+ {
+ for (k, v) in opts {
+ self.hudi_options.insert(k.into(), v.into());
+ }
self
}
- /// Sets the as-of timestamp for time travel queries.
- ///
- /// # Arguments
- /// * `timestamp` - The timestamp to query as of
- pub fn with_as_of_timestamp<S: Into<String>>(mut self, timestamp: S) -> Self {
- self.as_of_timestamp = Some(timestamp.into());
- self
+ // ---- typed accessors (read from hudi_options) ----
+
+ /// The query type (defaults to [`QueryType::Snapshot`] when unset). Errors if
+ /// the stored string is not a recognized variant.
+ pub fn query_type(&self) -> crate::Result<QueryType> {
+ match self.hudi_options.get(HudiReadConfig::QueryType.as_ref()) {
+ Some(s) => Ok(QueryType::from_str(s)?),
+ None => Ok(QueryType::default()),
+ }
}
-}
-impl std::fmt::Debug for ReadOptions {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- f.debug_struct("ReadOptions")
- .field("filters", &self.partition_filters)
- .field("projection", &self.projection)
- .field(
- "row_predicate",
- &self.row_predicate.as_ref().map(|_| "<predicate>"),
- )
- .field("batch_size", &self.batch_size)
- .field("as_of_timestamp", &self.as_of_timestamp)
- .finish()
+ /// The as-of timestamp for snapshot/time-travel queries, if set.
+ pub fn as_of_timestamp(&self) -> Option<&str> {
+ self.hudi_options
+ .get(HudiReadConfig::AsOfTimestamp.as_ref())
+ .map(|s| s.as_str())
+ }
+
+ /// The start timestamp (exclusive) for incremental queries, if set.
+ pub fn start_timestamp(&self) -> Option<&str> {
+ self.hudi_options
+ .get(HudiReadConfig::StartTimestamp.as_ref())
+ .map(|s| s.as_str())
+ }
+
+ /// The end timestamp (inclusive) for incremental queries, if set.
+ pub fn end_timestamp(&self) -> Option<&str> {
+ self.hudi_options
+ .get(HudiReadConfig::EndTimestamp.as_ref())
+ .map(|s| s.as_str())
+ }
+
+ /// 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) {
+ Some(s) => {
+ let parsed = s
+ .parse::<usize>()
+ .map_err(|e| ConfigError::ParseInt(key.to_string(), s.clone(), e))?;
+ if parsed == 0 {
+ return Err(
+ ConfigError::InvalidValue(format!("{key} must be > 0, got 0")).into(),
+ );
+ }
+ Ok(Some(parsed))
+ }
+ None => Ok(None),
+ }
}
}
@@ -182,34 +279,109 @@ mod tests {
}
#[test]
- fn test_with_row_predicate() {
- let options = ReadOptions::new()
- .with_row_predicate(|batch| Ok(BooleanArray::from(vec![true; batch.num_rows()])));
+ fn test_with_query_type_round_trip() -> crate::Result<()> {
+ let snapshot = ReadOptions::new();
+ assert_eq!(snapshot.query_type()?, QueryType::Snapshot);
- assert!(options.row_predicate.is_some());
+ let incr = ReadOptions::new().with_query_type(QueryType::Incremental);
+ assert_eq!(incr.query_type()?, QueryType::Incremental);
+ assert_eq!(
+ incr.hudi_options
+ .get(HudiReadConfig::QueryType.as_ref())
+ .map(String::as_str),
+ Some("incremental")
+ );
+ Ok(())
}
#[test]
- fn test_with_as_of_timestamp() {
- let options = ReadOptions::new().with_as_of_timestamp("20240101120000000");
+ fn test_with_timestamps_round_trip() -> crate::Result<()> {
+ let opts = ReadOptions::new()
+ .with_as_of_timestamp("20240101120000000")
+ .with_start_timestamp("20240101000000000")
+ .with_end_timestamp("20240201000000000");
+ assert_eq!(opts.as_of_timestamp(), Some("20240101120000000"));
+ assert_eq!(opts.start_timestamp(), Some("20240101000000000"));
+ assert_eq!(opts.end_timestamp(), Some("20240201000000000"));
+ Ok(())
+ }
+
+ #[test]
+ fn test_with_batch_size_round_trip() -> crate::Result<()> {
+ let opts = ReadOptions::new();
+ assert_eq!(opts.batch_size()?, None);
+
+ let opts = ReadOptions::new().with_batch_size(2048)?;
+ assert_eq!(opts.batch_size()?, Some(2048));
+ assert_eq!(
+ opts.hudi_options
+ .get(HudiReadConfig::StreamBatchSize.as_ref())
+ .map(String::as_str),
+ Some("2048")
+ );
+
+ // Builder rejects 0 immediately rather than deferring to read time.
+ let err = ReadOptions::new().with_batch_size(0).unwrap_err();
+ assert!(err.to_string().contains("must be > 0"));
+
+ // Out-of-band insertion via the bag still surfaces at the accessor.
+ let opts = ReadOptions::new()
+ .with_hudi_option(HudiReadConfig::StreamBatchSize.as_ref(), "not_a_number");
+ let err = opts.batch_size().unwrap_err();
+ assert!(err.to_string().contains("not_a_number"));
+ Ok(())
+ }
+ #[test]
+ fn test_with_filters_validates_at_build_time() {
+ // Bad operator surfaces at build, not at read.
+ let err = ReadOptions::new()
+ .with_filters([("col", "BAD_OP", "x")])
+ .unwrap_err();
+ assert!(err.to_string().contains("BAD_OP"));
+
+ // Empty IN value list (after split + trim) errors via cardinality check.
+ let err = ReadOptions::new()
+ .with_filters([("col", "IN", "")])
+ .unwrap_err();
+ assert!(err.to_string().contains("at least one value"));
+ }
+
+ #[test]
+ fn test_with_hudi_options() {
+ let options = ReadOptions::new()
+ .with_hudi_option("hoodie.read.use.read_optimized.mode", "true")
+ .with_hudi_options([("a", "1"), ("b", "2")]);
assert_eq!(
- options.as_of_timestamp,
- Some("20240101120000000".to_string())
+ options
+ .hudi_options
+ .get("hoodie.read.use.read_optimized.mode"),
+ Some(&"true".to_string())
);
+ assert_eq!(options.hudi_options.get("a"), Some(&"1".to_string()));
+ assert_eq!(options.hudi_options.get("b"), Some(&"2".to_string()));
+ }
+
+ #[test]
+ fn test_query_type_from_str_invalid_errors() {
+ let opts =
+ ReadOptions::new().with_hudi_option(HudiReadConfig::QueryType.as_ref(), "garbage");
+ let err = opts.query_type().unwrap_err();
+ assert!(err.to_string().contains("garbage"));
}
#[test]
- fn test_debug_format() {
+ fn test_debug_format() -> crate::Result<()> {
let options = ReadOptions::new()
- .with_filters([("city", "=", "sf")])
+ .with_filters([("city", "=", "sf")])?
.with_projection(["id"])
- .with_batch_size(1000);
+ .with_batch_size(1000)?;
let debug_str = format!("{options:?}");
assert!(debug_str.contains("ReadOptions"));
assert!(debug_str.contains("filters"));
assert!(debug_str.contains("projection"));
- assert!(debug_str.contains("batch_size"));
+ assert!(debug_str.contains("hudi_options"));
+ Ok(())
}
}
diff --git a/crates/core/src/timeline/mod.rs b/crates/core/src/timeline/mod.rs
index 5fc498f0..d4e7e3b4 100644
--- a/crates/core/src/timeline/mod.rs
+++ b/crates/core/src/timeline/mod.rs
@@ -134,7 +134,7 @@ impl Timeline {
}
}
- async fn load_instants_internal(
+ async fn load_instants_inner(
&self,
selector: &TimelineSelector,
desc: bool,
@@ -155,7 +155,7 @@ impl Timeline {
pub async fn get_completed_commits(&self, desc: bool) -> Result<Vec<Instant>> {
let selector =
TimelineSelector::completed_commits_in_range(self.hudi_configs.clone(), None, None)?;
- self.load_instants_internal(&selector, desc).await
+ self.load_instants_inner(&selector, desc).await
}
/// Get the completed deltacommit [Instant]s in the timeline.
@@ -171,7 +171,7 @@ impl Timeline {
None,
None,
)?;
- self.load_instants_internal(&selector, desc).await
+ self.load_instants_inner(&selector, desc).await
}
/// Get the completed replacecommit [Instant]s in the timeline.
@@ -185,7 +185,7 @@ impl Timeline {
None,
None,
)?;
- self.load_instants_internal(&selector, desc).await
+ self.load_instants_inner(&selector, desc).await
}
/// Get the completed clustering commit [Instant]s in the timeline.
@@ -199,7 +199,7 @@ impl Timeline {
None,
None,
)?;
- let instants = self.load_instants_internal(&selector, desc).await?;
+ let instants = self.load_instants_inner(&selector, desc).await?;
let mut clustering_instants = Vec::new();
for instant in instants {
let metadata = self.get_instant_metadata(&instant).await?;
diff --git a/crates/core/src/timeline/util.rs b/crates/core/src/timeline/util.rs
index f5bd238f..31bb5f0e 100644
--- a/crates/core/src/timeline/util.rs
+++ b/crates/core/src/timeline/util.rs
@@ -23,12 +23,12 @@ use CoreError::TimestampParsingError;
use chrono::{DateTime, Datelike, Local, TimeZone, Timelike, Utc};
use std::str::FromStr;
-/// Parse various timestamp formats and convert to Hudi timeline format
+/// Parse various timestamp formats and convert to Hudi timeline format.
///
/// Supported formats:
-/// - Hudi timeline format: `yyyyMMddHHmmSSSSS` or `yyyyMMddHHmmSS`
+/// - Hudi timeline format: `yyyyMMddHHmmssSSS` or `yyyyMMddHHmmss`
/// - Epoch time (seconds, milliseconds, microseconds, nanoseconds)
-/// - RFC3339 format like `2024-03-15T14:25:30Z` or `2024-03-15T14:25:30+00:00` or `2024-03-15`
+/// - RFC3339 format like `2024-03-15T14:25:30Z` or `2024-03-15T14:25:30+00:00`
///
/// # Arguments
/// * `ts_str` - The timestamp str to parse
diff --git a/crates/core/src/util/arrow.rs b/crates/core/src/util/arrow.rs
index 30cd9988..1f8bb9ec 100644
--- a/crates/core/src/util/arrow.rs
+++ b/crates/core/src/util/arrow.rs
@@ -102,6 +102,28 @@ where
.collect()
}
+/// Project a [`RecordBatch`] to a subset of columns by name. Returns the batch
+/// unchanged when `projection` is `None`. Errors when any name is not present in
+/// the batch's schema.
+pub fn project_batch_by_names(
+ batch: RecordBatch,
+ projection: Option<&[String]>,
+) -> Result<RecordBatch> {
+ let Some(cols) = projection else {
+ return Ok(batch);
+ };
+ let indices: Vec<usize> = cols
+ .iter()
+ .map(|name| {
+ batch
+ .schema()
+ .index_of(name)
+ .map_err(|e| CoreError::Schema(format!("Projection column not found: {e:?}")))
+ })
+ .collect::<Result<_>>()?;
+ batch.project(&indices).map_err(CoreError::ArrowError)
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -181,4 +203,42 @@ mod tests {
let result = lexsort_to_indices(&arrays, false);
assert_eq!(result.values(), &[0, 2, 1]);
}
+
+ #[test]
+ fn test_project_batch_by_names() {
+ use arrow_schema::{DataType, Field, Schema};
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, false),
+ Field::new("b", DataType::Utf8, false),
+ Field::new("c", DataType::Float64, false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef,
+ Arc::new(StringArray::from(vec!["x", "y"])) as ArrayRef,
+ Arc::new(Float64Array::from(vec![1.5, 2.5])) as ArrayRef,
+ ],
+ )
+ .unwrap();
+
+ // None returns input unchanged.
+ let same = project_batch_by_names(batch.clone(), None).unwrap();
+ assert_eq!(same.num_columns(), 3);
+ assert_eq!(same.schema().field(0).name(), "a");
+
+ // Subset projects in the requested order.
+ let cols = vec!["c".to_string(), "a".to_string()];
+ let projected = project_batch_by_names(batch.clone(), Some(&cols)).unwrap();
+ assert_eq!(projected.num_columns(), 2);
+ assert_eq!(projected.schema().field(0).name(), "c");
+ assert_eq!(projected.schema().field(1).name(), "a");
+
+ // Unknown column errors with a Schema error.
+ let bad = vec!["a".to_string(), "missing".to_string()];
+ let err = project_batch_by_names(batch, Some(&bad)).unwrap_err();
+ assert!(matches!(err, CoreError::Schema(_)));
+ assert!(err.to_string().contains("missing"));
+ }
}
diff --git a/crates/core/tests/table_read_tests.rs b/crates/core/tests/table_read_tests.rs
index 44efa53c..faaffd42 100644
--- a/crates/core/tests/table_read_tests.rs
+++ b/crates/core/tests/table_read_tests.rs
@@ -26,9 +26,8 @@ use arrow_array::RecordBatch;
use futures::StreamExt;
use futures::stream::BoxStream;
use hudi_core::config::read::HudiReadConfig;
-use hudi_core::config::util::empty_filters;
use hudi_core::error::Result;
-use hudi_core::table::Table;
+use hudi_core::table::{QueryType, ReadOptions, Table};
use hudi_test::{QuickstartTripsTable, SampleTable};
async fn collect_stream_batches(
@@ -52,7 +51,7 @@ mod v6_tables {
async fn test_empty_table() -> Result<()> {
for base_url in SampleTable::V6Empty.urls() {
let hudi_table = Table::new(base_url.path()).await?;
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
assert!(records.is_empty());
}
Ok(())
@@ -62,7 +61,7 @@ mod v6_tables {
async fn test_non_partitioned() -> Result<()> {
for base_url in SampleTable::V6Nonpartitioned.urls() {
let hudi_table = Table::new(base_url.path()).await?;
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -96,7 +95,7 @@ mod v6_tables {
.collect::<Vec<_>>();
let latest_commit = commit_timestamps.last().unwrap();
let records = hudi_table
- .read_snapshot_as_of(latest_commit, empty_filters())
+ .read(&ReadOptions::new().with_as_of_timestamp(latest_commit))
.await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -118,7 +117,7 @@ mod v6_tables {
async fn test_non_partitioned_rollback() -> Result<()> {
let base_url = SampleTable::V6NonpartitionedRollback.url_to_mor_parquet();
let hudi_table = Table::new(base_url.path()).await?;
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -144,7 +143,9 @@ mod v6_tables {
("byteField", "<", "20"),
("shortField", "!=", "100"),
];
- let records = hudi_table.read_snapshot(filters).await?;
+ let records = hudi_table
+ .read(&ReadOptions::new().with_filters(filters)?)
+ .await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -158,7 +159,7 @@ mod v6_tables {
async fn test_simple_keygen_hivestyle_no_metafields() -> Result<()> {
for base_url in SampleTable::V6SimplekeygenHivestyleNoMetafields.urls() {
let hudi_table = Table::new(base_url.path()).await?;
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -192,7 +193,7 @@ mod v6_tables {
.collect::<Vec<_>>();
let first_commit = commit_timestamps[0];
let records = hudi_table
- .read_snapshot_as_of(first_commit, empty_filters())
+ .read(&ReadOptions::new().with_as_of_timestamp(first_commit))
.await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -218,7 +219,7 @@ mod v6_tables {
let updated_rider = "rider-D";
// verify updated record as of the latest commit
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
let uuid_rider_and_fare = QuickstartTripsTable::uuid_rider_and_fare(&records)
@@ -241,7 +242,7 @@ mod v6_tables {
.collect::<Vec<_>>();
let first_commit = commit_timestamps[0];
let records = hudi_table
- .read_snapshot_as_of(first_commit, empty_filters())
+ .read(&ReadOptions::new().with_as_of_timestamp(first_commit))
.await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -267,7 +268,7 @@ mod v6_tables {
let deleted_riders = ["rider-A", "rider-C", "rider-D"];
// verify deleted record as of the latest commit
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
let riders = QuickstartTripsTable::uuid_rider_and_fare(&records)
@@ -289,7 +290,7 @@ mod v6_tables {
.collect::<Vec<_>>();
let first_commit = commit_timestamps[0];
let records = hudi_table
- .read_snapshot_as_of(first_commit, empty_filters())
+ .read(&ReadOptions::new().with_as_of_timestamp(first_commit))
.await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -317,7 +318,13 @@ mod v6_tables {
async fn test_empty_table() -> Result<()> {
for base_url in SampleTable::V6Empty.urls() {
let hudi_table = Table::new(base_url.path()).await?;
- let records = hudi_table.read_incremental_records("0", None).await?;
+ let records = hudi_table
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp("0"),
+ )
+ .await?;
assert!(records.is_empty())
}
Ok(())
@@ -340,7 +347,12 @@ mod v6_tables {
// read records changed from the beginning to the 1st commit
let records = hudi_table
- .read_incremental_records("19700101000000", Some(first_commit))
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp("19700101000000")
+ .with_end_timestamp(first_commit),
+ )
.await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -353,7 +365,12 @@ mod v6_tables {
// read records changed from the 1st to the 2nd commit
let records = hudi_table
- .read_incremental_records(first_commit, Some(second_commit))
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(first_commit)
+ .with_end_timestamp(second_commit),
+ )
.await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -366,7 +383,12 @@ mod v6_tables {
// read records changed from the 2nd to the 3rd commit
let records = hudi_table
- .read_incremental_records(second_commit, Some(third_commit))
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(second_commit)
+ .with_end_timestamp(third_commit),
+ )
.await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -379,7 +401,11 @@ mod v6_tables {
// read records changed from the 1st commit
let records = hudi_table
- .read_incremental_records(first_commit, None)
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(first_commit),
+ )
.await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -392,7 +418,11 @@ mod v6_tables {
// read records changed from the 3rd commit
let records = hudi_table
- .read_incremental_records(third_commit, None)
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(third_commit),
+ )
.await?;
assert!(
records.is_empty(),
@@ -415,7 +445,7 @@ mod v8_tables {
async fn test_empty_table() -> Result<()> {
let base_url = SampleTable::V8Empty.url_to_cow();
let hudi_table = Table::new(base_url.path()).await?;
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
assert!(records.is_empty());
Ok(())
}
@@ -424,7 +454,7 @@ mod v8_tables {
async fn test_non_partitioned() -> Result<()> {
let base_url = SampleTable::V8Nonpartitioned.url_to_cow();
let hudi_table = Table::new(base_url.path()).await?;
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -446,7 +476,7 @@ mod v8_tables {
let base_url = SampleTable::V8ComplexkeygenHivestyle.url_to_cow();
let hudi_table = Table::new(base_url.path()).await?;
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -468,7 +498,7 @@ mod v8_tables {
let base_url = SampleTable::V8SimplekeygenNonhivestyle.url_to_cow();
let hudi_table = Table::new(base_url.path()).await?;
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -490,7 +520,7 @@ mod v8_tables {
let base_url = SampleTable::V8SimplekeygenHivestyleNoMetafields.url_to_cow();
let hudi_table = Table::new(base_url.path()).await?;
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -521,7 +551,7 @@ mod v8_tables {
let deleted_riders = ["rider-F", "rider-J"];
// verify deleted records are not present in latest snapshot
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
let uuid_rider_and_fare = QuickstartTripsTable::uuid_rider_and_fare(&records);
@@ -562,7 +592,7 @@ mod v8_tables {
.collect::<Vec<_>>();
let first_commit = commit_timestamps[0];
let records = hudi_table
- .read_snapshot_as_of(first_commit, empty_filters())
+ .read(&ReadOptions::new().with_as_of_timestamp(first_commit))
.await?;
let schema = &records[0].schema();
let records = concat_batches(schema, &records)?;
@@ -581,6 +611,68 @@ mod v8_tables {
Ok(())
}
+
+ #[tokio::test]
+ async fn test_file_group_reader_read_file_slice_mor_applies_filters_and_projection_after_merge()
+ -> Result<()> {
+ // Direct FileGroupReader path on a MOR slice with log files. The
+ // reader merges base + log records first; only afterwards does
+ // `apply_eager_options` apply `filters` (as a row mask) and
+ // `projection` (column narrowing). Verifies the post-merge
+ // hand-off is wired correctly: schema narrows, rows are filtered,
+ // and the count matches the table-level read with the same options.
+ let base_url = QuickstartTripsTable::V8Trips8I3U1D.url_to_mor_avro();
+ let hudi_table = Table::new(base_url.path()).await?;
+ let all_slices = hudi_table.get_file_slices(&ReadOptions::new()).await?;
+ let mor_slice = all_slices
+ .iter()
+ .find(|fs| fs.has_log_file())
+ .expect("V8Trips8I3U1D MOR fixture should have at least one slice with log files");
+
+ let fg_reader = hudi_table.create_file_group_reader_with_options(
+ None,
+ std::iter::empty::<(&str, &str)>(),
+ std::iter::empty::<(&str, &str)>(),
+ )?;
+ // Sanity: read the merged slice unfiltered so we can pick a rider
+ // present in this slice and assert the filter actually narrows it.
+ let unfiltered = fg_reader
+ .read_file_slice(mor_slice, &ReadOptions::new())
+ .await?;
+ assert!(unfiltered.num_rows() > 0);
+ let unfiltered_riders = unfiltered
+ .column_by_name("rider")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<arrow::array::StringArray>()
+ .unwrap();
+ let target_rider = unfiltered_riders.value(0).to_string();
+
+ let options = ReadOptions::new()
+ .with_filters([("rider", "=", target_rider.as_str())])?
+ .with_projection(["rider", "fare"]);
+ let merged = fg_reader.read_file_slice(mor_slice, &options).await?;
+
+ // Schema narrowed to projection.
+ let field_names: Vec<_> = merged
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| f.name().clone())
+ .collect();
+ assert_eq!(field_names, vec!["rider", "fare"]);
+ // Filter applied post-merge: only the target rider's row survives,
+ // and there's exactly one row per rider in the merged result.
+ assert_eq!(merged.num_rows(), 1);
+ let riders = merged
+ .column_by_name("rider")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<arrow::array::StringArray>()
+ .unwrap();
+ assert_eq!(riders.value(0), target_rider);
+ Ok(())
+ }
}
/// Streaming query tests for v8 tables
@@ -593,7 +685,7 @@ mod v8_tables {
let base_url = SampleTable::V8Empty.url_to_cow();
let hudi_table = Table::new(base_url.path()).await?;
let options = ReadOptions::new().with_projection(["txn_id", "txn_type", "txn_ts"]);
- let mut stream = hudi_table.read_snapshot_stream(&options).await?;
+ let mut stream = hudi_table.read_stream(&options).await?;
// Collect all batches from stream
let mut batches = Vec::new();
@@ -609,7 +701,7 @@ mod v8_tables {
let base_url = SampleTable::V8Nonpartitioned.url_to_cow();
let hudi_table = Table::new(base_url.path()).await?;
let options = ReadOptions::new();
- let mut stream = hudi_table.read_snapshot_stream(&options).await?;
+ let mut stream = hudi_table.read_stream(&options).await?;
// Collect all batches from stream
let mut batches = Vec::new();
@@ -642,8 +734,8 @@ mod v8_tables {
let hudi_table = Table::new(base_url.path()).await?;
// Request small batch size
- let options = ReadOptions::new().with_batch_size(1);
- let mut stream = hudi_table.read_snapshot_stream(&options).await?;
+ let options = ReadOptions::new().with_batch_size(1)?;
+ let mut stream = hudi_table.read_stream(&options).await?;
// Collect all batches from stream
let mut batches = Vec::new();
@@ -668,8 +760,8 @@ mod v8_tables {
("byteField", ">=", "10"),
("byteField", "<", "20"),
("shortField", "!=", "100"),
- ]);
- let mut stream = hudi_table.read_snapshot_stream(&options).await?;
+ ])?;
+ let mut stream = hudi_table.read_stream(&options).await?;
// Collect all batches from stream
let mut batches = Vec::new();
@@ -695,15 +787,20 @@ mod v8_tables {
let hudi_table = Table::new(base_url.path()).await?;
// Get file slices first
- let file_slices = hudi_table.get_file_slices(empty_filters()).await?;
+ let file_slices = hudi_table.get_file_slices(&ReadOptions::new()).await?;
assert!(
!file_slices.is_empty(),
"Should have at least one file slice"
);
+ let fg_reader = hudi_table.create_file_group_reader_with_options(
+ None,
+ std::iter::empty::<(&str, &str)>(),
+ std::iter::empty::<(&str, &str)>(),
+ )?;
let options = ReadOptions::new();
let file_slice = &file_slices[0];
- let mut stream = hudi_table
+ let mut stream = fg_reader
.read_file_slice_stream(file_slice, &options)
.await?;
@@ -726,12 +823,17 @@ mod v8_tables {
let base_url = SampleTable::V8Nonpartitioned.url_to_cow();
let hudi_table = Table::new(base_url.path()).await?;
- let file_slices = hudi_table.get_file_slices(empty_filters()).await?;
+ let file_slices = hudi_table.get_file_slices(&ReadOptions::new()).await?;
let file_slice = &file_slices[0];
+ let fg_reader = hudi_table.create_file_group_reader_with_options(
+ None,
+ std::iter::empty::<(&str, &str)>(),
+ std::iter::empty::<(&str, &str)>(),
+ )?;
// Test with small batch size
- let options = ReadOptions::new().with_batch_size(1);
- let mut stream = hudi_table
+ let options = ReadOptions::new().with_batch_size(1)?;
+ let mut stream = fg_reader
.read_file_slice_stream(file_slice, &options)
.await?;
@@ -753,7 +855,7 @@ mod v8_tables {
let hudi_table = Table::new(base_url.path()).await?;
let options = ReadOptions::new();
- let mut stream = hudi_table.read_snapshot_stream(&options).await?;
+ let mut stream = hudi_table.read_stream(&options).await?;
let mut batches = Vec::new();
while let Some(result) = stream.next().await {
@@ -826,7 +928,7 @@ mod v9_tables {
}
async fn read_txn_rows_from_snapshot(hudi_table: &Table) -> Result<Vec<(String, String, i64)>> {
- let records = hudi_table.read_snapshot(empty_filters()).await?;
+ let records = hudi_table.read(&ReadOptions::new()).await?;
Ok(txn_rows(&records))
}
@@ -835,7 +937,7 @@ mod v9_tables {
timestamp: &str,
) -> Result<Vec<(String, String, i64)>> {
let records = hudi_table
- .read_snapshot_as_of(timestamp, empty_filters())
+ .read(&ReadOptions::new().with_as_of_timestamp(timestamp))
.await?;
Ok(txn_rows(&records))
}
@@ -1101,7 +1203,12 @@ mod v9_tables {
let third_commit = commit_timestamps[2];
let records = hudi_table
- .read_incremental_records("19700101000000000", Some(first_commit))
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp("19700101000000000")
+ .with_end_timestamp(first_commit),
+ )
.await?;
let rows = txn_rows(&records);
assert_eq!(
@@ -1118,7 +1225,12 @@ mod v9_tables {
);
let records = hudi_table
- .read_incremental_records(first_commit, Some(second_commit))
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(first_commit)
+ .with_end_timestamp(second_commit),
+ )
.await?;
let rows = txn_rows(&records);
assert_eq!(
@@ -1131,7 +1243,12 @@ mod v9_tables {
);
let records = hudi_table
- .read_incremental_records(second_commit, Some(third_commit))
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(second_commit)
+ .with_end_timestamp(third_commit),
+ )
.await?;
let rows = txn_rows(&records);
assert_eq!(
@@ -1145,7 +1262,11 @@ mod v9_tables {
);
let records = hudi_table
- .read_incremental_records(first_commit, None)
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(first_commit),
+ )
.await?;
let rows = txn_rows(&records);
assert_eq!(
@@ -1159,7 +1280,11 @@ mod v9_tables {
);
let records = hudi_table
- .read_incremental_records(third_commit, None)
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(third_commit),
+ )
.await?;
assert!(
records.is_empty(),
@@ -1183,7 +1308,12 @@ mod v9_tables {
let second_commit = &deltacommits[1].timestamp;
let records = hudi_table
- .read_incremental_records("19700101000000000", Some(first_commit))
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp("19700101000000000")
+ .with_end_timestamp(first_commit),
+ )
.await?;
let rows = txn_rows(&records);
assert_eq!(
@@ -1197,7 +1327,12 @@ mod v9_tables {
);
let records = hudi_table
- .read_incremental_records(first_commit, Some(second_commit))
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(first_commit)
+ .with_end_timestamp(second_commit),
+ )
.await?;
let rows = txn_rows(&records);
assert_eq!(
@@ -1207,7 +1342,11 @@ mod v9_tables {
);
let records = hudi_table
- .read_incremental_records(first_commit, None)
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(first_commit),
+ )
.await?;
let rows = txn_rows(&records);
assert_eq!(
@@ -1217,7 +1356,11 @@ mod v9_tables {
);
let records = hudi_table
- .read_incremental_records(second_commit, None)
+ .read(
+ &ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(second_commit),
+ )
.await?;
assert!(
records.is_empty(),
@@ -1238,7 +1381,7 @@ mod v9_tables {
let hudi_table = open_table(base_url.path(), false).await?;
let options = ReadOptions::new();
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
assert!(!batches.is_empty(), "Should produce at least one batch");
@@ -1260,8 +1403,8 @@ mod v9_tables {
let base_url = SampleTable::V9TxnsSimpleOverwrite.url_to_cow();
let hudi_table = open_table(base_url.path(), false).await?;
- let options = ReadOptions::new().with_batch_size(1);
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let options = ReadOptions::new().with_batch_size(1)?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
@@ -1274,8 +1417,8 @@ mod v9_tables {
let base_url = SampleTable::V9TxnsSimpleOverwrite.url_to_cow();
let hudi_table = open_table(base_url.path(), false).await?;
- let options = ReadOptions::new().with_filters([("region", "=", "us")]);
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let options = ReadOptions::new().with_filters([("region", "=", "us")])?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
assert!(
@@ -1295,8 +1438,8 @@ mod v9_tables {
let base_url = SampleTable::V9TxnsSimpleOverwrite.url_to_cow();
let hudi_table = open_table(base_url.path(), false).await?;
- let options = ReadOptions::new().with_filters([("region", "=", "latam")]);
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let options = ReadOptions::new().with_filters([("region", "=", "latam")])?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
assert!(
@@ -1311,15 +1454,20 @@ mod v9_tables {
let base_url = SampleTable::V9TxnsSimpleOverwrite.url_to_cow();
let hudi_table = open_table(base_url.path(), false).await?;
- let file_slices = hudi_table.get_file_slices(empty_filters()).await?;
+ let file_slices = hudi_table.get_file_slices(&ReadOptions::new()).await?;
assert!(
!file_slices.is_empty(),
"Should have at least one file slice"
);
+ let fg_reader = hudi_table.create_file_group_reader_with_options(
+ None,
+ std::iter::empty::<(&str, &str)>(),
+ std::iter::empty::<(&str, &str)>(),
+ )?;
let options = ReadOptions::new();
let file_slice = &file_slices[0];
- let stream = hudi_table
+ let stream = fg_reader
.read_file_slice_stream(file_slice, &options)
.await?;
let batches = collect_stream_batches(stream).await?;
@@ -1334,11 +1482,16 @@ mod v9_tables {
let base_url = SampleTable::V9TxnsSimpleOverwrite.url_to_cow();
let hudi_table = open_table(base_url.path(), false).await?;
- let file_slices = hudi_table.get_file_slices(empty_filters()).await?;
+ let file_slices = hudi_table.get_file_slices(&ReadOptions::new()).await?;
let file_slice = &file_slices[0];
- let options = ReadOptions::new().with_batch_size(1);
- let stream = hudi_table
+ let fg_reader = hudi_table.create_file_group_reader_with_options(
+ None,
+ std::iter::empty::<(&str, &str)>(),
+ std::iter::empty::<(&str, &str)>(),
+ )?;
+ let options = ReadOptions::new().with_batch_size(1)?;
+ let stream = fg_reader
.read_file_slice_stream(file_slice, &options)
.await?;
let batches = collect_stream_batches(stream).await?;
@@ -1355,7 +1508,7 @@ mod v9_tables {
let base_url = SampleTable::V9TimebasedkeygenNonhivestyle.url_to_mor_avro();
let hudi_table = open_table(base_url.path(), false).await?;
- let file_slices = hudi_table.get_file_slices(empty_filters()).await?;
+ let file_slices = hudi_table.get_file_slices(&ReadOptions::new()).await?;
assert!(
file_slices
.iter()
@@ -1364,7 +1517,7 @@ mod v9_tables {
);
let options = ReadOptions::new();
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let stream = hudi_table.read_stream(&options).await?;
let err = match collect_stream_batches(stream).await {
Ok(_) => panic!("Expected MOR streaming read with decimal log records to fail"),
Err(err) => err,
@@ -1383,7 +1536,7 @@ mod v9_tables {
let hudi_table = open_table(base_url.path(), true).await?;
let options = ReadOptions::new();
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
assert!(!batches.is_empty(), "Should produce batches from MOR table");
@@ -1410,7 +1563,7 @@ mod streaming_queries {
for base_url in SampleTable::V6Empty.urls() {
let hudi_table = Table::new(base_url.path()).await?;
let options = ReadOptions::new();
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
assert!(batches.is_empty(), "Empty table should produce no batches");
}
@@ -1422,7 +1575,7 @@ mod streaming_queries {
for base_url in SampleTable::V6Nonpartitioned.urls() {
let hudi_table = Table::new(base_url.path()).await?;
let options = ReadOptions::new();
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
assert!(!batches.is_empty(), "Should produce at least one batch");
@@ -1451,8 +1604,8 @@ mod streaming_queries {
let hudi_table = Table::new(base_url.path()).await?;
// Request small batch size
- let options = ReadOptions::new().with_batch_size(1);
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let options = ReadOptions::new().with_batch_size(1)?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
// With batch_size=1 and 4 rows, we expect multiple batches, but the
@@ -1472,8 +1625,8 @@ mod streaming_queries {
("byteField", ">=", "10"),
("byteField", "<", "20"),
("shortField", "!=", "100"),
- ]);
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ ])?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
assert!(
@@ -1494,15 +1647,20 @@ mod streaming_queries {
let hudi_table = Table::new(base_url.path()).await?;
// Get file slices first
- let file_slices = hudi_table.get_file_slices(empty_filters()).await?;
+ let file_slices = hudi_table.get_file_slices(&ReadOptions::new()).await?;
assert!(
!file_slices.is_empty(),
"Should have at least one file slice"
);
+ let fg_reader = hudi_table.create_file_group_reader_with_options(
+ None,
+ std::iter::empty::<(&str, &str)>(),
+ std::iter::empty::<(&str, &str)>(),
+ )?;
let options = ReadOptions::new();
let file_slice = &file_slices[0];
- let stream = hudi_table
+ let stream = fg_reader
.read_file_slice_stream(file_slice, &options)
.await?;
let batches = collect_stream_batches(stream).await?;
@@ -1520,12 +1678,17 @@ mod streaming_queries {
let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
let hudi_table = Table::new(base_url.path()).await?;
- let file_slices = hudi_table.get_file_slices(empty_filters()).await?;
+ let file_slices = hudi_table.get_file_slices(&ReadOptions::new()).await?;
let file_slice = &file_slices[0];
+ let fg_reader = hudi_table.create_file_group_reader_with_options(
+ None,
+ std::iter::empty::<(&str, &str)>(),
+ std::iter::empty::<(&str, &str)>(),
+ )?;
// Test with small batch size
- let options = ReadOptions::new().with_batch_size(1);
- let stream = hudi_table
+ let options = ReadOptions::new().with_batch_size(1)?;
+ let stream = fg_reader
.read_file_slice_stream(file_slice, &options)
.await?;
let batches = collect_stream_batches(stream).await?;
@@ -1542,7 +1705,7 @@ mod streaming_queries {
let hudi_table = Table::new(base_url.path()).await?;
let options = ReadOptions::new();
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
assert!(!batches.is_empty(), "Should produce batches from MOR table");
@@ -1560,7 +1723,7 @@ mod streaming_queries {
// Only request id and name columns (not isActive)
let options = ReadOptions::new().with_projection(["id", "name"]);
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
assert!(!batches.is_empty(), "Should produce at least one batch");
@@ -1588,25 +1751,15 @@ mod streaming_queries {
}
#[tokio::test]
- async fn test_read_snapshot_stream_with_row_predicate() -> Result<()> {
- use arrow::array::BooleanArray;
-
+ async fn test_read_snapshot_stream_with_non_partition_filter() -> Result<()> {
let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
let hudi_table = Table::new(base_url.path()).await?;
- // Filter rows where isActive = true (Carol and Diana)
- let options = ReadOptions::new().with_row_predicate(|batch: &RecordBatch| {
- let col = batch
- .column_by_name("isActive")
- .ok_or_else(|| hudi_core::error::CoreError::Schema("isActive not found".into()))?;
- let arr = col
- .as_any()
- .downcast_ref::<BooleanArray>()
- .ok_or_else(|| hudi_core::error::CoreError::Schema("Not boolean".into()))?;
- Ok(arr.clone())
- });
+ // Filter on a non-partition column. The filter applies as a row mask
+ // during reading even though file pruning can't use it.
+ let options = ReadOptions::new().with_filters([("isActive", "=", "true")])?;
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
assert!(!batches.is_empty(), "Should produce at least one batch");
@@ -1621,8 +1774,8 @@ mod streaming_queries {
}
#[tokio::test]
- async fn test_read_snapshot_stream_with_projection_and_row_predicate() -> Result<()> {
- use arrow::array::{BooleanArray, Int32Array};
+ async fn test_read_snapshot_stream_with_projection_and_filter() -> Result<()> {
+ use arrow::array::Int32Array;
let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
let hudi_table = Table::new(base_url.path()).await?;
@@ -1630,18 +1783,9 @@ mod streaming_queries {
// Project only id and isActive, filter where isActive = true
let options = ReadOptions::new()
.with_projection(["id", "isActive"])
- .with_row_predicate(|batch: &RecordBatch| {
- let col = batch.column_by_name("isActive").ok_or_else(|| {
- hudi_core::error::CoreError::Schema("isActive not found".into())
- })?;
- let arr = col
- .as_any()
- .downcast_ref::<BooleanArray>()
- .ok_or_else(|| hudi_core::error::CoreError::Schema("Not boolean".into()))?;
- Ok(arr.clone())
- });
-
- let stream = hudi_table.read_snapshot_stream(&options).await?;
+ .with_filters([("isActive", "=", "true")])?;
+
+ let stream = hudi_table.read_stream(&options).await?;
let batches = collect_stream_batches(stream).await?;
assert!(!batches.is_empty(), "Should produce at least one batch");
@@ -1668,6 +1812,182 @@ mod streaming_queries {
Ok(())
}
+ #[tokio::test]
+ async fn test_read_snapshot_stream_filter_column_not_in_projection() -> Result<()> {
+ use arrow::array::Int32Array;
+
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+
+ // Filter on `isActive` but project only `id` (filter column NOT projected).
+ // The filter must still apply at row level β implementation augments the
+ // read projection internally with filter columns, then projects down.
+ let options = ReadOptions::new()
+ .with_projection(["id"])
+ .with_filters([("isActive", "=", "true")])?;
+
+ let stream = hudi_table.read_stream(&options).await?;
+ let batches = collect_stream_batches(stream).await?;
+
+ let schema = &batches[0].schema();
+ assert_eq!(schema.fields().len(), 1);
+ assert!(schema.field_with_name("id").is_ok());
+ assert!(schema.field_with_name("isActive").is_err());
+
+ let records = concat_batches(schema, &batches)?;
+ assert_eq!(
+ records.num_rows(),
+ 2,
+ "filter must drop rows even when filter column is not projected"
+ );
+
+ let ids = records
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let id_values: Vec<i32> = ids.iter().flatten().collect();
+ assert!(id_values.contains(&3) && id_values.contains(&4));
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_read_snapshot_stream_with_as_of_timestamp() -> Result<()> {
+ // Cross-validate streaming time-travel against the eager API: with the
+ // same `as_of_timestamp`, both must return the same row count. (Before
+ // the fix, streaming silently used the latest commit and could diverge.)
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+ let first_commit = hudi_table
+ .get_timeline()
+ .get_completed_commits(false)
+ .await?
+ .first()
+ .map(|i| i.timestamp.to_string())
+ .expect("table must have at least one commit");
+ let options = ReadOptions::new().with_as_of_timestamp(&first_commit);
+
+ let eager = hudi_table.read(&options).await?;
+ let eager_rows: usize = eager.iter().map(|b| b.num_rows()).sum();
+
+ let stream = hudi_table.read_stream(&options).await?;
+ let stream_batches = collect_stream_batches(stream).await?;
+ let stream_rows: usize = stream_batches.iter().map(|b| b.num_rows()).sum();
+
+ assert_eq!(eager_rows, stream_rows);
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_filter_on_unknown_column_errors() -> Result<()> {
+ // A typo on a filter column should error rather than silently no-op,
+ // across all three dispatch paths (eager snapshot, eager incremental,
+ // streaming snapshot). Streaming surfaces the error synchronously,
+ // before the stream is constructed, so callers don't have to poll.
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+ let options = ReadOptions::new().with_filters([("rider_idd", "=", "x")])?;
+
+ let snapshot_err = hudi_table.read(&options).await.unwrap_err();
+ assert!(
+ snapshot_err.to_string().contains("rider_idd"),
+ "snapshot error should mention the bad column, got: {snapshot_err}"
+ );
+
+ let incremental_err = hudi_table
+ .read(&options.clone().with_query_type(QueryType::Incremental))
+ .await
+ .unwrap_err();
+ assert!(
+ incremental_err.to_string().contains("rider_idd"),
+ "incremental error should mention the bad column, got: {incremental_err}"
+ );
+
+ match hudi_table.read_stream(&options).await {
+ Ok(_) => panic!("read_stream must surface the typo synchronously"),
+ Err(e) => assert!(
+ e.to_string().contains("rider_idd"),
+ "stream error should mention the bad column, got: {e}"
+ ),
+ }
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_file_group_reader_filter_on_unknown_column_errors() -> Result<()> {
+ // FileGroupReader-direct reads bypass table-level filter validation; the
+ // reader must reject typoed filter fields itself rather than silently
+ // skipping them in `filters_to_row_mask`.
+ use hudi_core::file_group::reader::FileGroupReader;
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+ let file_slice = hudi_table.get_file_slices(&ReadOptions::new()).await?[0].clone();
+ let reader =
+ FileGroupReader::new_with_options(base_url.path(), std::iter::empty::<(&str, &str)>())
+ .await?;
+ let options = ReadOptions::new().with_filters([("rider_idd", "=", "x")])?;
+
+ let eager_err = reader
+ .read_file_slice(&file_slice, &options)
+ .await
+ .unwrap_err();
+ assert!(
+ eager_err.to_string().contains("rider_idd"),
+ "eager error should mention the bad column, got: {eager_err}"
+ );
+
+ // Streaming validates lazily on the first batch.
+ let stream = reader.read_file_slice_stream(&file_slice, &options).await?;
+ let mut stream_err = None;
+ let mut s = stream;
+ while let Some(result) = s.next().await {
+ if let Err(e) = result {
+ stream_err = Some(e);
+ break;
+ }
+ }
+ let stream_err = stream_err.expect("streaming must surface the typo as an error");
+ assert!(
+ stream_err.to_string().contains("rider_idd"),
+ "stream error should mention the bad column, got: {stream_err}"
+ );
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_file_group_reader_stream_projection_widens_for_commit_time_filter() -> Result<()>
+ {
+ // Regression: when commit-time filtering is active (StartTimestamp set
+ // and PopulatesMetaFields true), the streaming path must widen the parquet read
+ // projection to include `_hoodie_commit_time` even if the user's projection
+ // omits it. The widened column is dropped by the existing final-projection step
+ // so the user-visible schema still matches `options.projection`.
+ use hudi_core::file_group::reader::FileGroupReader;
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+ let file_slice = hudi_table.get_file_slices(&ReadOptions::new()).await?[0].clone();
+ let reader = FileGroupReader::new_with_options(
+ base_url.path(),
+ [(HudiReadConfig::StartTimestamp.as_ref(), "0")],
+ )
+ .await?;
+ let options = ReadOptions::new().with_projection(["id"]);
+
+ let stream = reader.read_file_slice_stream(&file_slice, &options).await?;
+ let batches = collect_stream_batches(stream).await?;
+
+ assert!(!batches.is_empty(), "should produce at least one batch");
+ let schema = batches[0].schema();
+ assert_eq!(
+ schema.fields().len(),
+ 1,
+ "projection should narrow to one column"
+ );
+ assert_eq!(schema.field(0).name(), "id");
+ Ok(())
+ }
+
#[tokio::test]
async fn test_read_snapshot_stream_projection_invalid_column() -> Result<()> {
let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
@@ -1675,7 +1995,7 @@ mod streaming_queries {
// Request a non-existent column
let options = ReadOptions::new().with_projection(["id", "nonexistent_column"]);
- let mut stream = hudi_table.read_snapshot_stream(&options).await?;
+ let mut stream = hudi_table.read_stream(&options).await?;
// Error occurs when polling the stream (lazy evaluation)
let mut found_error = false;
@@ -1698,6 +2018,133 @@ mod streaming_queries {
);
Ok(())
}
+
+ /// Regression: a stray `hoodie.read.start.timestamp` left in
+ /// `options.hudi_options` must NOT silently activate commit-time filtering on
+ /// a snapshot read. Snapshot dispatch + a future-dated start timestamp would
+ /// otherwise filter out every row at the FG-reader layer.
+ #[tokio::test]
+ async fn test_snapshot_ignores_stale_start_timestamp_in_options() -> Result<()> {
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+
+ let baseline = hudi_table.read(&ReadOptions::new()).await?;
+ let baseline_rows: usize = baseline.iter().map(|b| b.num_rows()).sum();
+ assert!(baseline_rows > 0);
+
+ // Snapshot dispatch (default), but with a future-dated start_timestamp
+ // sitting in the bag β a stray incremental knob the user might have set
+ // before flipping to snapshot. Snapshot must ignore it.
+ let polluted = ReadOptions::new().with_start_timestamp("99999999999999999");
+ let result = hudi_table.read(&polluted).await?;
+ let result_rows: usize = result.iter().map(|b| b.num_rows()).sum();
+
+ assert_eq!(
+ baseline_rows, result_rows,
+ "snapshot read must not honor an incremental-only start_timestamp"
+ );
+ Ok(())
+ }
+
+ /// Same regression for the streaming snapshot path.
+ #[tokio::test]
+ async fn test_snapshot_stream_ignores_stale_start_timestamp_in_options() -> Result<()> {
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+
+ let baseline_stream = hudi_table.read_stream(&ReadOptions::new()).await?;
+ let baseline = collect_stream_batches(baseline_stream).await?;
+ let baseline_rows: usize = baseline.iter().map(|b| b.num_rows()).sum();
+ assert!(baseline_rows > 0);
+
+ let polluted = ReadOptions::new().with_start_timestamp("99999999999999999");
+ let polluted_stream = hudi_table.read_stream(&polluted).await?;
+ let result = collect_stream_batches(polluted_stream).await?;
+ let result_rows: usize = result.iter().map(|b| b.num_rows()).sum();
+
+ assert_eq!(
+ baseline_rows, result_rows,
+ "snapshot stream must not honor an incremental-only start_timestamp"
+ );
+ Ok(())
+ }
+
+ /// Regression: the public `create_file_group_reader_with_options` API must
+ /// strip the four `Table`-owned read keys from `read_options.hudi_options`
+ /// before forwarding them to the FG reader. Otherwise a stray
+ /// `StartTimestamp` activates commit-time filtering at the physical layer
+ /// and silently drops every row.
+ #[tokio::test]
+ async fn test_create_fg_reader_strips_table_owned_keys() -> Result<()> {
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+ let file_slices = hudi_table.get_file_slices(&ReadOptions::new()).await?;
+ assert!(!file_slices.is_empty());
+ let file_slice = &file_slices[0];
+
+ // Bag pre-populated with all four Table-owned keys, including a
+ // future-dated StartTimestamp that would zero out commit-time filtering.
+ let polluted = ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_as_of_timestamp("19700101000000000")
+ .with_start_timestamp("99999999999999999")
+ .with_end_timestamp("19700101000000000");
+
+ let fg_reader = hudi_table.create_file_group_reader_with_options(
+ Some(&polluted),
+ std::iter::empty::<(&str, &str)>(),
+ std::iter::empty::<(&str, &str)>(),
+ )?;
+ let batch = fg_reader
+ .read_file_slice(file_slice, &ReadOptions::new())
+ .await?;
+ assert!(
+ batch.num_rows() > 0,
+ "Table-owned keys must not leak into the FG reader; got 0 rows"
+ );
+ Ok(())
+ }
+
+ /// Regression: filters on Hudi meta fields (e.g. `_hoodie_record_key`) must
+ /// be accepted by the table-level validation. Returned batches include meta
+ /// fields, and the row-level mask applies the filter against them.
+ #[tokio::test]
+ async fn test_table_read_accepts_meta_field_filter() -> Result<()> {
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+
+ // _hoodie_record_key is in the returned batch but not in the data
+ // schema. A typo'd column name should still error.
+ let options = ReadOptions::new().with_filters([(
+ hudi_core::metadata::meta_field::MetaField::RecordKey.as_ref(),
+ "!=",
+ "",
+ )])?;
+ let batches = hudi_table.read(&options).await?;
+ let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+ assert!(total_rows > 0, "meta-field filter should match real keys");
+
+ let typo = ReadOptions::new().with_filters([("_hoodie_record_keyy", "!=", "")])?;
+ let err = hudi_table.read(&typo).await.unwrap_err();
+ assert!(
+ err.to_string().contains("_hoodie_record_keyy"),
+ "typo'd filter should still fail validation; got: {err}"
+ );
+ Ok(())
+ }
+
+ /// Regression: `with_batch_size(0)` must error at the builder so callers
+ /// catch the misuse synchronously, not when the parquet stream reader
+ /// silently yields zero rows.
+ #[tokio::test]
+ async fn test_batch_size_zero_errors() -> Result<()> {
+ let err = ReadOptions::new().with_batch_size(0).unwrap_err();
+ assert!(
+ err.to_string().contains("must be > 0"),
+ "expected validation error for batch_size=0, got: {err}"
+ );
+ Ok(())
+ }
}
/// Test module for tables with metadata table (MDT) enabled.
@@ -1726,7 +2173,7 @@ mod mdt_enabled_tables {
);
// Get file slices - this uses MDT file listing
- let file_slices = hudi_table.get_file_slices(empty_filters()).await?;
+ let file_slices = hudi_table.get_file_slices(&ReadOptions::new()).await?;
// Should have file slices for the non-partitioned table
assert!(
diff --git a/crates/datafusion/src/lib.rs b/crates/datafusion/src/lib.rs
index 1785c639..53a5d885 100644
--- a/crates/datafusion/src/lib.rs
+++ b/crates/datafusion/src/lib.rs
@@ -51,7 +51,7 @@ use hudi_core::config::read::HudiReadConfig::{InputPartitions, UseReadOptimizedM
use hudi_core::config::table::{BaseFileFormatValue, HudiTableConfig};
use hudi_core::config::util::empty_options;
use hudi_core::storage::util::{get_scheme_authority, join_url_segments};
-use hudi_core::table::Table as HudiTable;
+use hudi_core::table::{ReadOptions, Table as HudiTable};
/// Create a `HudiDataSource`.
/// Used for Datafusion to query Hudi tables
@@ -371,11 +371,16 @@ impl TableProvider for HudiDataSource {
.cloned()
.collect();
let pushdown_filters = exprs_to_filters(&partition_filters);
- let file_slices = self
+ let read_options = ReadOptions::new()
+ .with_filters(pushdown_filters)
+ .map_err(|e| Execution(format!("Invalid pushdown filter: {e}")))?;
+ let flat_slices = self
.table
- .get_file_slices_splits(input_partitions, pushdown_filters)
+ .get_file_slices(&read_options)
.await
.map_err(|e| Execution(format!("Failed to get file slices from Hudi table: {e}")))?;
+ let file_slices =
+ hudi_core::util::collection::split_into_chunks(flat_slices, input_partitions);
let base_url = self.table.base_url();
let mut parquet_file_groups: Vec<Vec<PartitionedFile>> = Vec::new();
for file_slice_vec in file_slices {
diff --git a/crates/datafusion/src/util/expr.rs b/crates/datafusion/src/util/expr.rs
index 8a67876a..36261c97 100644
--- a/crates/datafusion/src/util/expr.rs
+++ b/crates/datafusion/src/util/expr.rs
@@ -62,7 +62,7 @@ use log::{debug, warn};
/// * `exprs` - A slice of DataFusion expressions to convert
///
/// # Returns
-/// A vector of filter tuples `(field_name, operator, value)`. All returned filters
+/// A vector of filter tuples `(field, operator, value)`. All returned filters
/// are implicitly AND-ed together.
pub fn exprs_to_filters(exprs: &[Expr]) -> Vec<(String, String, String)> {
exprs
@@ -123,7 +123,7 @@ fn binary_expr_to_filter(binary_expr: &BinaryExpr) -> Option<HudiFilter> {
Some(filter)
}
-/// Converts a NOT expression (`Expr::Not`) into a `PartitionFilter`.
+/// Converts a NOT expression (`Expr::Not`) into a [`HudiFilter`].
fn not_expr_to_filter(not_expr: &Expr) -> Option<HudiFilter> {
match not_expr {
Expr::BinaryExpr(binary_expr) => {
@@ -246,14 +246,14 @@ mod tests {
assert_eq!(result.len(), 1);
let expected_filter = HudiFilter {
- field_name: schema.field(0).name().to_string(),
+ field: schema.field(0).name().to_string(),
operator: ExprOperator::Eq,
values: vec!["42".to_string()],
};
assert_eq!(
result[0],
(
- expected_filter.field_name,
+ expected_filter.field,
expected_filter.operator.to_string(),
expected_filter.values.join(",")
)
@@ -278,14 +278,14 @@ mod tests {
assert_eq!(result.len(), 1);
let expected_filter = HudiFilter {
- field_name: schema.field(0).name().to_string(),
+ field: schema.field(0).name().to_string(),
operator: ExprOperator::Ne,
values: vec!["42".to_string()],
};
assert_eq!(
result[0],
(
- expected_filter.field_name,
+ expected_filter.field,
expected_filter.operator.to_string(),
expected_filter.values.join(",")
)
@@ -299,7 +299,7 @@ mod tests {
(
col("int32_col").eq(lit(42i32)),
Some(HudiFilter {
- field_name: String::from("int32_col"),
+ field: String::from("int32_col"),
operator: ExprOperator::Eq,
values: vec![String::from("42")],
}),
@@ -307,7 +307,7 @@ mod tests {
(
col("int64_col").gt_eq(lit(100i64)),
Some(HudiFilter {
- field_name: String::from("int64_col"),
+ field: String::from("int64_col"),
operator: ExprOperator::Gte,
values: vec![String::from("100")],
}),
@@ -315,7 +315,7 @@ mod tests {
(
col("float64_col").lt(lit(32.666)),
Some(HudiFilter {
- field_name: String::from("float64_col"),
+ field: String::from("float64_col"),
operator: ExprOperator::Lt,
values: vec!["32.666".to_string()],
}),
@@ -323,7 +323,7 @@ mod tests {
(
col("string_col").not_eq(lit("test")),
Some(HudiFilter {
- field_name: String::from("string_col"),
+ field: String::from("string_col"),
operator: ExprOperator::Ne,
values: vec![String::from("test")],
}),
@@ -343,7 +343,7 @@ mod tests {
assert_eq!(
result,
&(
- expected_filter.field_name.clone(),
+ expected_filter.field.clone(),
expected_filter.operator.to_string(),
expected_filter.values.join(",").clone()
)
@@ -377,14 +377,14 @@ mod tests {
assert_eq!(result.len(), 1);
let expected_filter = HudiFilter {
- field_name: schema.field(0).name().to_string(),
+ field: schema.field(0).name().to_string(),
operator: expected_op,
values: vec![String::from("42")],
};
assert_eq!(
result[0],
(
- expected_filter.field_name,
+ expected_filter.field,
expected_filter.operator.to_string(),
expected_filter.values.join(",")
)
diff --git a/crates/datafusion/tests/plan_tests.rs b/crates/datafusion/tests/plan_tests.rs
index 56f3c6bd..47156edc 100644
--- a/crates/datafusion/tests/plan_tests.rs
+++ b/crates/datafusion/tests/plan_tests.rs
@@ -247,6 +247,24 @@ mod v6_tests {
}
}
+ #[tokio::test]
+ async fn test_datafusion_read_hudi_table_with_partition_column_filter() {
+ let test_table = V6SimplekeygenNonhivestyle;
+ let ctx = register_table_direct(&test_table, [(InputPartitions, "2")])
+ .await
+ .unwrap();
+
+ let sql = format!(
+ "SELECT id FROM {} WHERE byteField = 20 ORDER BY id",
+ test_table.as_ref()
+ );
+ let df = ctx.sql(&sql).await.unwrap();
+ let rb = df.collect().await.unwrap();
+ let rb = rb.first().unwrap();
+
+ assert_eq!(get_i32_column(rb, "id"), &[2]);
+ }
+
#[tokio::test]
async fn test_datafusion_read_hudi_table_with_replacecommits_with_partition_filter_pushdown() {
for (test_table, use_sql, planned_input_partitions) in
diff --git a/demo/apps/hudi-file-group-api/cpp/main.cpp b/demo/apps/hudi-file-group-api/cpp/main.cpp
index 568725e4..9f434246 100644
--- a/demo/apps/hudi-file-group-api/cpp/main.cpp
+++ b/demo/apps/hudi-file-group-api/cpp/main.cpp
@@ -142,7 +142,8 @@ int main() {
auto file_group_reader = new_file_group_reader_with_options(base_uri, opts);
auto base_file_path = "a079bdb3-731c-4894-b855-abfcd6921007-0_0-203-274_20240418173551906.parquet";
- ArrowArrayStream* stream_ptr = file_group_reader->read_file_slice_by_base_file_path(base_file_path);
+ std::vector<std::string> log_file_paths{};
+ ArrowArrayStream* stream_ptr = file_group_reader->read_file_slice_from_paths(base_file_path, log_file_paths);
if (!stream_ptr) {
std::cerr << "Error: Received null stream pointer" << std::endl;
diff --git a/demo/apps/hudi-table-api/python/src/main.py b/demo/apps/hudi-table-api/python/src/main.py
index 15e19e9e..211d46f2 100644
--- a/demo/apps/hudi-table-api/python/src/main.py
+++ b/demo/apps/hudi-table-api/python/src/main.py
@@ -24,7 +24,7 @@ for url in [
"s3://hudi-demo/mor/parquet/v6_complexkeygen_hivestyle",
]:
hudi_table = HudiTableBuilder.from_base_uri(url).build()
- batches = hudi_table.read_snapshot()
+ batches = hudi_table.read()
arrow_table = pa.Table.from_batches(batches)
assert arrow_table.schema.names == [
diff --git a/demo/apps/hudi-table-api/rust/src/main.rs b/demo/apps/hudi-table-api/rust/src/main.rs
index 056c247e..823b281a 100644
--- a/demo/apps/hudi-table-api/rust/src/main.rs
+++ b/demo/apps/hudi-table-api/rust/src/main.rs
@@ -18,7 +18,7 @@
*/
use arrow::compute::concat_batches;
-use hudi::config::util::empty_filters;
+use hudi::table::ReadOptions;
use hudi::error::Result;
use hudi::table::builder::TableBuilder as HudiTableBuilder;
@@ -29,7 +29,7 @@ async fn main() -> Result<()> {
"s3://hudi-demo/mor/parquet/v6_complexkeygen_hivestyle",
] {
let hudi_table = HudiTableBuilder::from_base_uri(url).build().await?;
- let batches = hudi_table.read_snapshot(empty_filters()).await?;
+ let batches = hudi_table.read(&ReadOptions::new()).await?;
let batch = concat_batches(&batches[0].schema(), &batches)?;
assert_eq!(
diff --git a/docs/reader-spec.md b/docs/reader-spec.md
new file mode 100644
index 00000000..e6b99b44
--- /dev/null
+++ b/docs/reader-spec.md
@@ -0,0 +1,319 @@
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. The ASF licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~ with the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing,
+ ~ software distributed under the License is distributed on an
+ ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ ~ KIND, either express or implied. See the License for the
+ ~ specific language governing permissions and limitations
+ ~ under the License.
+-->
+
+# hudi-rs Reader Spec
+
+The public reader APIs hudi-rs exposes for Apache Hudi tables: the methods callers can call, the options they accept, and what callers can expect back.
+
+For tutorials and end-to-end examples, see the [README](../README.md). Covers Rust core (`hudi`) and the Python bindings (`hudi`); writer APIs, DataFusion, and C++ bindings are out of scope here.
+
+## Contents
+
+1. [Query types](#1-query-types)
+2. [`ReadOptions`](#2-readoptions)
+3. [Filter expressions](#3-filter-expressions)
+4. [Rust API](#4-rust-api)
+5. [Python API](#5-python-api)
+6. [Caller expectations](#6-caller-expectations)
+7. [Stability and out of scope](#7-stability-and-out-of-scope)
+
+## 1. Query types
+
+The two `QueryType` variants on `ReadOptions`:
+
+- **Snapshot** β latest table state at one commit (the latest by default; an explicit `as_of_timestamp` for time-travel).
+- **Incremental** β records changed in the half-open range (`start_timestamp`, `end_timestamp`].
+
+Snapshot reads have an **eager** form returning all batches and a **streaming** form yielding batches as they're read. Incremental reads currently expose only the eager form.
+
+Per-slice reads β reading a single `FileSlice` the caller already selected (typically obtained from `get_file_slices` with the right options) β are not a query type; they live on `FileGroupReader` and have both eager and streaming forms.
+
+## 2. `ReadOptions`
+
+`ReadOptions` (Rust) / `HudiReadOptions` (Python) is the single config object accepted by every reader API. The struct stores three fields β `filters`, `projection`, and `hudi_options`. Convenience builders (`with_query_type`, `with_as_of_timestamp`, `with_start_timestamp`, `with_end_timestamp`, `with_batch_size`) translate into `hudi_options` under the matching `HudiReadConfig` key, so `hudi_options` is the single source of truth for per-read Hudi configs. Typed accessors (`query_type() [...]
+
+| Stored field | Type | Default | Notes |
+|----------------|-------------------------------|---------|----------------------------------------------------|
+| `filters` | `Vec<(field, op, value)>` | empty | pruning + row-level mask |
+| `projection` | `Option<Vec<String>>` | None | all columns when None |
+| `hudi_options` | `HashMap<String, String>` | empty | per-read `hoodie.*` overrides |
+
+| Convenience builder | Stored under `hudi_options` key | Default at use time |
+|----------------------------|------------------------------------------------|-------------------------------------------|
+| `with_query_type(QueryType)` | `hoodie.read.query.type` | `snapshot` |
+| `with_as_of_timestamp(ts)` | `hoodie.read.as.of.timestamp` | latest commit (Snapshot only) |
+| `with_start_timestamp(ts)` | `hoodie.read.start.timestamp` | `19700101000000000` (Incremental only) |
+| `with_end_timestamp(ts)` | `hoodie.read.end.timestamp` | latest commit (Incremental only) |
+| `with_batch_size(n)` | `hoodie.read.stream.batch_size` | `1024` (streaming only) |
+
+Which knobs each API consumes:
+
+| API | query type | as-of | start/end | filters | projection | batch size | hudi_options pass-through |
+|------------------------------------------------------------------|:----------:|:-----:|:---------:|:-------:|:----------:|:----------:|:-------------------------:|
+| `read` / `read_stream` | yes | when Snapshot | when Incremental | yes | yes | streaming | yes |
+| `get_file_slices` | yes | when Snapshot | when Incremental | yes | β | β | β |
+| `FileGroupReader::read_file_slice` / `_from_paths` | β | β | β | yes | yes | β | β |
+| `FileGroupReader::read_file_slice_stream` / `_from_paths_stream` | β | β | β | yes | yes | yes | β |
+
+Notes:
+
+- `read_stream` errors with `Unsupported` for `query_type = Incremental` β incremental streaming is not yet implemented.
+- The `hudi_options` bag is a per-read override layer β set arbitrary `hoodie.*` configs (e.g. `hoodie.read.use.read_optimized.mode = true`) without mutating the `Table`.
+- Per-slice reads are exposed only by `FileGroupReader`. The `Table` type owns logical reads (snapshot, incremental); per-slice reads are physical and belong at the file-group layer. To read one slice with table-level configs, build a `FileGroupReader` via `Table::create_file_group_reader_with_options` and call its per-slice methods.
+- For parallel reads, call `get_file_slices(...)` and bucket the result with `hudi::util::collection::split_into_chunks` or your engine's preferred partitioning policy.
+
+Timestamp formats are documented in [Β§6](#timestamps).
+
+## 3. Filter expressions
+
+A filter is a `(field, operator, value)` tuple of strings.
+
+| Operator (case-insensitive) | Cardinality |
+|-------------------------------------|:-----------:|
+| `=` `!=` `<` `<=` `>` `>=` | 1 |
+| `IN` `NOT IN` | β₯1 |
+
+For `IN` / `NOT IN`, the value string is split on unescaped commas and trimmed: `("city", "IN", "sf,la,nyc")`. `\,` is a literal comma and `\\` is a literal backslash, so values that contain commas survive: `("name", "IN", "Smith\\, John,Jane")` parses to `["Smith, John", "Jane"]`.
+
+`with_filters` parses and cardinality-validates upfront; an unrecognized operator or empty `IN` / `NOT IN` value list errors at the builder rather than at read time.
+
+The `field` may be any column. Filters drive three things:
+
+- **Partition pruning** when the field is a partition column. Always applied.
+- **File-level stats pruning** when the field is a data column with min/max stats in the metadata table. Snapshot/time-travel only β incremental file planning does not stats-prune.
+- **Row-level mask** applied to every returned batch. This is the authoritative filter; pruning is best-effort.
+
+Values are strings; they are cast to the target column's Arrow type at filter time. Unparseable values (e.g. `"abc"` against `Int64`) error.
+
+A filter on an unknown column errors before any rows are returned. `Table` / `HudiTable` paths validate up front against the loaded schemas. `FileGroupReader` paths validate strictly against the read batch schema on the first batch β direct callers must not pass filters on columns absent from the parquet (notably partition columns when `hoodie.datasource.write.drop.partition.columns = true`); the table-level paths strip such filters automatically after using them for pruning.
+
+## 4. Rust API
+
+All public symbols are re-exported from the `hudi` crate.
+
+### `Table`
+
+| Method | Returns |
+|----------------------------------------------------------------------------|------------------------------------------------------|
+| `Table::new(base_uri)` | `Result<Table>` |
+| `Table::new_with_options(base_uri, options)` | `Result<Table>` |
+| `hudi_options()` / `storage_options()` | `HashMap<String, String>` |
+| `base_url()` / `table_name()` / `table_type()` / `is_mor()` / `timezone()` | `Url` / `String` / `String` / `bool` / `String` |
+| `get_schema()` / `get_schema_with_meta_fields()` | `Result<Schema>` |
+| `get_schema_in_avro_str()` / `get_schema_in_avro_str_with_meta_fields()` | `Result<String>` |
+| `get_partition_schema()` | `Result<Schema>` |
+| `get_timeline()` | `&Timeline` |
+| `get_file_slices(&ReadOptions)` | `Result<Vec<FileSlice>>` (dispatches on `query_type`) |
+| `create_file_group_reader_with_options(read_options, extra_hudi_overrides, extra_storage_overrides)` | `Result<FileGroupReader>` |
+| `read(&ReadOptions)` | `Result<Vec<RecordBatch>>` (dispatches on `query_type`) |
+| `read_stream(&ReadOptions)` | `Result<BoxStream<'static, Result<RecordBatch>>>` (errors on `Incremental`) |
+| `compute_table_stats()` | `Option<(u64, u64)>` β `(rows, byte_size)` |
+
+### `FileGroupReader`
+
+| Method | Returns |
+|-------------------------------------------------------------------------|------------------------------------------------------|
+| `FileGroupReader::new_with_options(base_uri, options)` | `Result<FileGroupReader>` |
+| `read_file_slice(&FileSlice, &ReadOptions)` | `Result<RecordBatch>` (base + merge logs) |
+| `read_file_slice_from_paths(base_path, log_paths, &ReadOptions)` | `Result<RecordBatch>` (pass empty log_paths for base-only) |
+| `read_file_slice_stream(&FileSlice, &ReadOptions)` | `Result<BoxStream<'static, Result<RecordBatch>>>` |
+| `read_file_slice_from_paths_stream(base_path, log_paths, &ReadOptions)` | `Result<BoxStream<'static, Result<RecordBatch>>>` |
+| `is_metadata_table()` | `bool` |
+
+### `ReadOptions` builder
+
+`with_filters` and `with_batch_size` validate eagerly and return `Result<Self>`; the others are infallible. Chains intermix with `?` propagation:
+
+```rust
+let options = ReadOptions::new()
+ .with_query_type(QueryType::Snapshot)
+ .with_filters([("city", "=", "san_francisco")])?
+ .with_projection(["rider", "city", "ts", "fare"])
+ .with_batch_size(4096)?
+ .with_as_of_timestamp("20240101000000000")
+ .with_start_timestamp("20240101000000000")
+ .with_end_timestamp("20240201000000000")
+ .with_hudi_option("hoodie.read.use.read_optimized.mode", "true");
+```
+
+`with_batch_size(0)` errors at the builder (a zero-row batch yields no batches at the parquet stream reader). `with_filters` parses + cardinality-validates upfront; an unrecognized operator or empty `IN` / `NOT IN` value list errors here rather than at read time.
+
+### `TableBuilder`
+
+```rust
+use hudi::table::builder::TableBuilder as HudiTableBuilder;
+
+let table = HudiTableBuilder::from_base_uri("/tmp/trips_table")
+ .with_hudi_option("hoodie.read.use.read_optimized.mode", "true")
+ .with_storage_option("aws_region", "us-west-2")
+ .build()
+ .await?;
+```
+
+Available pairs: `with_hudi_option` / `with_hudi_options`, `with_storage_option` / `with_storage_options`, `with_option` / `with_options` (the generic forms route by key prefix).
+
+### Filter, Timeline, FileSlice
+
+| Item | Notes |
+|---------------------------------------------------------------------------------------------------------------|--------------------------------------------------------|
+| `Filter { field, operator, values }`, `Filter::new(...)`, `Filter::negate()` | One column predicate; cardinality-validated. |
+| `from_str_tuples(tuples)` | Parse `(&str, &str, &str)` tuples into `Vec<Filter>`. |
+| `enum ExprOperator { Eq, Ne, Lt, Lte, Gt, Gte, In, NotIn }` | Comparison operators. |
+| `col(name).eq / ne / lt / lte / gt / gte / in_list / not_in_list` | DSL for building filters. |
+| `Timeline::get_completed_commits(desc)` and `..._deltacommits` / `_replacecommits` / `_clustering_commits` | Successful instants. `desc` reverses order. |
+| `Timeline::get_latest_commit_timestamp()` | Latest commit timestamp. |
+| `Timeline::get_instant_metadata_in_json(&Instant)` | Commit metadata for one instant. |
+| `Timeline::get_latest_avro_schema()` / `get_latest_schema()` | Latest schema (Avro string / Arrow `Schema`). |
+| `FileSlice::file_id()` / `creation_instant_time()` / `has_log_file()` | Slice identity / version / MOR-with-deltas flag. |
+| `FileSlice::base_file_relative_path()` / `log_file_relative_path(&LogFile)` | Paths relative to the table base URI. |
+
+## 5. Python API
+
+All symbols are exported from the top-level `hudi` package.
+
+### `HudiTableBuilder`
+
+```python
+from hudi import HudiTableBuilder
+
+table = (
+ HudiTableBuilder
+ .from_base_uri("/tmp/trips_table")
+ .with_hudi_option("hoodie.read.use.read_optimized.mode", "true")
+ .with_storage_option("aws_region", "us-west-2")
+ .build()
+)
+```
+
+`with_hudi_option` and `with_option` accept a string key or a `HudiReadConfig` / `HudiTableConfig` enum member. The bulk variants (`with_hudi_options`, `with_options`) currently accept dicts of string keys.
+
+### `HudiTable`
+
+| Method / property | Returns |
+|----------------------------------------------------------------------------------------------------|------------------------------------------|
+| `HudiTable(base_uri, options=None)` | `HudiTable` |
+| `hudi_options()` / `storage_options()` | `Dict[str, str]` |
+| properties: `table_name`, `table_type`, `is_mor`, `timezone`, `base_url` | table-level metadata |
+| `get_schema()` / `get_schema_with_meta_fields()` / `get_partition_schema()` | `pyarrow.Schema` |
+| `get_schema_in_avro_str()` / `get_schema_in_avro_str_with_meta_fields()` | `str` |
+| `get_timeline()` | `HudiTimeline` |
+| `get_file_slices(options=None)` | `List[HudiFileSlice]` (dispatches on `options.query_type`) |
+| `create_file_group_reader_with_options(read_options=None, extra_hudi_overrides=None, extra_storage_overrides=None)` | `HudiFileGroupReader` |
+| `read(options=None)` | `List[pyarrow.RecordBatch]` (dispatches on `query_type`) |
+| `read_stream(options=None)` | `HudiRecordBatchStream` (errors on `Incremental`) |
+| `compute_table_stats()` | `Optional[Tuple[int, int]]` |
+
+### `HudiFileGroupReader`
+
+| Method | Returns |
+|----------------------------------------------------------------------------|----------------------------------|
+| `HudiFileGroupReader(base_uri, options=None)` | `HudiFileGroupReader` |
+| `read_file_slice(file_slice, options=None)` | `pyarrow.RecordBatch` |
+| `read_file_slice_from_paths(base_path, log_paths, options=None)` | `pyarrow.RecordBatch` (pass empty log_paths for base-only) |
+| `read_file_slice_stream(file_slice, options=None)` | `HudiRecordBatchStream` |
+| `read_file_slice_from_paths_stream(base_path, log_paths, options=None)` | `HudiRecordBatchStream` |
+| property: `is_metadata_table` | `bool` |
+
+### `HudiReadOptions`
+
+```python
+from hudi import HudiQueryType, HudiReadOptions
+
+# Constructor takes only the three stored fields; everything else is set via builders.
+options = (
+ HudiReadOptions(
+ filters=[("city", "=", "san_francisco")],
+ projection=["rider", "city", "ts", "fare"],
+ hudi_options={"hoodie.read.use.read_optimized.mode": "true"},
+ )
+ .with_query_type(HudiQueryType.Snapshot) # or HudiQueryType.Incremental
+ .with_batch_size(4096)
+ .with_as_of_timestamp("20240101000000000")
+ .with_start_timestamp("20240101000000000")
+ .with_end_timestamp("20240201000000000")
+)
+```
+
+All builders return a new `HudiReadOptions` for chaining. Typed accessors
+(`query_type()`, `as_of_timestamp()`, `start_timestamp()`, `end_timestamp()`,
+`batch_size()`) read back from the bag. Defaults match [Β§2](#2-readoptions).
+
+`with_batch_size(0)` raises immediately, and `with_filters` parses + cardinality-validates upfront β an unrecognized operator or empty `IN` / `NOT IN` value list raises here rather than at read time. The constructor's `filters` argument has the same eager validation.
+
+### `HudiRecordBatchStream`
+
+A single-use iterator returned by streaming APIs. `for batch in stream:` or `next(stream)`. Each yielded value is a `pyarrow.RecordBatch`.
+
+### `HudiTimeline`, `HudiInstant`, `HudiFileSlice`
+
+| Item | Notes |
+|--------------------------------------------------------------------------------------------------|--------------------------------------------------------|
+| `HudiTimeline.get_completed_commits(desc=False)` and `..._deltacommits` / `_replacecommits` / `_clustering_commits` | `List[HudiInstant]`. `desc` reverses order. |
+| `HudiTimeline.get_instant_metadata_in_json(instant)` | `str` |
+| `HudiTimeline.get_latest_commit_timestamp()` | `str` |
+| `HudiTimeline.get_latest_avro_schema()` / `get_latest_schema()` | `str` / `pyarrow.Schema` |
+| `HudiInstant` properties: `timestamp`, `action`, `state`, `epoch_mills` | read-only |
+| `HudiFileSlice` attributes: `file_id`, `partition_path`, `creation_instant_time`, `base_file_name`, `base_file_size`, `base_file_byte_size`, `log_file_names`, `num_records` | read-only |
+| `HudiFileSlice.base_file_relative_path()` / `log_files_relative_paths()` | `str` / `List[str]` |
+
+## 6. Caller expectations
+
+### Snapshot atomicity
+
+A snapshot read is pinned to one completed commit timestamp; mid-read writes are not visible. The timeline is loaded once when `Table` / `HudiTable` is constructed and reused for every subsequent read on that instance β commits that land *after* construction require building a new instance to observe. When `as_of_timestamp` is unset, "latest commit" means the latest commit in the cached timeline, not in storage at call time.
+
+### Incremental semantics
+
+The range is half-open: (`start_timestamp`, `end_timestamp`]. A record updated multiple times within the range yields its latest in-range state, not the full update history. Records updated only outside the range are not returned.
+
+### MOR streaming fallback
+
+Streaming yields true streaming batches when the slice is base-file-only or `hoodie.read.use.read_optimized.mode = true`. For MOR slices with log files, the implementation collects-and-merges and yields the result as a single batch on the stream.
+
+### `batch_size` and `projection`
+
+`batch_size` controls rows per batch for streaming reads (default 1024); eager reads return one merged batch per file slice and ignore `batch_size`. Streaming pushes `projection` down to the parquet reader; eager reads project after merging. When `projection` is combined with `filters` on data columns not in `projection`, the read transparently widens to read those columns, then projects back down after the filter mask runs.
+
+### Timestamps
+
+`as_of_timestamp`, `start_timestamp`, and `end_timestamp` accept:
+
+- Hudi timeline format (highest precedence): `yyyyMMddHHmmssSSS` or `yyyyMMddHHmmss`.
+- Unix epoch in seconds, milliseconds, microseconds, or nanoseconds.
+- RFC 3339 with timezone offset: `2024-03-15T14:25:30Z`, `2024-03-15T14:25:30+00:00`, `2024-03-15T14:25:30.123Z`.
+
+A timezone offset (`Z` or `Β±HH:MM`) is required for RFC 3339 inputs β naive `T`-separated strings and date-only strings are rejected. Inputs are normalized into `hoodie.table.timeline.timezone`.
+
+### Empty results
+
+A table with no completed commits yields empty `Vec` / `List` for eager reads and `get_file_slices`, and an empty stream for `read_stream`.
+
+### Errors
+
+- Unknown filter column: schema error before any rows are returned. `Table` paths validate up front; `FileGroupReader` paths validate strictly on the first batch.
+- Unparseable filter value (e.g. `"abc"` against `Int64`): schema error at filter evaluation.
+- I/O failures (missing files, permission errors, malformed parquet): read errors.
+
+## 7. Stability and out of scope
+
+Reader APIs documented here are the supported public surface as of this release. The `FileGroupReader` direct-paths APIs are still labeled experimental in the README; expect minor signature evolution before they finalize.
+
+`compute_table_stats()` returns `None` when statistics cannot be computed (no metadata table, non-Parquet base files, or footer reads fail). It does not yet account for log files in MOR tables, so estimates skew low for write-heavy MOR workloads.
+
+Out of scope for this version: writer APIs, internal architecture (timeline parsing, log-record merging, metadata table layout), the full configuration key glossary (`HudiReadConfig` / `HudiTableConfig` members), and the DataFusion and C++ bindings β separate spec follow-ups.
diff --git a/python/hudi/__init__.py b/python/hudi/__init__.py
index 9a3cb04a..83ccad67 100644
--- a/python/hudi/__init__.py
+++ b/python/hudi/__init__.py
@@ -22,6 +22,9 @@ from hudi._internal import (
HudiFileGroupReader,
HudiFileSlice,
HudiInstant,
+ HudiQueryType,
+ HudiReadOptions,
+ HudiRecordBatchStream,
HudiTable,
HudiTimeline,
)
@@ -33,7 +36,10 @@ __all__ = [
"HudiFileGroupReader",
"HudiFileSlice",
"HudiInstant",
+ "HudiQueryType",
"HudiReadConfig",
+ "HudiReadOptions",
+ "HudiRecordBatchStream",
"HudiTable",
"HudiTableBuilder",
"HudiTableConfig",
diff --git a/python/hudi/_internal.pyi b/python/hudi/_internal.pyi
index 19c18c92..e4f50d48 100644
--- a/python/hudi/_internal.pyi
+++ b/python/hudi/_internal.pyi
@@ -15,14 +15,89 @@
# specific language governing permissions and limitations
# under the License.
from dataclasses import dataclass
-from typing import Dict, List, Optional, Tuple
+from typing import ClassVar, Dict, List, Optional, Tuple
-import pyarrow # type: ignore
+import pyarrow # type: ignore[import-untyped]
__version__: str
def _config_keys() -> Dict[str, List[Tuple[str, str]]]: ...
+class HudiQueryType:
+ """Selects the read semantic on :class:`HudiReadOptions`."""
+
+ Snapshot: ClassVar["HudiQueryType"]
+ Incremental: ClassVar["HudiQueryType"]
+
+ @property
+ def name(self) -> str: ...
+ @property
+ def value(self) -> str: ...
+
+@dataclass(init=False)
+class HudiReadOptions:
+ """
+ Options for all Hudi read APIs (snapshot, time-travel, incremental, eager and streaming).
+
+ Mirrors the Rust ``ReadOptions`` struct shape: only three stored attributes β
+ ``filters``, ``projection``, ``hudi_options``. All other knobs (query type,
+ timestamps, batch size) are set via the chainable ``with_*`` builders, which
+ insert into ``hudi_options`` under the matching ``HudiReadConfig`` key.
+ Typed accessors (``query_type()``, ``as_of_timestamp()``, etc.) read back
+ from the bag.
+
+ Attributes:
+ filters (List[Tuple[str, str, str]]): Column filters as ``(field, op, value)`` tuples.
+ projection (Optional[List[str]]): Column names to read. If None, all columns are read.
+ hudi_options (Dict[str, str]): Per-read Hudi configs (``hoodie.*``).
+ """
+
+ filters: List[Tuple[str, str, str]]
+ projection: Optional[List[str]]
+ hudi_options: Dict[str, str]
+
+ def __init__(
+ self,
+ filters: Optional[List[Tuple[str, str, str]]] = None,
+ projection: Optional[List[str]] = None,
+ hudi_options: Optional[Dict[str, str]] = None,
+ ): ...
+
+ # Builders return a new HudiReadOptions for chaining.
+ def with_query_type(self, query_type: HudiQueryType) -> "HudiReadOptions": ...
+ def with_as_of_timestamp(self, timestamp: str) -> "HudiReadOptions": ...
+ def with_start_timestamp(self, timestamp: str) -> "HudiReadOptions": ...
+ def with_end_timestamp(self, timestamp: str) -> "HudiReadOptions": ...
+ def with_batch_size(self, size: int) -> "HudiReadOptions":
+ """Target rows per batch for streaming reads. Raises ``HudiCoreError`` if ``size == 0``."""
+ ...
+ def with_filters(self, filters: List[Tuple[str, str, str]]) -> "HudiReadOptions":
+ """Set column filters as ``(field, op, value)`` tuples. Parses and cardinality-validates immediately; an unrecognized operator or empty ``IN`` / ``NOT IN`` value list raises ``HudiCoreError``."""
+ ...
+ def with_projection(self, columns: List[str]) -> "HudiReadOptions": ...
+ def with_hudi_option(self, key: str, value: str) -> "HudiReadOptions": ...
+ def with_hudi_options(self, opts: Dict[str, str]) -> "HudiReadOptions": ...
+
+ # Typed accessors read from hudi_options.
+ def query_type(self) -> HudiQueryType: ...
+ def as_of_timestamp(self) -> Optional[str]: ...
+ def start_timestamp(self) -> Optional[str]: ...
+ def end_timestamp(self) -> Optional[str]: ...
+ def batch_size(self) -> Optional[int]:
+ """Target batch size for streaming reads, or ``None`` if unset. Raises ``HudiCoreError`` if the stored value is malformed."""
+ ...
+
+@dataclass(init=False)
+class HudiRecordBatchStream:
+ """
+ A stream of record batches yielded one at a time.
+
+ Iterating consumes the stream. The stream is single-use.
+ """
+
+ def __iter__(self) -> "HudiRecordBatchStream": ...
+ def __next__(self) -> "pyarrow.RecordBatch": ...
+
@dataclass(init=False)
class HudiFileGroupReader:
"""
@@ -41,43 +116,56 @@ class HudiFileGroupReader:
options (Optional[Dict[str, str]]): Additional configuration options (optional).
"""
...
- def read_file_slice_by_base_file_path(
- self, relative_path: str
+ def read_file_slice(
+ self,
+ file_slice: HudiFileSlice,
+ options: Optional[HudiReadOptions] = None,
) -> "pyarrow.RecordBatch":
"""
- Reads the data from the base file at the given relative path.
-
- Parameters:
- relative_path (str): The relative path to the base file.
-
- Returns:
- pyarrow.RecordBatch: A record batch read from the base file.
+ Read the data from the given file slice (base + log files merged).
"""
...
- def read_file_slice(self, file_slice: HudiFileSlice) -> "pyarrow.RecordBatch":
+
+ def read_file_slice_from_paths(
+ self,
+ base_file_path: str,
+ log_file_paths: List[str],
+ options: Optional[HudiReadOptions] = None,
+ ) -> "pyarrow.RecordBatch":
+ """
+ Read a file slice from a base file path and a list of log file paths.
"""
- Reads the data from the given file slice.
+ ...
- Parameters:
- file_slice (HudiFileSlice): The file slice to read from.
+ def read_file_slice_stream(
+ self,
+ file_slice: HudiFileSlice,
+ options: Optional[HudiReadOptions] = None,
+ ) -> HudiRecordBatchStream:
+ """
+ Reads a file slice as a stream of record batches.
- Returns:
- pyarrow.RecordBatch: A record batch read from the file slice.
+ For COW tables or read-optimized mode, this yields batches as they are read
+ without loading all data into memory. For MOR tables with log files, this
+ currently falls back to a single merged batch.
"""
...
- def read_file_slice_from_paths(
- self, base_file_path: str, log_file_paths: List[str]
- ) -> "pyarrow.RecordBatch":
+ def read_file_slice_from_paths_stream(
+ self,
+ base_file_path: str,
+ log_file_paths: List[str],
+ options: Optional[HudiReadOptions] = None,
+ ) -> HudiRecordBatchStream:
"""
- Read a file slice from a base file and a list of log files.
-
- Args:
- base_file_path (str): The relative path to the base file.
- log_file_paths (List[str]): A list of relative paths to log files.
+ Reads a file slice from explicit paths as a stream of record batches.
+ """
+ ...
- Returns:
- pyarrow.RecordBatch: The merged record batch from base file and log files.
+ @property
+ def is_metadata_table(self) -> bool:
+ """
+ Whether this reader targets a metadata table (base path ends with ``.hoodie/metadata``).
"""
...
@@ -193,12 +281,12 @@ class HudiTable:
"""
...
@property
- def is_mor(self) -> str:
+ def is_mor(self) -> bool:
"""
Get whether the table is an MOR table.
Returns:
- str: True if the table is a MOR table, False otherwise.
+ bool: True if the table is a MOR table, False otherwise.
"""
...
@property
@@ -255,101 +343,74 @@ class HudiTable:
Returns the timeline of the Hudi table.
"""
...
- def get_file_slices_splits(
- self, num_splits: int, filters: Optional[List[Tuple[str, str, str]]]
- ) -> List[List[HudiFileSlice]]:
- """
- Retrieves all file slices in the Hudi table in 'num_splits' splits, optionally filtered by given filters.
-
- Parameters:
- num_splits (int): The number of parts to split the file slices into.
- filters (Optional[List[Tuple[str, str, str]]]): Optional filters for selecting file slices.
-
- Returns:
- List[List[HudiFileSlice]]: A list of file slice groups, each group being a list of HudiFileSlice objects.
- """
- ...
- def get_file_slices_splits_as_of(
- self,
- num_splits: int,
- timestamp: str,
- filters: Optional[List[Tuple[str, str, str]]],
- ) -> List[List[HudiFileSlice]]:
+ @property
+ def base_url(self) -> str:
"""
- Retrieves all file slices in the Hudi table as of a timestamp in 'num_splits' splits, optionally filtered by given filters.
+ Get the base URL of the Hudi table.
"""
...
def get_file_slices(
- self, filters: Optional[List[Tuple[str, str, str]]]
+ self, options: Optional[HudiReadOptions] = None
) -> List[HudiFileSlice]:
"""
- Retrieves all file slices in the Hudi table, optionally filtered by the provided filters.
-
- Parameters:
- filters (Optional[List[Tuple[str, str, str]]]): Optional filters for selecting file slices.
+ Get the file slices the read targets, dispatching on ``options.query_type``.
- Returns:
- List[HudiFileSlice]: A list of file slices matching the filters.
- """
- ...
- def get_file_slices_as_of(
- self, timestamp: str, filters: Optional[List[Tuple[str, str, str]]]
- ) -> List[HudiFileSlice]:
- """
- Retrieves all file slices in the Hudi table as of a timestamp, optionally filtered by the provided filters.
- """
- ...
- def get_file_slices_between(
- self,
- start_timestamp: Optional[str],
- end_timestamp: Optional[str],
- ) -> List[HudiFileSlice]:
- """
- Retrieves all changed file slices in the Hudi table between the given timestamps.
+ - Snapshot: slices visible at ``options.as_of_timestamp`` (defaults to the
+ latest commit). ``options.filters`` drive partition + file-level stats
+ pruning.
+ - Incremental: slices changed in ``(options.start_timestamp,
+ options.end_timestamp]``. ``options.filters`` drive partition pruning only.
"""
...
def create_file_group_reader_with_options(
- self, options: Optional[Dict[str, str]] = None
+ self,
+ read_options: Optional[HudiReadOptions] = None,
+ extra_hudi_overrides: Optional[Dict[str, str]] = None,
+ extra_storage_overrides: Optional[Dict[str, str]] = None,
) -> HudiFileGroupReader:
"""
- Creates a HudiFileGroupReader for reading records from file groups in the Hudi table.
+ Create a :class:`HudiFileGroupReader` using the table's Hudi configs.
- Returns:
- HudiFileGroupReader: A reader object for reading file groups.
- """
- ...
- def read_snapshot(
- self, filters: Optional[List[Tuple[str, str, str]]]
- ) -> List["pyarrow.RecordBatch"]:
- """
- Reads the latest snapshot of the Hudi table, optionally filtered by the provided filters.
+ Two override channels keep Hudi configs and storage credentials cleanly
+ separated β a ``hoodie.*`` key can't be misclassified as storage, and a
+ stray storage option can't be silently picked up as a Hudi config.
- Parameters:
- filters (Optional[List[Tuple[str, str, str]]]): Optional filters for selecting file slices.
+ **Hudi configs** (last-writer-wins):
- Returns:
- List[pyarrow.RecordBatch]: A list of record batches from the snapshot of the table.
+ 1. Table-level Hudi configs.
+ 2. ``read_options.hudi_options`` when ``read_options`` is provided, **excluding**
+ the four keys the ``Table`` layer interprets directly (``query_type``,
+ ``as_of_timestamp``, ``start_timestamp``, ``end_timestamp``).
+ 3. ``extra_hudi_overrides`` β caller-supplied resolved Hudi configs; always win.
+
+ **Storage options** (last-writer-wins):
+
+ 1. Table-level storage options (cloud credentials, endpoints, etc).
+ 2. ``extra_storage_overrides`` β caller-supplied per-path storage overrides.
"""
...
- def read_snapshot_as_of(
- self, timestamp: str, filters: Optional[List[Tuple[str, str, str]]]
+ def read(
+ self, options: Optional[HudiReadOptions] = None
) -> List["pyarrow.RecordBatch"]:
"""
- Reads the snapshot of the Hudi table as of a timestamp, optionally filtered by the provided filters.
+ Read records, dispatching on ``options.query_type``.
"""
...
- def read_incremental_records(
- self, start_timestamp: str, end_timestamp: Optional[str]
- ) -> List["pyarrow.RecordBatch"]:
+ def compute_table_stats(self) -> Optional[Tuple[int, int]]:
"""
- Reads incremental records from the Hudi table between the given timestamps.
-
- Parameters:
- start_timestamp (str): The start timestamp (exclusive).
- end_timestamp (Optional[str]): The end timestamp (inclusive).
+ Compute estimated table-level statistics from the metadata table.
Returns:
- List[pyarrow.RecordBatch]: A list of record batches containing incremental records.
+ Optional[Tuple[int, int]]: ``(estimated_num_rows, estimated_total_byte_size)``,
+ or ``None`` if the metadata table is not enabled or statistics cannot be computed.
+ """
+ ...
+ def read_stream(
+ self, options: Optional[HudiReadOptions] = None
+ ) -> HudiRecordBatchStream:
+ """
+ Streaming read; dispatches on ``options.query_type``. Incremental streaming
+ is not yet supported and raises ``HudiCoreError``.
"""
...
diff --git a/python/src/internal.rs b/python/src/internal.rs
index b2ff14ea..298c65f5 100644
--- a/python/src/internal.rs
+++ b/python/src/internal.rs
@@ -18,23 +18,29 @@
*/
use arrow::pyarrow::ToPyArrow;
+use arrow::record_batch::RecordBatch;
+use futures::StreamExt;
+use futures::stream::BoxStream;
use std::collections::HashMap;
use std::convert::From;
use std::path::PathBuf;
-use std::sync::OnceLock;
+use std::sync::{Arc, OnceLock};
use tokio::runtime::Runtime;
+use tokio::sync::Mutex;
#[cfg(feature = "datafusion")]
use datafusion::error::DataFusionError;
+
use hudi::config::read::HudiReadConfig;
use hudi::config::table::HudiTableConfig;
use hudi::error::CoreError;
+use hudi::error::Result as HudiResult;
use hudi::file_group::FileGroup;
use hudi::file_group::file_slice::FileSlice;
use hudi::file_group::reader::FileGroupReader;
use hudi::storage::error::StorageError;
-use hudi::table::Table;
use hudi::table::builder::TableBuilder;
+use hudi::table::{QueryType, ReadOptions, Table};
use hudi::timeline::Timeline;
use hudi::timeline::instant::Instant;
use pyo3::exceptions::PyException;
@@ -42,6 +48,8 @@ use pyo3::prelude::*;
use pyo3::{PyErr, PyResult, Python, create_exception, pyclass, pyfunction, pymethods};
use std::error::Error;
+type RecordBatchBoxStream = BoxStream<'static, HudiResult<RecordBatch>>;
+
create_exception!(_internal, HudiCoreError, PyException);
fn convert_to_py_err<I>(err: I) -> PyErr
@@ -72,6 +80,270 @@ impl From<PythonError> for PyErr {
}
}
+/// Python wrapper around [`hudi::table::QueryType`].
+#[cfg(not(tarpaulin_include))]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+#[pyclass(eq)]
+pub struct HudiQueryType {
+ inner: QueryType,
+}
+
+#[cfg(not(tarpaulin_include))]
+#[pymethods]
+impl HudiQueryType {
+ #[classattr]
+ #[pyo3(name = "Snapshot")]
+ fn snapshot() -> Self {
+ QueryType::Snapshot.into()
+ }
+
+ #[classattr]
+ #[pyo3(name = "Incremental")]
+ fn incremental() -> Self {
+ QueryType::Incremental.into()
+ }
+
+ fn __repr__(&self) -> String {
+ format!("HudiQueryType.{}", self.name())
+ }
+
+ #[getter]
+ fn name(&self) -> String {
+ format!("{:?}", self.inner)
+ }
+
+ #[getter]
+ fn value(&self) -> String {
+ self.inner.as_ref().to_string()
+ }
+}
+
+#[cfg(not(tarpaulin_include))]
+#[derive(Clone, Debug, Default)]
+#[pyclass]
+pub struct HudiReadOptions {
+ inner: ReadOptions,
+}
+
+impl From<QueryType> for HudiQueryType {
+ fn from(inner: QueryType) -> Self {
+ Self { inner }
+ }
+}
+
+#[cfg(not(tarpaulin_include))]
+#[pymethods]
+impl HudiReadOptions {
+ /// Construct read options. Mirrors the Rust `ReadOptions` struct shape:
+ /// only the three stored fields are accepted directly. All other knobs
+ /// (`query_type`, timestamps, `batch_size`) are set via the chainable
+ /// `with_*` builders, matching the Rust API.
+ ///
+ /// `filters` are parsed and cardinality-validated here; an unrecognized
+ /// operator or empty `IN`/`NOT IN` value list raises immediately.
+ #[new]
+ #[pyo3(signature = (filters=None, projection=None, hudi_options=None))]
+ fn new(
+ filters: Option<Vec<(String, String, String)>>,
+ projection: Option<Vec<String>>,
+ hudi_options: Option<HashMap<String, String>>,
+ ) -> PyResult<Self> {
+ let mut inner = ReadOptions::new()
+ .with_filters(filters.unwrap_or_default())
+ .map_err(PythonError::from)?
+ .with_hudi_options(hudi_options.unwrap_or_default());
+ if let Some(projection) = projection {
+ inner = inner.with_projection(projection);
+ }
+ Ok(Self { inner })
+ }
+
+ fn __repr__(&self) -> String {
+ format!(
+ "HudiReadOptions(filters={:?}, projection={:?}, hudi_options={:?})",
+ self.filters(),
+ self.inner.projection,
+ self.inner.hudi_options,
+ )
+ }
+
+ #[getter]
+ fn filters(&self) -> Vec<(String, String, String)> {
+ self.inner
+ .filters
+ .iter()
+ .cloned()
+ .map(|f| f.into())
+ .collect()
+ }
+
+ #[getter]
+ fn projection(&self) -> Option<Vec<String>> {
+ self.inner.projection.clone()
+ }
+
+ #[getter]
+ fn hudi_options(&self) -> HashMap<String, String> {
+ self.inner.hudi_options.clone()
+ }
+
+ // ---- typed builders (return a new instance for chaining) ----
+
+ /// Sets the query type. Stored under `hoodie.read.query.type`.
+ fn with_query_type(&self, query_type: &HudiQueryType) -> Self {
+ Self {
+ inner: self.inner.clone().with_query_type(query_type.inner),
+ }
+ }
+
+ /// Sets the as-of timestamp for snapshot/time-travel queries.
+ fn with_as_of_timestamp(&self, timestamp: &str) -> Self {
+ Self {
+ inner: self.inner.clone().with_as_of_timestamp(timestamp),
+ }
+ }
+
+ /// Sets the lower-bound timestamp (exclusive) for incremental queries.
+ fn with_start_timestamp(&self, timestamp: &str) -> Self {
+ Self {
+ inner: self.inner.clone().with_start_timestamp(timestamp),
+ }
+ }
+
+ /// Sets the upper-bound timestamp (inclusive) for incremental queries.
+ fn with_end_timestamp(&self, timestamp: &str) -> Self {
+ Self {
+ inner: self.inner.clone().with_end_timestamp(timestamp),
+ }
+ }
+
+ /// Sets the target batch size (rows per batch) for streaming reads.
+ /// Raises if `size == 0` β the parquet stream reader yields no batches
+ /// for a zero-row target, so this is almost certainly a caller mistake.
+ fn with_batch_size(&self, size: usize) -> PyResult<Self> {
+ Ok(Self {
+ inner: self
+ .inner
+ .clone()
+ .with_batch_size(size)
+ .map_err(PythonError::from)?,
+ })
+ }
+
+ /// Sets column filters. Parses and cardinality-validates here; an
+ /// unrecognized operator or empty `IN`/`NOT IN` value list raises.
+ fn with_filters(&self, filters: Vec<(String, String, String)>) -> PyResult<Self> {
+ Ok(Self {
+ inner: self
+ .inner
+ .clone()
+ .with_filters(filters)
+ .map_err(PythonError::from)?,
+ })
+ }
+
+ /// Sets the column projection (which columns to read).
+ fn with_projection(&self, columns: Vec<String>) -> Self {
+ Self {
+ inner: self.inner.clone().with_projection(columns),
+ }
+ }
+
+ /// Sets a single Hudi config that applies to this read only.
+ fn with_hudi_option(&self, key: &str, value: &str) -> Self {
+ Self {
+ inner: self.inner.clone().with_hudi_option(key, value),
+ }
+ }
+
+ /// Sets a batch of Hudi configs that apply to this read only.
+ fn with_hudi_options(&self, opts: HashMap<String, String>) -> Self {
+ Self {
+ inner: self.inner.clone().with_hudi_options(opts),
+ }
+ }
+
+ // ---- typed accessors (read from hudi_options) ----
+
+ /// The query type (defaults to `Snapshot` when unset). Raises on bad strings.
+ fn query_type(&self) -> PyResult<HudiQueryType> {
+ self.inner
+ .query_type()
+ .map(HudiQueryType::from)
+ .map_err(PythonError::from)
+ .map_err(PyErr::from)
+ }
+
+ /// The as-of timestamp for snapshot/time-travel queries, if set.
+ fn as_of_timestamp(&self) -> Option<String> {
+ self.inner.as_of_timestamp().map(String::from)
+ }
+
+ /// The start timestamp (exclusive) for incremental queries, if set.
+ fn start_timestamp(&self) -> Option<String> {
+ self.inner.start_timestamp().map(String::from)
+ }
+
+ /// The end timestamp (inclusive) for incremental queries, if set.
+ fn end_timestamp(&self) -> Option<String> {
+ self.inner.end_timestamp().map(String::from)
+ }
+
+ /// The target batch size (rows per batch) for streaming reads, if set.
+ /// Raises if the stored value is not a valid integer or is `0` (a zero-row
+ /// batch yields no batches at the parquet stream reader).
+ fn batch_size(&self) -> PyResult<Option<usize>> {
+ self.inner
+ .batch_size()
+ .map_err(PythonError::from)
+ .map_err(PyErr::from)
+ }
+}
+
+impl HudiReadOptions {
+ fn to_inner(&self) -> ReadOptions {
+ self.inner.clone()
+ }
+}
+
+#[cfg(not(tarpaulin_include))]
+#[pyclass]
+pub struct HudiRecordBatchStream {
+ inner: Arc<Mutex<RecordBatchBoxStream>>,
+}
+
+#[cfg(not(tarpaulin_include))]
+#[pymethods]
+impl HudiRecordBatchStream {
+ fn __iter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> {
+ slf
+ }
+
+ fn __next__(slf: PyRef<'_, Self>, py: Python) -> PyResult<Option<Py<PyAny>>> {
+ let stream = slf.inner.clone();
+ let result = py.detach(|| {
+ rt().block_on(async move {
+ let mut stream = stream.lock().await;
+ stream.next().await
+ })
+ });
+
+ match result {
+ Some(Ok(batch)) => Ok(Some(batch.to_pyarrow(py)?.unbind())),
+ Some(Err(e)) => Err(PythonError::from(e).into()),
+ None => Ok(None),
+ }
+ }
+}
+
+impl HudiRecordBatchStream {
+ fn from_stream(stream: RecordBatchBoxStream) -> Self {
+ Self {
+ inner: Arc::new(Mutex::new(stream)),
+ }
+ }
+}
+
#[cfg(not(tarpaulin_include))]
#[derive(Clone, Debug)]
#[pyclass]
@@ -99,20 +371,14 @@ impl HudiFileGroupReader {
Ok(HudiFileGroupReader { inner })
}
- fn read_file_slice_by_base_file_path(
+ #[pyo3(signature = (file_slice, options=None))]
+ fn read_file_slice(
&self,
- relative_path: &str,
+ file_slice: &HudiFileSlice,
+ options: Option<HudiReadOptions>,
py: Python,
) -> PyResult<Py<PyAny>> {
- py.detach(|| {
- rt().block_on(self.inner.read_file_slice_by_base_file_path(relative_path))
- .map_err(PythonError::from)
- })?
- .to_pyarrow(py)
- .map(|b| b.unbind())
- }
-
- fn read_file_slice(&self, file_slice: &HudiFileSlice, py: Python) -> PyResult<Py<PyAny>> {
+ let read_options = options.unwrap_or_default().to_inner();
let mut file_group = FileGroup::new_with_base_file_name(
&file_slice.base_file_name,
&file_slice.partition_path,
@@ -133,29 +399,90 @@ impl HudiFileGroupReader {
})
.map_err(PythonError::from)?;
py.detach(|| {
- rt().block_on(self.inner.read_file_slice(file_slice))
+ rt().block_on(self.inner.read_file_slice(file_slice, &read_options))
.map_err(PythonError::from)
})?
.to_pyarrow(py)
.map(|b| b.unbind())
}
+ #[pyo3(signature = (base_file_path, log_file_paths, options=None))]
fn read_file_slice_from_paths(
&self,
base_file_path: &str,
log_file_paths: Vec<String>,
+ options: Option<HudiReadOptions>,
py: Python,
) -> PyResult<Py<PyAny>> {
+ let read_options = options.unwrap_or_default().to_inner();
py.detach(|| {
- rt().block_on(
- self.inner
- .read_file_slice_from_paths(base_file_path, log_file_paths),
- )
+ rt().block_on(self.inner.read_file_slice_from_paths(
+ base_file_path,
+ log_file_paths,
+ &read_options,
+ ))
.map_err(PythonError::from)
})?
.to_pyarrow(py)
.map(|b| b.unbind())
}
+
+ #[pyo3(signature = (file_slice, options=None))]
+ fn read_file_slice_stream(
+ &self,
+ file_slice: &HudiFileSlice,
+ options: Option<HudiReadOptions>,
+ py: Python,
+ ) -> PyResult<HudiRecordBatchStream> {
+ let read_options = options.unwrap_or_default().to_inner();
+ let mut file_group = FileGroup::new_with_base_file_name(
+ &file_slice.base_file_name,
+ &file_slice.partition_path,
+ )
+ .map_err(PythonError::from)?;
+ file_group
+ .add_log_files_from_names(&file_slice.log_file_names)
+ .map_err(PythonError::from)?;
+ let inner_reader = self.inner.clone();
+ let stream = py.detach(|| {
+ rt().block_on(async move {
+ let (_, fs) = file_group.file_slices.iter().next().ok_or_else(|| {
+ CoreError::FileGroup(format!(
+ "Failed to get file slice from file group: {file_group:?}"
+ ))
+ })?;
+ inner_reader.read_file_slice_stream(fs, &read_options).await
+ })
+ .map_err(PythonError::from)
+ })?;
+ Ok(HudiRecordBatchStream::from_stream(stream))
+ }
+
+ #[pyo3(signature = (base_file_path, log_file_paths, options=None))]
+ fn read_file_slice_from_paths_stream(
+ &self,
+ base_file_path: &str,
+ log_file_paths: Vec<String>,
+ options: Option<HudiReadOptions>,
+ py: Python,
+ ) -> PyResult<HudiRecordBatchStream> {
+ let read_options = options.unwrap_or_default().to_inner();
+ let stream = py.detach(|| {
+ rt().block_on(self.inner.read_file_slice_from_paths_stream(
+ base_file_path,
+ log_file_paths,
+ &read_options,
+ ))
+ .map_err(PythonError::from)
+ })?;
+ Ok(HudiRecordBatchStream::from_stream(stream))
+ }
+
+ /// Whether this reader targets a metadata table (its base path ends with `.hoodie/metadata`).
+ #[getter]
+ fn is_metadata_table(&self) -> bool {
+ self.inner.is_metadata_table()
+ }
}
#[cfg(not(tarpaulin_include))]
@@ -387,160 +714,72 @@ impl HudiTable {
})
}
- #[pyo3(signature = (num_splits, filters=None))]
- fn get_file_slices_splits(
- &self,
- num_splits: usize,
- filters: Option<Vec<(String, String, String)>>,
- py: Python,
- ) -> PyResult<Vec<Vec<HudiFileSlice>>> {
- py.detach(|| {
- let file_slices = rt()
- .block_on(
- self.inner
- .get_file_slices_splits(num_splits, filters.unwrap_or_default()),
- )
- .map_err(PythonError::from)?;
- Ok(file_slices
- .iter()
- .map(|inner_vec| inner_vec.iter().map(HudiFileSlice::from).collect())
- .collect())
- })
- }
-
- #[pyo3(signature = (num_splits, timestamp, filters=None))]
- fn get_file_slices_splits_as_of(
- &self,
- num_splits: usize,
- timestamp: &str,
- filters: Option<Vec<(String, String, String)>>,
- py: Python,
- ) -> PyResult<Vec<Vec<HudiFileSlice>>> {
- py.detach(|| {
- let file_slices = rt()
- .block_on(self.inner.get_file_slices_splits_as_of(
- num_splits,
- timestamp,
- filters.unwrap_or_default(),
- ))
- .map_err(PythonError::from)?;
- Ok(file_slices
- .iter()
- .map(|inner_vec| inner_vec.iter().map(HudiFileSlice::from).collect())
- .collect())
- })
- }
-
- #[pyo3(signature = (filters=None))]
+ #[pyo3(signature = (options=None))]
fn get_file_slices(
&self,
- filters: Option<Vec<(String, String, String)>>,
- py: Python,
- ) -> PyResult<Vec<HudiFileSlice>> {
- py.detach(|| {
- let file_slices = rt()
- .block_on(self.inner.get_file_slices(filters.unwrap_or_default()))
- .map_err(PythonError::from)?;
- Ok(file_slices.iter().map(HudiFileSlice::from).collect())
- })
- }
-
- #[pyo3(signature = (timestamp, filters=None))]
- fn get_file_slices_as_of(
- &self,
- timestamp: &str,
- filters: Option<Vec<(String, String, String)>>,
- py: Python,
- ) -> PyResult<Vec<HudiFileSlice>> {
- py.detach(|| {
- let file_slices = rt()
- .block_on(
- self.inner
- .get_file_slices_as_of(timestamp, filters.unwrap_or_default()),
- )
- .map_err(PythonError::from)?;
- Ok(file_slices.iter().map(HudiFileSlice::from).collect())
- })
- }
-
- #[pyo3(signature = (start_timestamp=None, end_timestamp=None))]
- fn get_file_slices_between(
- &self,
- start_timestamp: Option<&str>,
- end_timestamp: Option<&str>,
+ options: Option<HudiReadOptions>,
py: Python,
) -> PyResult<Vec<HudiFileSlice>> {
+ let read_options = options.unwrap_or_default().to_inner();
py.detach(|| {
let file_slices = rt()
- .block_on(
- self.inner
- .get_file_slices_between(start_timestamp, end_timestamp),
- )
+ .block_on(self.inner.get_file_slices(&read_options))
.map_err(PythonError::from)?;
Ok(file_slices.iter().map(HudiFileSlice::from).collect())
})
}
- #[pyo3(signature = (options=None))]
+ #[pyo3(signature = (read_options=None, extra_hudi_overrides=None, extra_storage_overrides=None))]
fn create_file_group_reader_with_options(
&self,
- options: Option<HashMap<String, String>>,
+ read_options: Option<HudiReadOptions>,
+ extra_hudi_overrides: Option<HashMap<String, String>>,
+ extra_storage_overrides: Option<HashMap<String, String>>,
) -> PyResult<HudiFileGroupReader> {
+ let read_options = read_options.map(|o| o.to_inner());
let fg_reader = self
.inner
- .create_file_group_reader_with_options(options.unwrap_or_default())
+ .create_file_group_reader_with_options(
+ read_options.as_ref(),
+ extra_hudi_overrides.unwrap_or_default(),
+ extra_storage_overrides.unwrap_or_default(),
+ )
.map_err(PythonError::from)?;
Ok(HudiFileGroupReader { inner: fg_reader })
}
- #[pyo3(signature = (filters=None))]
- fn read_snapshot(
- &self,
- filters: Option<Vec<(String, String, String)>>,
- py: Python,
- ) -> PyResult<Py<PyAny>> {
+ #[pyo3(signature = (options=None))]
+ fn read(&self, options: Option<HudiReadOptions>, py: Python) -> PyResult<Py<PyAny>> {
+ let read_options = options.unwrap_or_default().to_inner();
py.detach(|| {
- rt().block_on(self.inner.read_snapshot(filters.unwrap_or_default()))
+ rt().block_on(self.inner.read(&read_options))
.map_err(PythonError::from)
})?
.to_pyarrow(py)
.map(|b| b.unbind())
}
- #[pyo3(signature = (timestamp, filters=None))]
- fn read_snapshot_as_of(
- &self,
- timestamp: &str,
- filters: Option<Vec<(String, String, String)>>,
- py: Python,
- ) -> PyResult<Py<PyAny>> {
- py.detach(|| {
- rt().block_on(
- self.inner
- .read_snapshot_as_of(timestamp, filters.unwrap_or_default()),
- )
- .map_err(PythonError::from)
- })?
- .to_pyarrow(py)
- .map(|b| b.unbind())
+ #[getter]
+ fn base_url(&self) -> String {
+ self.inner.base_url().to_string()
+ }
+
+ fn compute_table_stats(&self, py: Python) -> Option<(u64, u64)> {
+ py.detach(|| rt().block_on(self.inner.compute_table_stats()))
}
- #[pyo3(signature = (start_timestamp, end_timestamp=None))]
- fn read_incremental_records(
+ #[pyo3(signature = (options=None))]
+ fn read_stream(
&self,
- start_timestamp: &str,
- end_timestamp: Option<&str>,
+ options: Option<HudiReadOptions>,
py: Python,
- ) -> PyResult<Py<PyAny>> {
- py.detach(|| {
- rt().block_on(
- self.inner
- .read_incremental_records(start_timestamp, end_timestamp),
- )
- .map_err(PythonError::from)
- })?
- .to_pyarrow(py)
- .map(|b| b.unbind())
+ ) -> PyResult<HudiRecordBatchStream> {
+ let read_options = options.unwrap_or_default().to_inner();
+ let stream = py.detach(|| {
+ rt().block_on(self.inner.read_stream(&read_options))
+ .map_err(PythonError::from)
+ })?;
+ Ok(HudiRecordBatchStream::from_stream(stream))
}
}
diff --git a/python/src/lib.rs b/python/src/lib.rs
index 5111b385..6308ff37 100644
--- a/python/src/lib.rs
+++ b/python/src/lib.rs
@@ -31,10 +31,16 @@ mod testing_internal;
fn _internal(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add("__version__", env!("CARGO_PKG_VERSION"))?;
- use internal::{HudiFileGroupReader, HudiFileSlice, HudiInstant, HudiTable, HudiTimeline};
+ use internal::{
+ HudiFileGroupReader, HudiFileSlice, HudiInstant, HudiQueryType, HudiReadOptions,
+ HudiRecordBatchStream, HudiTable, HudiTimeline,
+ };
m.add_class::<HudiFileGroupReader>()?;
m.add_class::<HudiFileSlice>()?;
m.add_class::<HudiInstant>()?;
+ m.add_class::<HudiQueryType>()?;
+ m.add_class::<HudiReadOptions>()?;
+ m.add_class::<HudiRecordBatchStream>()?;
m.add_class::<HudiTable>()?;
m.add_class::<HudiTimeline>()?;
diff --git a/python/tests/test_config_enums.py b/python/tests/test_config_enums.py
index 69bcce66..c566adc3 100644
--- a/python/tests/test_config_enums.py
+++ b/python/tests/test_config_enums.py
@@ -37,10 +37,7 @@ def test_internal_config_keys_returns_both_enums():
assert table_dict["TABLE_NAME"] == "hoodie.table.name"
assert table_dict["BASE_FILE_FORMAT"] == "hoodie.table.base.file.format"
assert read_dict["INPUT_PARTITIONS"] == "hoodie.read.input.partitions"
- assert (
- read_dict["FILE_GROUP_START_TIMESTAMP"]
- == "hoodie.read.file_group.start_timestamp"
- )
+ assert read_dict["START_TIMESTAMP"] == "hoodie.read.start.timestamp"
def test_config_enums_are_real_python_enums():
diff --git a/python/tests/test_file_group_read.py b/python/tests/test_file_group_read.py
index 145af8c1..75534145 100644
--- a/python/tests/test_file_group_read.py
+++ b/python/tests/test_file_group_read.py
@@ -35,9 +35,7 @@ def test_file_group_api_read_file_slice(v8_trips_table):
f for f in file_slices if "san_francisco" in f.base_file_relative_path()
][0]
- batch = file_group_reader.read_file_slice_by_base_file_path(
- sf_slice.base_file_relative_path()
- )
+ batch = file_group_reader.read_file_slice(sf_slice)
t = pa.Table.from_batches([batch]).select(["rider", "fare"]).sort_by("rider")
rows = t.to_pylist()
@@ -65,7 +63,3 @@ def test_file_group_api_read_file_slice_from_paths(v8_trips_table):
assert batch.num_rows == 4
assert batch.num_columns > 0
-
- batch_original = file_group_reader.read_file_slice_by_base_file_path(sf_path)
- assert batch.num_rows == batch_original.num_rows
- assert batch.num_columns == batch_original.num_columns
diff --git a/python/tests/test_streaming_read.py b/python/tests/test_streaming_read.py
new file mode 100644
index 00000000..cdce16ba
--- /dev/null
+++ b/python/tests/test_streaming_read.py
@@ -0,0 +1,167 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Streaming read tests using v8_trips_8i3u1d (MOR, partitioned by city)."""
+
+import pyarrow as pa
+import pytest
+
+from hudi import (
+ HudiFileGroupReader,
+ HudiQueryType,
+ HudiReadOptions,
+ HudiRecordBatchStream,
+ HudiTable,
+)
+
+
+def test_read_options_default():
+ options = HudiReadOptions()
+ assert options.filters == []
+ assert options.projection is None
+ assert options.hudi_options == {}
+
+
+def test_read_options_typed_accessors_round_trip():
+ """Every builder that writes into hudi_options must round-trip via its typed accessor."""
+ options = (
+ HudiReadOptions(filters=[("city", "=", "san_francisco")])
+ .with_projection(["uuid", "rider"])
+ .with_query_type(HudiQueryType.Incremental)
+ .with_batch_size(2048)
+ .with_as_of_timestamp("20240101000000000")
+ .with_start_timestamp("20240101000000000")
+ .with_end_timestamp("20240301000000000")
+ )
+ assert options.filters == [("city", "=", "san_francisco")]
+ assert options.projection == ["uuid", "rider"]
+ # Builders insert into hudi_options under the matching HudiReadConfig keys.
+ assert options.hudi_options["hoodie.read.stream.batch_size"] == "2048"
+ assert options.hudi_options["hoodie.read.as.of.timestamp"] == "20240101000000000"
+ assert options.hudi_options["hoodie.read.start.timestamp"] == "20240101000000000"
+ assert options.hudi_options["hoodie.read.end.timestamp"] == "20240301000000000"
+ # Typed accessors round-trip.
+ assert options.query_type() == HudiQueryType.Incremental
+ assert options.batch_size() == 2048
+ assert options.as_of_timestamp() == "20240101000000000"
+ assert options.start_timestamp() == "20240101000000000"
+ assert options.end_timestamp() == "20240301000000000"
+ assert "HudiReadOptions" in repr(options)
+
+
+def test_read_options_query_type_bad_value_raises():
+ """A bogus value left in hudi_options under the query-type key surfaces as an error."""
+ bogus = HudiReadOptions(hudi_options={"hoodie.read.query.type": "bogus"})
+ with pytest.raises(Exception):
+ bogus.query_type()
+
+
+def test_read_options_batch_size_bad_value_raises():
+ """Bad batch_size values: non-integers stuck in the bag surface at the accessor,
+ and `with_batch_size(0)` is rejected at the builder so callers catch the misuse
+ synchronously rather than at read time."""
+ bogus = HudiReadOptions(
+ hudi_options={"hoodie.read.stream.batch_size": "not_a_number"}
+ )
+ with pytest.raises(Exception):
+ bogus.batch_size()
+
+ with pytest.raises(Exception, match="must be > 0"):
+ HudiReadOptions().with_batch_size(0)
+
+
+def test_read_options_with_filters_validates_at_build_time():
+ """`with_filters` parses + cardinality-validates upfront, so a bad operator or
+ empty IN value list raises here rather than at read time."""
+ with pytest.raises(Exception):
+ HudiReadOptions().with_filters([("col", "BAD_OP", "x")])
+
+ with pytest.raises(Exception, match="at least one value"):
+ HudiReadOptions().with_filters([("col", "IN", "")])
+
+
+def test_read_options_bulk_setters():
+ """The plural builders (with_filters, with_hudi_options) chain identically to the singular ones."""
+ options = (
+ HudiReadOptions()
+ .with_filters([("city", "=", "x"), ("rider", "=", "rider-A")])
+ .with_hudi_options({"k1": "v1", "k2": "v2"})
+ )
+ assert options.filters == [("city", "=", "x"), ("rider", "=", "rider-A")]
+ assert options.hudi_options["k1"] == "v1"
+ assert options.hudi_options["k2"] == "v2"
+
+
+def test_read_snapshot_stream_yields_all_rows(v8_trips_table):
+ table = HudiTable(v8_trips_table)
+ stream = table.read_stream()
+ assert isinstance(stream, HudiRecordBatchStream)
+
+ batches = list(stream)
+ assert len(batches) > 0
+ t = pa.Table.from_batches(batches)
+ assert t.num_rows == 6
+
+
+def test_read_snapshot_stream_with_partition_filter(v8_trips_table):
+ table = HudiTable(v8_trips_table)
+ options = HudiReadOptions(filters=[("city", "=", "san_francisco")])
+ batches = list(table.read_stream(options))
+ t = pa.Table.from_batches(batches)
+ assert t.num_rows == 4
+
+
+def test_read_snapshot_stream_is_single_use(v8_trips_table):
+ table = HudiTable(v8_trips_table)
+ stream = table.read_stream()
+ first = list(stream)
+ assert len(first) > 0
+ # Re-iterating an exhausted stream yields nothing
+ assert list(stream) == []
+
+
+def test_read_file_slice_stream_file_group_reader(v8_trips_table):
+ table = HudiTable(v8_trips_table)
+ file_slices = table.get_file_slices(
+ HudiReadOptions(filters=[("city", "=", "san_francisco")])
+ )
+ reader = HudiFileGroupReader(v8_trips_table)
+
+ batches = list(reader.read_file_slice_stream(file_slices[0]))
+ t = pa.Table.from_batches(batches)
+ assert t.num_rows == 4
+
+
+def test_read_file_slice_from_paths_stream(v8_trips_table):
+ table = HudiTable(v8_trips_table)
+ file_slices = table.get_file_slices(
+ HudiReadOptions(filters=[("city", "=", "san_francisco")])
+ )
+ reader = HudiFileGroupReader(v8_trips_table)
+ base_path = file_slices[0].base_file_relative_path()
+
+ batches = list(reader.read_file_slice_from_paths_stream(base_path, []))
+ t = pa.Table.from_batches(batches)
+ assert t.num_rows == 4
+
+
+def test_read_snapshot_stream_with_batch_size(v8_trips_table):
+ table = HudiTable(v8_trips_table)
+ options = HudiReadOptions().with_batch_size(1)
+ batches = list(table.read_stream(options))
+ t = pa.Table.from_batches(batches)
+ assert t.num_rows == 6
diff --git a/python/tests/test_table_builder.py b/python/tests/test_table_builder.py
index d8b83377..a9629272 100644
--- a/python/tests/test_table_builder.py
+++ b/python/tests/test_table_builder.py
@@ -73,7 +73,7 @@ def test_with_multiple_options(builder, method, attr):
def test_read_table_returns_correct_data(v8_trips_table):
table = HudiTableBuilder.from_base_uri(v8_trips_table).build()
- batches = table.read_snapshot()
+ batches = table.read()
t = pa.Table.from_batches(batches).select(["rider", "fare"]).sort_by("rider")
rows = t.to_pylist()
@@ -94,14 +94,14 @@ def test_read_table_returns_correct_data(v8_trips_table):
"hudi_options,storage_options,options",
[
(
- {"hoodie.read.file_group.start_timestamp": "resolved value"},
- {"hoodie.read.file_group.start_timestamp": "not taking"},
- {"hoodie.read.file_group.start_timestamp": "lower precedence"},
+ {"hoodie.read.start.timestamp": "resolved value"},
+ {"hoodie.read.start.timestamp": "not taking"},
+ {"hoodie.read.start.timestamp": "lower precedence"},
),
(
{},
- {"hoodie.read.file_group.start_timestamp": "not taking"},
- {"hoodie.read.file_group.start_timestamp": "resolved value"},
+ {"hoodie.read.start.timestamp": "not taking"},
+ {"hoodie.read.start.timestamp": "resolved value"},
),
],
)
@@ -114,10 +114,7 @@ def test_setting_table_options(v8_trips_table, hudi_options, storage_options, op
.build()
)
- assert (
- table.hudi_options().get("hoodie.read.file_group.start_timestamp")
- == "resolved value"
- )
+ assert table.hudi_options().get("hoodie.read.start.timestamp") == "resolved value"
def test_with_hudi_option_enum(builder):
@@ -150,14 +147,8 @@ def test_enum_values_match_expected_strings():
HudiReadConfig.USE_READ_OPTIMIZED_MODE.value
== "hoodie.read.use.read_optimized.mode"
)
- assert (
- HudiReadConfig.FILE_GROUP_START_TIMESTAMP.value
- == "hoodie.read.file_group.start_timestamp"
- )
- assert (
- HudiReadConfig.FILE_GROUP_END_TIMESTAMP.value
- == "hoodie.read.file_group.end_timestamp"
- )
+ assert HudiReadConfig.START_TIMESTAMP.value == "hoodie.read.start.timestamp"
+ assert HudiReadConfig.END_TIMESTAMP.value == "hoodie.read.end.timestamp"
def test_mixed_string_and_enum_usage(builder):
diff --git a/python/tests/test_table_incremental_read.py b/python/tests/test_table_incremental_read.py
index 2c716c7e..8e51553a 100644
--- a/python/tests/test_table_incremental_read.py
+++ b/python/tests/test_table_incremental_read.py
@@ -24,24 +24,79 @@ return the updated TXN-001 row.
import pyarrow as pa
-from hudi import HudiTable
+from hudi import HudiQueryType, HudiReadOptions, HudiTable
from hudi._internal import get_test_table_path
-def test_table_incremental_read_returns_correct_data():
- table_path = get_test_table_path("v9_txns_simple_nometa", "cow")
- table = HudiTable(table_path)
+def _commits(table: HudiTable):
+ return table.get_timeline().get_completed_commits()
- timeline = table.get_timeline()
- commits = timeline.get_completed_commits()
- assert len(commits) >= 2
- insert_ts = commits[0].timestamp
- update_ts = commits[1].timestamp
+def test_table_incremental_read_returns_correct_data():
+ table = HudiTable(get_test_table_path("v9_txns_simple_nometa", "cow"))
+ commits = _commits(table)
+ assert len(commits) >= 2
- batches = table.read_incremental_records(insert_ts, update_ts)
+ options = (
+ HudiReadOptions()
+ .with_query_type(HudiQueryType.Incremental)
+ .with_start_timestamp(commits[0].timestamp)
+ .with_end_timestamp(commits[1].timestamp)
+ )
+ batches = table.read(options)
t = pa.Table.from_batches(batches)
assert t.num_rows == 1
assert t.column("txn_id").to_pylist() == ["TXN-001"]
assert t.column("txn_type").to_pylist() == ["reversal"]
+
+
+def test_incremental_read_with_partition_filter():
+ table = HudiTable(get_test_table_path("v9_txns_simple_nometa", "cow"))
+ commits = _commits(table)
+ insert_ts, update_ts = commits[0].timestamp, commits[1].timestamp
+
+ # TXN-001 lives in region 'us'; filtering to 'us' keeps it.
+ options_us = (
+ HudiReadOptions(filters=[("region", "=", "us")])
+ .with_query_type(HudiQueryType.Incremental)
+ .with_start_timestamp(insert_ts)
+ .with_end_timestamp(update_ts)
+ )
+ batches = table.read(options_us)
+ t = pa.Table.from_batches(batches)
+ assert t.num_rows == 1
+ assert t.column("txn_id").to_pylist() == ["TXN-001"]
+
+ # Filtering to a non-matching region prunes the change away.
+ options_eu = (
+ HudiReadOptions(filters=[("region", "=", "eu")])
+ .with_query_type(HudiQueryType.Incremental)
+ .with_start_timestamp(insert_ts)
+ .with_end_timestamp(update_ts)
+ )
+ batches = table.read(options_eu)
+ assert sum(b.num_rows for b in batches) == 0
+
+
+def test_get_incremental_file_slices_with_partition_filter():
+ table = HudiTable(get_test_table_path("v9_txns_simple_nometa", "cow"))
+ commits = _commits(table)
+ insert_ts, update_ts = commits[0].timestamp, commits[1].timestamp
+
+ slices_us = table.get_file_slices(
+ HudiReadOptions(filters=[("region", "=", "us")])
+ .with_query_type(HudiQueryType.Incremental)
+ .with_start_timestamp(insert_ts)
+ .with_end_timestamp(update_ts)
+ )
+ assert len(slices_us) == 1
+ assert "region=us" in slices_us[0].base_file_relative_path()
+
+ slices_eu = table.get_file_slices(
+ HudiReadOptions(filters=[("region", "=", "eu")])
+ .with_query_type(HudiQueryType.Incremental)
+ .with_start_timestamp(insert_ts)
+ .with_end_timestamp(update_ts)
+ )
+ assert len(slices_eu) == 0
diff --git a/python/tests/test_table_misc.py b/python/tests/test_table_misc.py
new file mode 100644
index 00000000..2108df63
--- /dev/null
+++ b/python/tests/test_table_misc.py
@@ -0,0 +1,124 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Tests for HudiTable accessor and stats APIs."""
+
+from hudi import HudiFileGroupReader, HudiQueryType, HudiReadOptions, HudiTable
+from hudi._internal import get_test_table_path
+
+
+def test_table_metadata_getters(v8_trips_table):
+ """Table-level getters round-trip the values from hoodie.properties."""
+ table = HudiTable(v8_trips_table)
+ base_url = table.base_url
+ assert base_url.startswith("file://")
+ assert base_url.rstrip("/").endswith("v8_trips_8i3u1d")
+ assert table.table_name == "v8_trips_8i3u1d"
+ assert table.table_type == "MERGE_ON_READ"
+ assert table.is_mor is True
+ assert isinstance(table.timezone, str) and len(table.timezone) > 0
+ assert isinstance(table.storage_options(), dict)
+ assert isinstance(table.hudi_options(), dict)
+
+
+def test_file_slice_log_files_relative_paths(v8_trips_table):
+ """The log-files accessor must mirror log_file_names with the partition prefix attached."""
+ table = HudiTable(v8_trips_table)
+ # Pick any slice with log files (V8Trips8I3U1D has updates β log files exist).
+ slice_with_logs = next(
+ (s for s in table.get_file_slices() if s.log_file_names),
+ None,
+ )
+ assert slice_with_logs is not None, (
+ "MOR fixture should have at least one slice with log files"
+ )
+ relative_paths = slice_with_logs.log_files_relative_paths()
+ assert len(relative_paths) == len(slice_with_logs.log_file_names)
+ for path in relative_paths:
+ assert path.startswith(slice_with_logs.partition_path)
+
+
+def test_file_group_reader_with_options_constructor(v8_trips_table):
+ """HudiFileGroupReader must accept an options dict (PyO3-bound constructor)."""
+ reader = HudiFileGroupReader(
+ v8_trips_table,
+ {"hoodie.read.use.read_optimized.mode": "true"},
+ )
+ table = HudiTable(v8_trips_table)
+ sf_slices = table.get_file_slices(
+ HudiReadOptions(filters=[("city", "=", "san_francisco")])
+ )
+ batch = reader.read_file_slice(sf_slices[0])
+ # Read-optimized skips log merging; the result is still a valid batch with the table schema.
+ assert batch.num_columns > 0
+
+
+def test_get_incremental_file_slices():
+ table_path = get_test_table_path("v9_txns_simple_nometa", "cow")
+ table = HudiTable(table_path)
+ commits = table.get_timeline().get_completed_commits()
+ assert len(commits) >= 2
+
+ slices = table.get_file_slices(
+ HudiReadOptions()
+ .with_query_type(HudiQueryType.Incremental)
+ .with_start_timestamp(commits[0].timestamp)
+ )
+ assert len(slices) >= 1
+
+
+def test_compute_table_stats(v8_trips_table):
+ table = HudiTable(v8_trips_table)
+ stats = table.compute_table_stats()
+ assert stats is not None, "v8_trips fixture is MDT-enabled; stats must not be None"
+ num_rows, byte_size = stats
+ assert num_rows > 0
+ assert byte_size > 0
+
+
+def test_read_dispatches_on_query_type(v8_trips_table):
+ table = HudiTable(v8_trips_table)
+
+ # Default (Snapshot) returns the latest 6 rows.
+ snapshot_rows = sum(b.num_rows for b in table.read(HudiReadOptions()))
+ assert snapshot_rows == 6
+
+ # Calling read() with no options uses the default Snapshot behavior.
+ snapshot_via_default = sum(b.num_rows for b in table.read())
+ assert snapshot_rows == snapshot_via_default
+
+
+def test_read_stream_errors_on_incremental(v8_trips_table):
+ import pytest
+
+ table = HudiTable(v8_trips_table)
+ with pytest.raises(Exception, match="not yet supported"):
+ table.read_stream(HudiReadOptions().with_query_type(HudiQueryType.Incremental))
+
+
+def test_read_options_hudi_options_plumbed_to_reader(v8_trips_table):
+ """Per-read hudi_options should affect file-group reader behavior."""
+ table = HudiTable(v8_trips_table)
+ options = HudiReadOptions().with_hudi_option(
+ "hoodie.read.use.read_optimized.mode", "true"
+ )
+ # Read-optimized snapshot just skips log merging β should still produce
+ # valid batches; on this MOR table with deltacommits the row count differs
+ # from the merged snapshot, but we only assert that the option flowed
+ # through without error.
+ batches = table.read(options)
+ assert isinstance(batches, list)
diff --git a/python/tests/test_table_read.py b/python/tests/test_table_read.py
index 6702b6f0..3466a71d 100644
--- a/python/tests/test_table_read.py
+++ b/python/tests/test_table_read.py
@@ -22,11 +22,9 @@ UPDATE rider-J fare=0, DELETE rider-J, UPDATE rider-G fare=0.
Final state: 6 rows (riders A, C, D, E, G, I).
"""
-from itertools import chain
-
import pyarrow as pa
-from hudi import HudiTable
+from hudi import HudiReadOptions, HudiTable
def test_read_table_has_correct_schema(v8_trips_table):
@@ -73,24 +71,21 @@ def test_read_table_can_read_from_batches(v8_trips_table):
file_slices = table.get_file_slices()
file_slice_paths = [f.base_file_relative_path() for f in file_slices]
- batch = (
- table.create_file_group_reader_with_options().read_file_slice_by_base_file_path(
- file_slice_paths[0]
- )
+ batch = table.create_file_group_reader_with_options().read_file_slice_from_paths(
+ file_slice_paths[0], []
)
t = pa.Table.from_batches([batch])
assert t.num_rows > 0
assert t.num_columns == 11
- file_slices_gen = iter(table.get_file_slices_splits(2))
- assert len(next(file_slices_gen)) == 2
- assert len(next(file_slices_gen)) == 1
+ all_slices = table.get_file_slices()
+ assert len(all_slices) == 3
def test_read_table_returns_correct_data(v8_trips_table):
table = HudiTable(v8_trips_table)
- batches = table.read_snapshot()
+ batches = table.read()
t = (
pa.Table.from_batches(batches)
.select(["ts", "uuid", "rider", "fare"])
@@ -123,7 +118,7 @@ def test_read_table_returns_correct_data(v8_trips_table):
def test_read_table_for_partition(v8_trips_table):
table = HudiTable(v8_trips_table)
- batches = table.read_snapshot([("city", "=", "san_francisco")])
+ batches = table.read(HudiReadOptions(filters=[("city", "=", "san_francisco")]))
t = (
pa.Table.from_batches(batches)
.select(["ts", "uuid", "rider", "fare"])
@@ -148,8 +143,8 @@ def test_table_apis_as_of_timestamp(v8_trips_table):
all_commits = timeline.get_completed_commits()
first_commit = all_commits[0].timestamp
- file_slices_gen = table.get_file_slices_splits_as_of(2, first_commit)
- all_slices = list(chain.from_iterable(file_slices_gen))
+ options_at_first = HudiReadOptions().with_as_of_timestamp(first_commit)
+ all_slices = table.get_file_slices(options_at_first)
assert len(all_slices) == 3
partition_prefixes = sorted(
f.base_file_relative_path().split("/")[0] for f in all_slices
@@ -162,7 +157,7 @@ def test_table_apis_as_of_timestamp(v8_trips_table):
# get_completed_commits() returns compaction commits for MOR tables.
# The sole compaction commit predates the final deltacommit (rider-G update).
- batches = table.read_snapshot_as_of(first_commit)
+ batches = table.read(options_at_first)
t = (
pa.Table.from_batches(batches)
.select(["ts", "uuid", "rider", "fare"])
@@ -192,7 +187,30 @@ def test_convert_filters_valid(v8_trips_table):
expected = [1, 1, 1, 2, 2]
for f, exp in zip(filters, expected):
- file_slices = table.get_file_slices(filters=[f])
+ file_slices = table.get_file_slices(HudiReadOptions(filters=[f]))
assert len(file_slices) == exp, (
f"Filter {f} expected {exp} slices, got {len(file_slices)}"
)
+
+
+def test_read_snapshot_filters_apply_as_row_predicate(v8_trips_table):
+ """Filters on a non-partition column drop rows during reading.
+
+ 'rider' is a non-partition column. Filtering on it does not prune any files
+ (since stats may not exclude the file), but rows must still be filtered.
+ """
+ table = HudiTable(v8_trips_table)
+ options = HudiReadOptions(filters=[("rider", "=", "rider-A")])
+ batches = table.read(options)
+ t = pa.Table.from_batches(batches)
+ assert t.num_rows == 1
+ assert t.column("rider").to_pylist() == ["rider-A"]
+
+
+def test_read_snapshot_stream_filters_apply_as_row_predicate(v8_trips_table):
+ table = HudiTable(v8_trips_table)
+ options = HudiReadOptions(filters=[("rider", "=", "rider-A")])
+ batches = list(table.read_stream(options))
+ t = pa.Table.from_batches(batches)
+ assert t.num_rows == 1
+ assert t.column("rider").to_pylist() == ["rider-A"]