You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "xushiyan (via GitHub)" <gi...@apache.org> on 2026/01/04 00:52:53 UTC
[PR] feat: support streaming read in table and file group APIs [hudi-rs]
xushiyan opened a new pull request, #507:
URL: https://github.com/apache/hudi-rs/pull/507
## Description
<!--- Describe your changes in detail -->
<!--- If it fixes an open issue, please link to the issue here. -->
<!--- Please link any related issues and PRs as well. -->
## How are the changes test-covered
- [ ] N/A
- [ ] Automated tests (unit and/or integration tests)
- [ ] Manual tests
- [ ] Details are described below
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support streaming read in table and file group APIs [hudi-rs]
Posted by "xushiyan (via GitHub)" <gi...@apache.org>.
xushiyan closed pull request #507: feat: support streaming read in table and file group APIs
URL: https://github.com/apache/hudi-rs/pull/507
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support streaming read in table and file group APIs [hudi-rs]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #507:
URL: https://github.com/apache/hudi-rs/pull/507#discussion_r2659241275
##########
python/src/internal.rs:
##########
@@ -348,154 +355,106 @@ 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))]
+ /// Get file slices with optional partition filters and time-travel timestamp.
+ ///
+ /// Args:
+ /// filters: Optional partition filter tuples in (field, op, value) format
+ /// timestamp: Optional timestamp for time-travel queries
+ #[pyo3(signature = (filters=None, timestamp=None))]
fn get_file_slices(
&self,
filters: Option<Vec<(String, String, String)>>,
+ timestamp: Option<&str>,
py: Python,
) -> PyResult<Vec<HudiFileSlice>> {
py.detach(|| {
+ let options = build_read_options(filters, timestamp).map_err(PythonError::from)?;
let file_slices = rt()
- .block_on(self.inner.get_file_slices(filters.unwrap_or_default()))
+ .block_on(self.inner.get_file_slices(options))
.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(
+ /// Get file slices for incremental reads between timestamps.
+ ///
+ /// Args:
+ /// start_timestamp: Start timestamp (exclusive)
+ /// end_timestamp: Optional end timestamp (inclusive), defaults to latest
+ #[pyo3(signature = (start_timestamp, end_timestamp=None))]
+ fn get_file_slices_incremental(
&self,
- start_timestamp: Option<&str>,
+ start_timestamp: &str,
end_timestamp: Option<&str>,
py: Python,
) -> PyResult<Vec<HudiFileSlice>> {
py.detach(|| {
+ let mut options = ReadOptions::new().from_timestamp(start_timestamp);
+ if let Some(end_ts) = end_timestamp {
+ options = options.to_timestamp(end_ts);
+ }
let file_slices = rt()
- .block_on(
- self.inner
- .get_file_slices_between(start_timestamp, end_timestamp),
- )
+ .block_on(self.inner.get_file_slices_incremental(options))
.map_err(PythonError::from)?;
Ok(file_slices.iter().map(HudiFileSlice::from).collect())
})
}
- #[pyo3(signature = (options=None))]
- fn create_file_group_reader_with_options(
- &self,
- options: Option<HashMap<String, String>>,
- ) -> PyResult<HudiFileGroupReader> {
+ /// Create a file group reader for this table.
+ fn create_file_group_reader(&self) -> PyResult<HudiFileGroupReader> {
+ let options = ReadOptions::new();
let fg_reader = self
.inner
- .create_file_group_reader_with_options(options.unwrap_or_default())
+ .create_file_group_reader(&options)
.map_err(PythonError::from)?;
Ok(HudiFileGroupReader { inner: fg_reader })
Review Comment:
The Python bindings create ReadOptions with `ReadOptions::new()` on lines 99 and 405, but ignore all the options that could be passed from Python (like batch_size, projection, row predicates). This means Python users cannot control these aspects of file slice reading.
Consider either:
1. Adding parameters to expose these options to Python
2. Documenting that these options are not yet supported in the Python API
##########
crates/core/src/table/mod.rs:
##########
@@ -619,8 +391,8 @@ 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.
+ pub(crate) fn create_file_group_reader_with_options<I, K, V>(
Review Comment:
The visibility of `create_file_group_reader_with_options` has been changed from `pub` to `pub(crate)` on line 395. This is a breaking API change that could affect external crates that depend on this function. If this is intentional for the new design where users should use `create_file_group_reader` instead, consider documenting this breaking change or providing a deprecation period.
```suggestion
pub fn create_file_group_reader_with_options<I, K, V>(
```
##########
crates/core/src/file_group/reader.rs:
##########
@@ -275,10 +282,18 @@ impl FileGroupReader {
.into();
let base_file_only = log_file_paths.is_empty() || use_read_optimized;
+ let projection = options.projection.as_deref();
+ let row_predicate = options.row_predicate.as_ref();
+
+ // For now, only projection is supported for base file only reads
+ // Log file merging with projection/predicate is more complex and deferred
Review Comment:
The comment on line 288 states "For now, only projection is supported for base file only reads" but the code immediately below (lines 291-292) appears to pass both projection and row_predicate to `read_parquet_file_with_opts`. This suggests either:
1. The comment is outdated and row predicates are actually supported
2. The implementation is incomplete and row predicates are incorrectly passed but not properly handled
Review and update either the comment or the implementation to ensure they match.
```suggestion
// For base file only reads, apply projection and row predicates via read_parquet_file_with_opts.
// For MOR tables with log files, projection and predicates are applied after merging log records.
```
##########
cpp/src/lib.rs:
##########
@@ -87,74 +84,41 @@ pub fn new_file_group_reader_with_options(
}
impl HudiFileGroupReader {
- pub fn read_file_slice_by_base_file_path(
+ pub fn read_file_slice_by_paths(
&self,
- relative_path: &CxxString,
+ base_file_path: &CxxString,
+ log_file_paths: &CxxVector<CxxString>,
) -> *mut ffi::ArrowArrayStream {
- let relative_path = relative_path
+ let base_file_path = base_file_path
.to_str()
.expect("Failed to convert CxxString to str: Invalid UTF-8 sequence");
- let record_batch = self
- .inner
- .read_file_slice_by_base_file_path_blocking(relative_path)
- .expect("Failed to read file batch");
- let schema = record_batch.schema();
-
- create_raw_pointer_for_record_batches(vec![record_batch], schema)
- }
-
- pub fn read_file_slice(&self, file_slice: &HudiFileSlice) -> *mut ffi::ArrowArrayStream {
- let record_batch = self
- .inner
- .read_file_slice_blocking(&file_slice.inner)
- .expect("Failed to read file slice");
- let schema = record_batch.schema();
+ let log_file_paths: Vec<String> = log_file_paths
+ .iter()
+ .map(|path| {
+ path.to_str()
+ .expect("Failed to convert CxxString to str: Invalid UTF-8 sequence")
+ .to_string()
+ })
+ .collect();
+
+ let options = ReadOptions::new();
+ let stream = rt()
+ .block_on(
+ self.inner
+ .read_file_slice_by_paths(base_file_path, log_file_paths, &options),
+ )
+ .expect("Failed to read file slice by paths");
+
+ let batches: Vec<_> = rt()
+ .block_on(stream.try_collect())
+ .expect("Failed to collect record batches");
+
+ if batches.is_empty() {
+ panic!("No record batches read from file slice");
Review Comment:
In the C++ bindings, line 118 panics if no record batches are read. This is problematic because:
1. Empty results are valid (e.g., empty table, all rows filtered out)
2. Panicking in FFI code can cause undefined behavior
3. The error should be propagated to C++ via the FFI boundary properly
Instead of panicking, handle the empty case gracefully by either returning an empty stream or a proper error result that C++ can handle.
```suggestion
// No record batches were read; return a null pointer so the caller can
// treat this as an empty result or error as appropriate.
return std::ptr::null_mut();
```
##########
crates/core/src/file_group/reader.rs:
##########
@@ -91,21 +93,54 @@ impl FileGroupReader {
K: AsRef<str>,
V: Into<String>,
{
- tokio::runtime::Builder::new_current_thread()
- .enable_all()
- .build()?
- .block_on(async {
- let mut resolver = OptionResolver::new_with_options(base_uri, options);
- resolver.resolve_options().await?;
- let hudi_configs = Arc::new(HudiConfigs::new(resolver.hudi_options));
- let storage =
- Storage::new(Arc::new(resolver.storage_options), hudi_configs.clone())?;
-
- Ok(Self {
- hudi_configs,
- storage,
- })
+ // Collect options upfront so we can move them across thread boundaries if needed
+ let options_vec: Vec<(String, String)> = options
+ .into_iter()
+ .map(|(k, v)| (k.as_ref().to_string(), v.into()))
+ .collect();
+ let base_uri = base_uri.to_string();
+
+ // Helper function to run the async resolution
+ async fn resolve_reader(
+ base_uri: String,
+ options_vec: Vec<(String, String)>,
+ ) -> Result<FileGroupReader> {
+ let mut resolver = OptionResolver::new_with_options(&base_uri, options_vec);
+ resolver.resolve_options().await?;
+ let hudi_configs = Arc::new(HudiConfigs::new(resolver.hudi_options));
+ let storage =
+ Storage::new(Arc::new(resolver.storage_options), hudi_configs.clone())?;
+
+ Ok(FileGroupReader {
+ hudi_configs,
+ storage,
})
+ }
+
+ // Check if we're already in a tokio runtime
+ match tokio::runtime::Handle::try_current() {
+ Ok(_handle) => {
+ // We're inside a runtime, spawn a new thread with its own runtime
+ std::thread::scope(|s| {
+ s.spawn(|| {
+ tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap()
+ .block_on(resolve_reader(base_uri, options_vec))
+ })
+ .join()
+ .unwrap()
+ })
+ }
+ Err(_) => {
+ // No runtime, create one
+ tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()?
+ .block_on(resolve_reader(base_uri, options_vec))
+ }
+ }
Review Comment:
The runtime creation and thread spawning logic in the `new` method is complex and could have concurrency issues. When already in a tokio runtime, spawning a new thread with `std::thread::scope` that creates another runtime may cause issues:
1. The scoped thread pattern forces synchronous waiting, which defeats the purpose of async
2. Creating a new runtime in a thread spawned from an existing runtime is unusual and may lead to unexpected behavior
3. The `.unwrap()` calls on lines 129-130 and 133 can panic, which is undesirable in a library
Consider simplifying this by either:
- Requiring callers to use the async `new` method when already in a runtime
- Using `tokio::task::spawn_blocking` instead of `std::thread::scope`
- Documenting that this constructor should not be called from within an async context
##########
cpp/src/lib.rs:
##########
@@ -87,74 +84,41 @@ pub fn new_file_group_reader_with_options(
}
impl HudiFileGroupReader {
- pub fn read_file_slice_by_base_file_path(
+ pub fn read_file_slice_by_paths(
&self,
- relative_path: &CxxString,
+ base_file_path: &CxxString,
+ log_file_paths: &CxxVector<CxxString>,
) -> *mut ffi::ArrowArrayStream {
- let relative_path = relative_path
+ let base_file_path = base_file_path
.to_str()
.expect("Failed to convert CxxString to str: Invalid UTF-8 sequence");
- let record_batch = self
- .inner
- .read_file_slice_by_base_file_path_blocking(relative_path)
- .expect("Failed to read file batch");
- let schema = record_batch.schema();
-
- create_raw_pointer_for_record_batches(vec![record_batch], schema)
- }
-
- pub fn read_file_slice(&self, file_slice: &HudiFileSlice) -> *mut ffi::ArrowArrayStream {
- let record_batch = self
- .inner
- .read_file_slice_blocking(&file_slice.inner)
- .expect("Failed to read file slice");
- let schema = record_batch.schema();
+ let log_file_paths: Vec<String> = log_file_paths
+ .iter()
+ .map(|path| {
+ path.to_str()
+ .expect("Failed to convert CxxString to str: Invalid UTF-8 sequence")
+ .to_string()
+ })
+ .collect();
+
+ let options = ReadOptions::new();
+ let stream = rt()
+ .block_on(
+ self.inner
+ .read_file_slice_by_paths(base_file_path, log_file_paths, &options),
+ )
+ .expect("Failed to read file slice by paths");
Review Comment:
Multiple `.expect()` calls with panics on lines 93-94, 99-102, and 110-111 can cause the process to abort. In FFI code, panics are particularly dangerous as they can corrupt the calling C++ program's state. These should be changed to return proper error results through the FFI boundary using Result types or error codes.
##########
crates/core/src/file_group/reader.rs:
##########
@@ -305,24 +320,333 @@ impl FileGroupReader {
all_batches.extend(log_batches);
let merger = RecordMerger::new(schema.clone(), self.hudi_configs.clone());
- merger.merge_record_batches(all_batches)
+ let batch = merger.merge_record_batches(all_batches)?;
+
+ // Apply projection if specified
+ let batch = if let Some(cols) = projection {
+ Self::apply_projection(&batch, cols)?
+ } else {
+ batch
+ };
+
+ // Apply row predicate if specified
+ let batch = if let Some(predicate) = row_predicate {
+ Self::apply_row_predicate(&batch, predicate)?
+ } else {
+ batch
+ };
+
+ Ok(batch)
}
}
- /// Same as [FileGroupReader::read_file_slice_from_paths], but blocking.
- pub fn read_file_slice_from_paths_blocking<I, S>(
+ /// Reads a file slice as a stream of record batches with [ReadOptions].
+ ///
+ /// This is the primary read API for FileGroupReader. It returns a stream that
+ /// yields record batches as they are read.
+ ///
+ /// For COW tables or read-optimized mode (base file only), this returns a true
+ /// streaming iterator from the underlying parquet file, yielding batches as they
+ /// are read without loading all data into memory.
+ ///
+ /// For MOR tables with log files, this falls back to the collect-and-merge approach
+ /// and yields the merged result as a single batch.
+ ///
+ /// # Arguments
+ /// * `file_slice` - The file slice to read.
+ /// * `options` - Read options for configuring the read operation:
+ /// - `projection`: Column names to project (select)
+ /// - `row_predicate`: Row-level filter predicate
+ /// - `batch_size`: Target rows per batch
+ ///
+ /// # Returns
+ /// A stream of record batches. The stream owns all necessary data and is `'static`.
+ ///
+ /// # Example
+ /// ```ignore
+ /// use futures::StreamExt;
+ ///
+ /// let options = ReadOptions::new().with_batch_size(4096);
+ /// let mut stream = reader.read_file_slice(&file_slice, &options).await?;
+ ///
+ /// while let Some(result) = stream.next().await {
+ /// let batch = result?;
+ /// // Process batch...
+ /// }
+ /// ```
+ pub async fn read_file_slice(
+ &self,
+ file_slice: &FileSlice,
+ options: &ReadOptions,
+ ) -> Result<BoxStream<'static, Result<RecordBatch>>> {
+ let base_file_path = file_slice.base_file_relative_path()?;
+ let log_file_paths: Vec<String> = if file_slice.has_log_file() {
+ file_slice
+ .log_files
+ .iter()
+ .map(|log_file| file_slice.log_file_relative_path(log_file))
+ .collect::<Result<Vec<String>>>()?
+ } else {
+ vec![]
+ };
+
+ let use_read_optimized: bool = self
+ .hudi_configs
+ .get_or_default(HudiReadConfig::UseReadOptimizedMode)
+ .into();
+ let base_file_only = log_file_paths.is_empty() || use_read_optimized;
+
+ if base_file_only {
+ // True streaming: return the parquet stream directly
+ self.read_parquet_file_stream(&base_file_path, options).await
+ } else {
+ // Fallback: collect + merge, then yield as single-item stream
+ let batch = self
+ .read_file_slice_from_paths_with_opts(&base_file_path, log_file_paths, options)
+ .await?;
+ Ok(Box::pin(futures::stream::once(async { Ok(batch) })))
+ }
Review Comment:
The streaming implementation for MOR tables with log files (lines 403-408) falls back to collecting all data into memory first via `read_file_slice_from_paths_with_opts`, then yields it as a single batch. This defeats the purpose of streaming for MOR tables and could cause memory issues with large datasets.
Consider documenting this limitation clearly in the function documentation, or better yet, implementing true streaming for MOR tables in a future iteration.
##########
crates/core/tests/table_read_tests.rs:
##########
@@ -35,21 +37,23 @@ mod v6_tables {
mod snapshot_queries {
use super::*;
- #[test]
- fn test_empty_table() -> Result<()> {
+ #[tokio::test]
+ async fn test_empty_table() -> Result<()> {
for base_url in SampleTable::V6Empty.urls() {
- let hudi_table = Table::new_blocking(base_url.path())?;
- let records = hudi_table.read_snapshot_blocking(empty_filters())?;
+ let hudi_table = Table::new(base_url.path()).await?;
+ let stream = hudi_table.read_snapshot(ReadOptions::new()).await?;
+ let records: Vec<_> = stream.try_collect().await?;
assert!(records.is_empty());
}
Ok(())
}
Review Comment:
The test at line 41 is named `test_empty_table` but there are multiple tests with this same name in the file (lines 41, 312, 418). While Rust allows this in different modules, it makes it harder to identify which specific test failed when running tests. Consider using more descriptive names like `test_v6_empty_table_snapshot`, `test_v6_empty_table_incremental`, `test_v8_empty_table_snapshot` etc.
##########
crates/core/src/table/mod.rs:
##########
@@ -642,153 +414,198 @@ impl Table {
)
}
- /// Get all the latest records in the table.
+ /// Create a [FileGroupReader] for this table.
///
/// # 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())
+ /// * `options` - Read options for configuring the reader
+ pub fn create_file_group_reader(&self, options: &ReadOptions) -> Result<FileGroupReader> {
+ let mut read_configs: Vec<(HudiReadConfig, String)> = vec![];
+
+ if let Some(ref ts) = options.as_of_timestamp {
+ let formatted = format_timestamp(ts, &self.timezone())?;
+ read_configs.push((HudiReadConfig::FileGroupEndTimestamp, formatted));
+ } else if let Some(ts) = self.timeline.get_latest_commit_timestamp_as_option() {
+ read_configs.push((HudiReadConfig::FileGroupEndTimestamp, ts.to_string()));
}
- }
- /// Same as [Table::read_snapshot], but blocking.
- pub fn read_snapshot_blocking<I, S>(&self, filters: I) -> Result<Vec<RecordBatch>>
- where
- I: IntoIterator<Item = (S, S, S)>,
- S: AsRef<str>,
- {
- tokio::runtime::Builder::new_current_thread()
- .enable_all()
- .build()?
- .block_on(async { self.read_snapshot(filters).await })
+ if let Some(ref ts) = options.start_timestamp {
+ let formatted = format_timestamp(ts, &self.timezone())?;
+ read_configs.push((HudiReadConfig::FileGroupStartTimestamp, formatted));
+ }
+
+ if let Some(batch_size) = options.batch_size {
+ read_configs.push((HudiReadConfig::StreamBatchSize, batch_size.to_string()));
+ }
+
+ self.create_file_group_reader_with_options(read_configs)
}
- /// Get all the records in the table at a given timestamp.
+ /// Read records from the table as a stream using [ReadOptions].
+ ///
+ /// This is the primary read API. It returns a stream that yields record batches
+ /// as they are read, enabling memory-efficient processing of large tables.
+ ///
+ /// For COW tables or read-optimized mode, each parquet file is streamed batch-by-batch.
+ /// For MOR tables with log files, each file slice is read, merged, and yielded as a batch.
///
/// # Arguments
- /// * `timestamp` - The timestamp which records associated with.
- /// * `filters` - Partition filters to apply.
- pub async fn read_snapshot_as_of<I, S>(
+ /// * `options` - Read options for configuring the read operation:
+ /// - `as_of_timestamp`: If set, reads data as of this timestamp (time travel)
+ /// - `partition_filter`: Filters for partition pruning
+ /// - `projection`: Column names to project (select)
+ /// - `row_predicate`: Row-level filter predicate
+ /// - `batch_size`: Target rows per batch (effective for COW/read-optimized)
+ ///
+ /// # Example
+ ///
+ /// ```ignore
+ /// use hudi_core::table::{Table, ReadOptions};
+ /// use hudi_core::expr::filter::col;
+ /// use futures::StreamExt;
+ ///
+ /// // Read latest snapshot
+ /// let mut stream = table.read_snapshot(ReadOptions::new()).await?;
+ ///
+ /// // Read with partition filter and column projection
+ /// let options = ReadOptions::new()
+ /// .with_partition_filter(col("date").eq("2024-01-01"))
+ /// .with_column(["id", "name", "value"])
+ /// .with_batch_size(4096);
+ ///
+ /// let mut stream = table.read_snapshot(options).await?;
+ /// while let Some(result) = stream.next().await {
+ /// let batch = result?;
+ /// // Process batch...
+ /// }
+ ///
+ /// // Time travel: read snapshot as of a specific timestamp
+ /// let options = ReadOptions::new().as_of("2024-01-01T12:00:00Z");
+ /// let mut stream = table.read_snapshot(options).await?;
+ /// ```
+ pub async fn read_snapshot(
&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
- }
+ options: ReadOptions,
+ ) -> Result<BoxStream<'static, Result<RecordBatch>>> {
+ // Determine the timestamp: use as_of_timestamp from options, or latest
+ let timestamp = if let Some(ref ts) = options.as_of_timestamp {
+ format_timestamp(ts, &self.timezone())?
+ } else if let Some(ts) = self.timeline.get_latest_commit_timestamp_as_option() {
+ ts.to_string()
+ } else {
+ return Ok(Box::pin(futures::stream::empty()));
+ };
- /// Same as [Table::read_snapshot_as_of], but blocking.
- pub fn read_snapshot_as_of_blocking<I, S>(
- &self,
- timestamp: &str,
- filters: I,
- ) -> Result<Vec<RecordBatch>>
- where
- I: IntoIterator<Item = (S, S, S)>,
- S: AsRef<str>,
- {
- tokio::runtime::Builder::new_current_thread()
- .enable_all()
- .build()?
- .block_on(async { self.read_snapshot_as_of(timestamp, filters).await })
+ self.read_snapshot_internal(×tamp, options).await
}
async fn read_snapshot_internal(
&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,
+ ) -> Result<BoxStream<'static, Result<RecordBatch>>> {
+ let file_slices = self
+ .get_file_slices_internal(timestamp, &options.partition_filter)
+ .await?;
+
+ // Build the base read config options
+ let mut read_configs: Vec<(HudiReadConfig, String)> =
+ vec![(HudiReadConfig::FileGroupEndTimestamp, timestamp.to_string())];
+
+ // Add batch size if specified
+ if let Some(batch_size) = options.batch_size {
+ read_configs.push((HudiReadConfig::StreamBatchSize, batch_size.to_string()));
+ }
+
+ let fg_reader = self.create_file_group_reader_with_options(read_configs)?;
+
+ // Create a stream that yields batches from all file slices
+ // Each file slice produces a 'static stream, which we flatten
+ let stream = futures::stream::iter(file_slices)
+ .then(move |file_slice| {
+ let fg_reader = fg_reader.clone();
+ let options = options.clone();
+ async move {
+ fg_reader.read_file_slice(&file_slice, &options).await
+ }
+ })
+ .try_flatten();
+
+ Ok(Box::pin(stream))
}
Review Comment:
The streaming read APIs return `BoxStream<'static, Result<RecordBatch>>`, but for truly large datasets, users may want to apply backpressure or limit memory usage. Consider documenting best practices for:
1. How to limit concurrent processing of batches
2. How to apply backpressure when consumers are slower than producers
3. Memory implications of the 'static lifetime
Additionally, consider whether a bounded buffer or backpressure mechanism should be built into the stream.
##########
crates/core/src/file_group/reader.rs:
##########
@@ -305,24 +320,333 @@ impl FileGroupReader {
all_batches.extend(log_batches);
let merger = RecordMerger::new(schema.clone(), self.hudi_configs.clone());
- merger.merge_record_batches(all_batches)
+ let batch = merger.merge_record_batches(all_batches)?;
+
+ // Apply projection if specified
+ let batch = if let Some(cols) = projection {
+ Self::apply_projection(&batch, cols)?
+ } else {
+ batch
+ };
+
+ // Apply row predicate if specified
+ let batch = if let Some(predicate) = row_predicate {
+ Self::apply_row_predicate(&batch, predicate)?
+ } else {
+ batch
+ };
+
+ Ok(batch)
}
}
- /// Same as [FileGroupReader::read_file_slice_from_paths], but blocking.
- pub fn read_file_slice_from_paths_blocking<I, S>(
+ /// Reads a file slice as a stream of record batches with [ReadOptions].
+ ///
+ /// This is the primary read API for FileGroupReader. It returns a stream that
+ /// yields record batches as they are read.
+ ///
+ /// For COW tables or read-optimized mode (base file only), this returns a true
+ /// streaming iterator from the underlying parquet file, yielding batches as they
+ /// are read without loading all data into memory.
+ ///
+ /// For MOR tables with log files, this falls back to the collect-and-merge approach
+ /// and yields the merged result as a single batch.
+ ///
+ /// # Arguments
+ /// * `file_slice` - The file slice to read.
+ /// * `options` - Read options for configuring the read operation:
+ /// - `projection`: Column names to project (select)
+ /// - `row_predicate`: Row-level filter predicate
+ /// - `batch_size`: Target rows per batch
+ ///
+ /// # Returns
+ /// A stream of record batches. The stream owns all necessary data and is `'static`.
+ ///
+ /// # Example
+ /// ```ignore
+ /// use futures::StreamExt;
+ ///
+ /// let options = ReadOptions::new().with_batch_size(4096);
+ /// let mut stream = reader.read_file_slice(&file_slice, &options).await?;
+ ///
+ /// while let Some(result) = stream.next().await {
+ /// let batch = result?;
+ /// // Process batch...
+ /// }
+ /// ```
+ pub async fn read_file_slice(
+ &self,
+ file_slice: &FileSlice,
+ options: &ReadOptions,
+ ) -> Result<BoxStream<'static, Result<RecordBatch>>> {
+ let base_file_path = file_slice.base_file_relative_path()?;
+ let log_file_paths: Vec<String> = if file_slice.has_log_file() {
+ file_slice
+ .log_files
+ .iter()
+ .map(|log_file| file_slice.log_file_relative_path(log_file))
+ .collect::<Result<Vec<String>>>()?
+ } else {
+ vec![]
+ };
+
+ let use_read_optimized: bool = self
+ .hudi_configs
+ .get_or_default(HudiReadConfig::UseReadOptimizedMode)
+ .into();
+ let base_file_only = log_file_paths.is_empty() || use_read_optimized;
+
+ if base_file_only {
+ // True streaming: return the parquet stream directly
+ self.read_parquet_file_stream(&base_file_path, options).await
+ } else {
+ // Fallback: collect + merge, then yield as single-item stream
+ let batch = self
+ .read_file_slice_from_paths_with_opts(&base_file_path, log_file_paths, options)
+ .await?;
+ Ok(Box::pin(futures::stream::once(async { Ok(batch) })))
+ }
+ }
+
+ /// Reads a file slice specified by file paths as a stream of record batches.
+ ///
+ /// This is a convenience method that constructs a [FileSlice] from the given paths
+ /// and delegates to [Self::read_file_slice].
+ ///
+ /// # Arguments
+ /// * `base_file_path` - Relative path to the base file (e.g., "city=chennai/file.parquet")
+ /// * `log_file_paths` - Optional iterator of relative paths to log files
+ /// * `options` - Read options for configuring the read operation
+ ///
+ /// # Returns
+ /// A stream of record batches.
+ ///
+ /// # Example
+ /// ```ignore
+ /// use futures::StreamExt;
+ ///
+ /// let options = ReadOptions::new();
+ /// let mut stream = reader.read_file_slice_by_paths(
+ /// "city=chennai/file-id-0_0-7-24_20240101120000.parquet",
+ /// [".file-id-0_20240101120000.log.1_0-51-115"],
+ /// &options,
+ /// ).await?;
+ ///
+ /// while let Some(result) = stream.next().await {
+ /// let batch = result?;
+ /// // Process batch...
+ /// }
+ /// ```
+ pub async fn read_file_slice_by_paths<I, S>(
&self,
base_file_path: &str,
log_file_paths: I,
- ) -> Result<RecordBatch>
+ options: &ReadOptions,
+ ) -> Result<BoxStream<'static, Result<RecordBatch>>>
where
I: IntoIterator<Item = S>,
S: AsRef<str>,
{
- tokio::runtime::Builder::new_current_thread()
- .enable_all()
- .build()?
- .block_on(self.read_file_slice_from_paths(base_file_path, log_file_paths))
+ use std::path::Path;
+
+ // Parse base file path to extract partition path and file name
+ let path = Path::new(base_file_path);
+ let partition_path = path
+ .parent()
+ .map(|p| p.to_string_lossy().to_string())
+ .unwrap_or_default();
+ let base_file_name = path
+ .file_name()
+ .ok_or_else(|| {
+ CoreError::FileGroup(format!(
+ "Cannot extract file name from base file path: {base_file_path}"
+ ))
+ })?
+ .to_string_lossy();
+
+ // Create a FileGroup from the base file
+ let mut file_group = FileGroup::new_with_base_file_name(&base_file_name, &partition_path)?;
+
+ // Add log files if provided
+ for log_file_path in log_file_paths {
+ let log_path = Path::new(log_file_path.as_ref());
+ let log_file_name = log_path
+ .file_name()
+ .ok_or_else(|| {
+ CoreError::FileGroup(format!(
+ "Cannot extract file name from log file path: {}",
+ log_file_path.as_ref()
+ ))
+ })?
+ .to_string_lossy();
+ file_group.add_log_file_from_name(&log_file_name)?;
+ }
+
+ // Get the file slice (there should be exactly one after adding base file)
+ let file_slice = file_group
+ .file_slices
+ .values()
+ .next()
+ .ok_or_else(|| CoreError::FileGroup("No file slice found in file group".to_string()))?
+ .clone();
+
+ // Delegate to the primary read API
+ self.read_file_slice(&file_slice, options).await
+ }
+
+ /// Reads a parquet file as a stream of record batches.
+ ///
+ /// This method propagates the underlying parquet stream, applying Hudi-specific
+ /// filtering (commit time filtering for incremental reads) to each batch.
+ /// The returned stream owns all necessary data and is `'static`.
+ async fn read_parquet_file_stream(
+ &self,
+ relative_path: &str,
+ options: &ReadOptions,
+ ) -> Result<BoxStream<'static, Result<RecordBatch>>> {
+ let batch_size = options.batch_size.unwrap_or(8192);
+ let parquet_options = ParquetReadOptions::with_batch_size(batch_size);
+
+ let stream = self
+ .storage
+ .get_parquet_file_stream(relative_path, parquet_options)
+ .await
+ .map_err(|e| ReadFileSliceError(format!("Failed to read path {relative_path}: {e:?}")))?;
+
+ // Clone data for the closure so the stream is 'static
+ let reader = self.clone();
+ let projection = options.projection.clone();
+ let row_predicate = options.row_predicate.clone();
+
+ let mapped_stream = stream.map(move |result| {
+ let batch = result
+ .map_err(|e| ReadFileSliceError(format!("Failed to read batch: {e:?}")))?;
+
+ // Apply Hudi commit time filtering
+ let batch = if let Some(mask) = reader.create_filtering_mask_for_base_file_records(&batch)? {
+ filter_record_batch(&batch, &mask)
+ .map_err(|e| ReadFileSliceError(format!("Failed to filter records: {e:?}")))?
+ } else {
+ batch
+ };
+
+ // Apply column projection if specified
+ let batch = if let Some(ref cols) = projection {
+ FileGroupReader::apply_projection(&batch, cols)?
+ } else {
+ batch
+ };
+
+ // Apply row predicate if specified
+ let batch = if let Some(ref predicate) = row_predicate {
+ FileGroupReader::apply_row_predicate(&batch, predicate)?
+ } else {
+ batch
+ };
+
+ Ok(batch)
+ });
+
+ Ok(Box::pin(mapped_stream))
+ }
+
+ /// Creates an instant range for log file scanning using ReadOptions.
+ fn create_instant_range_for_log_file_scan_with_opts(&self, options: &ReadOptions) -> InstantRange {
+ let timezone = self
+ .hudi_configs
+ .get_or_default(HudiTableConfig::TimelineTimezone)
+ .into();
+
+ // Use as_of_timestamp from options if provided, otherwise fall back to config
+ let start_timestamp = self
+ .hudi_configs
+ .try_get(HudiReadConfig::FileGroupStartTimestamp)
+ .map(|v| -> String { v.into() });
+
+ let end_timestamp = options
+ .as_of_timestamp
+ .clone()
+ .or_else(|| {
+ self.hudi_configs
+ .try_get(HudiReadConfig::FileGroupEndTimestamp)
+ .map(|v| -> String { v.into() })
+ });
+
+ InstantRange::new(timezone, start_timestamp, end_timestamp, false, true)
+ }
+
+ /// Applies column projection to a record batch.
+ fn apply_projection(batch: &RecordBatch, columns: &[String]) -> Result<RecordBatch> {
+ let schema = batch.schema();
+ let indices: Vec<usize> = columns
+ .iter()
+ .filter_map(|col| schema.index_of(col).ok())
+ .collect();
+
+ if indices.is_empty() {
+ return Err(ReadFileSliceError(
+ "No matching columns found for projection".to_string(),
+ ));
+ }
+
+ batch
+ .project(&indices)
+ .map_err(|e| ReadFileSliceError(format!("Failed to project columns: {e:?}")))
+ }
+
+ /// Applies a row predicate filter to a record batch.
+ fn apply_row_predicate(batch: &RecordBatch, predicate: &Filter) -> Result<RecordBatch> {
+ let schema = batch.schema();
+ let schemable = SchemableFilter::try_from((predicate.clone(), schema.as_ref()))?;
+
+ // Get the column to filter on
+ let column = batch.column_by_name(&predicate.field_name).ok_or_else(|| {
+ ReadFileSliceError(format!("Column '{}' not found for predicate", predicate.field_name))
+ })?;
+
+ // Apply the comparison to get a boolean mask
+ let mask = schemable.apply_comparsion(column)?;
+
+ // Filter the batch
+ filter_record_batch(batch, &mask)
+ .map_err(|e| ReadFileSliceError(format!("Failed to apply row predicate: {e:?}")))
+ }
Review Comment:
The `apply_row_predicate` method only supports single-field predicates. Line 603 gets a single column by `predicate.field_name`, but the Filter type might support more complex predicates with multiple fields or nested conditions. If a compound filter is passed, this will fail or only partially apply the filter.
Consider adding validation to ensure only simple single-field predicates are supported, or extend the implementation to handle compound filters properly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: support streaming read in table and file group APIs [hudi-rs]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #507:
URL: https://github.com/apache/hudi-rs/pull/507#issuecomment-3707514535
## [Codecov](https://app.codecov.io/gh/apache/hudi-rs/pull/507?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `59.11950%` with `130 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 83.23%. Comparing base ([`065fd31`](https://app.codecov.io/gh/apache/hudi-rs/commit/065fd3131e855cb26d0882eb2e7b814539f711a2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`2508d28`](https://app.codecov.io/gh/apache/hudi-rs/commit/2508d280e5b2bb59fa88c854f292f121a14ffdb3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi-rs/pull/507?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [crates/core/src/file\_group/reader.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/507?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Ffile_group%2Freader.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL2ZpbGVfZ3JvdXAvcmVhZGVyLnJz) | 55.81% | [76 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/507?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/core/src/table/mod.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/507?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Ftable%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL3RhYmxlL21vZC5ycw==) | 63.88% | [26 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/507?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/core/src/storage/mod.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/507?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Fstorage%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL3N0b3JhZ2UvbW9kLnJz) | 18.18% | [18 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/507?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [python/src/internal.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/507?src=pr&el=tree&filepath=python%2Fsrc%2Finternal.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHl0aG9uL3NyYy9pbnRlcm5hbC5ycw==) | 0.00% | [7 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/507?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/datafusion/src/lib.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/507?src=pr&el=tree&filepath=crates%2Fdatafusion%2Fsrc%2Flib.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2RhdGFmdXNpb24vc3JjL2xpYi5ycw==) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/507?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #507 +/- ##
==========================================
- Coverage 85.65% 83.23% -2.43%
==========================================
Files 65 66 +1
Lines 3897 4068 +171
==========================================
+ Hits 3338 3386 +48
- Misses 559 682 +123
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/hudi-rs/pull/507?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org