yihua commented on code in PR #701:
URL: https://github.com/apache/hudi-rs/pull/701#discussion_r4406245518
##########
crates/core/src/table/listing.rs:
##########
@@ -32,10 +33,18 @@ use crate::table::partition::{
};
use crate::timeline::completion_time::CompletionTimeView;
use dashmap::DashMap;
+use futures::stream::BoxStream;
use futures::{StreamExt, TryStreamExt, stream};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::sync::Arc;
+/// A storage-backed streamed scan fails closed before retaining more unique
+/// partition paths or path text than these bounds. Tables beyond this scale
+/// should use the metadata table, whose files partition avoids object-path
+/// deduplication.
+const MAX_STREAMED_PARTITION_PATHS: usize = 100_000;
Review Comment:
non-blocking: I wonder if the partition-count and path-byte caps are needed,
since the eager path already holds every pruned partition path plus the full
file-group map. Dropping them (constants, byte counter, tests, and the
README/reader-spec text) would leave the dedup set as the only state.
##########
crates/core/src/storage/mod.rs:
##########
@@ -398,6 +400,47 @@ impl Storage {
}
Ok(file_metadata)
}
+
+ /// Recursively stream files below the table base path.
+ ///
+ /// Each item is a table-relative object path. Cloud object stores paginate
+ /// this stream internally, while local storage walks directory entries
+ /// incrementally. Unlike [`Self::list_files`], this does not materialize a
+ /// delimiter result containing every sibling prefix.
+ pub(crate) fn list_relative_paths_stream(&self) ->
Result<BoxStream<'static, Result<String>>> {
+ let prefix_url = join_url_segments(&self.base_url, &[""])?;
+ let prefix_path = ObjPath::from_url_path(prefix_url.path())?;
+ let prefix = prefix_path.as_ref().trim_end_matches('/').to_string();
+
+ Ok(self
+ .object_store
+ .list(Some(&prefix_path))
Review Comment:
Listing recursively from the table root means object stores return all of
`.hoodie/` first (`.` sorts before alphanumerics), so v6 tables page through
the metadata table and timeline before emitting a partition. Could you list the
root once, skip the metadata dirs as `list_relevant_partition_paths` does, and
recursively list each remaining top-level dir?
##########
crates/core/src/table/mod.rs:
##########
@@ -2177,6 +2349,130 @@ mod tests {
assert!(!incremental_slices.is_empty());
}
+ #[tokio::test]
+ async fn hudi_table_get_file_slices_stream_matches_eager_snapshot() ->
Result<()> {
+ use futures::TryStreamExt;
+
+ let base_url = SampleTable::V6SimplekeygenNonhivestyle.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+ let options = ReadOptions::new().with_filters([("byteField", ">=",
"20")])?;
+
+ let mut eager = hudi_table.get_file_slices(&options).await?;
+ let mut streamed = hudi_table
+ .get_file_slices_stream(&options)
+ .await?
+ .try_collect::<Vec<_>>()
+ .await?;
+
+ let compare = |left: &FileSlice, right: &FileSlice| {
+ left.partition_path
+ .cmp(&right.partition_path)
+ .then_with(|| left.file_id().cmp(right.file_id()))
+ .then_with(|| {
+ left.creation_instant_time()
+ .cmp(right.creation_instant_time())
+ })
+ };
+ eager.sort_by(compare);
+ streamed.sort_by(compare);
+ assert_eq!(streamed, eager);
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn
hudi_table_get_file_slices_stream_matches_eager_metadata_snapshot() ->
Result<()> {
+ use futures::TryStreamExt;
+
+ let base_url = SampleTable::V9TxnsSimpleMeta.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+ assert!(hudi_table.is_metadata_table_enabled());
+
+ let mut eager = hudi_table.get_file_slices(&ReadOptions::new()).await?;
+ let mut streamed = hudi_table
+ .get_file_slices_stream(&ReadOptions::new())
+ .await?
+ .try_collect::<Vec<_>>()
+ .await?;
+ eager.sort_by(|left, right| {
+ left.partition_path
+ .cmp(&right.partition_path)
+ .then_with(|| left.file_id().cmp(right.file_id()))
+ });
+ streamed.sort_by(|left, right| {
+ left.partition_path
+ .cmp(&right.partition_path)
+ .then_with(|| left.file_id().cmp(right.file_id()))
+ });
+ assert_eq!(streamed, eager);
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn hudi_table_get_file_slices_stream_matches_eager_incremental() ->
Result<()> {
+ use futures::TryStreamExt;
+
+ let base_url = SampleTable::V6Nonpartitioned.url_to_cow();
+ let hudi_table = Table::new(base_url.path()).await?;
+ let options = ReadOptions::new()
+ .with_query_type(QueryType::Incremental)
+ .with_start_timestamp(EARLIEST_START_TIMESTAMP);
+
+ let eager = hudi_table.get_file_slices(&options).await?;
+ let streamed = hudi_table
+ .get_file_slices_stream(&options)
+ .await?
+ .try_collect::<Vec<_>>()
+ .await?;
+ assert_eq!(streamed, eager);
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn
hudi_table_get_file_slices_stream_emits_before_later_partition_error() ->
Result<()> {
Review Comment:
Does this test depend on directory order? `LocalFileSystem::list` walks
without a sorter, so if `z-bad` is yielded first the first `try_next` errors;
driving `relevant_partition_paths_from_objects` with a fixed `stream::iter`
would make the order explicit.
##########
README.md:
##########
@@ -343,7 +343,8 @@ All read APIs accept a `ReadOptions` (Rust) /
`HudiReadOptions` (Python) value.
| Stage | API
| Description
|
|-----------------|------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------|
| Query planning | `get_file_slices(options)`
| Get the file slices the read targets, dispatched on
`options.query_type`. To bucket for parallel reads, call
`hudi::util::collection::split_into_chunks` on the result. |
-| | `compute_table_stats(options)`
| Estimated `(num_rows, byte_size)` for scan planning, derived from the
metadata table. Snapshot only. Returns `None` for incremental queries, and
whenever the estimate cannot be computed (no metadata table, non-Parquet base
files, footer sampling failure). |
+| | `get_file_slices_stream(options)` (Rust)
| Stream targeted file slices as partition planning completes. Global
order is unspecified. Storage-backed snapshots use recursive object listing
(paginated by cloud backends), bound in-flight partition listings, and fail
closed before retaining more than 100,000 unique partition paths or 64 MiB of
path text for exact-once emission; use the metadata table or eager API beyond
those limits. Metadata-table records and the incremental touched-file-group
lookup remain eager. |
+| | `compute_table_stats(options)`
| Estimated `(num_rows, byte_size)` for scan planning. Snapshot (default)
uses the metadata table; incremental aggregates from changed file slices.
Returns `None` when stats cannot be computed. |
Review Comment:
non-blocking: this row looks like a rebase leftover, since
`compute_table_stats` still returns `None` for incremental and #716 fixed the
docs to say so. Could you keep main's version of the row?
##########
crates/core/src/table/mod.rs:
##########
@@ -519,6 +519,169 @@ impl Table {
}
}
+ /// Stream the [`FileSlice`]s targeted by a read.
+ ///
+ /// Snapshot queries list files with bounded partition concurrency and emit
+ /// slices as each partition completes instead of retaining the table-wide
+ /// file-group and slice collections. Metadata-table-backed snapshots still
+ /// fetch the metadata records eagerly because that reader does not yet
+ /// expose a row stream. Incremental planning retains its existing eager
+ /// touched-file-group lookup, then exposes the resulting slices as a
stream.
+ ///
+ /// The stream has no stable global ordering; consumers that require one
+ /// should sort by partition path and file ID after collection. Listing
work
+ /// is driven by polling the stream; dropping it cancels the in-flight
+ /// partition futures without spawning detached tasks.
+ ///
+ /// # Example
+ /// ```ignore
+ /// use futures::TryStreamExt;
+ /// use hudi::table::ReadOptions;
+ ///
+ /// let mut slices =
table.get_file_slices_stream(&ReadOptions::new()).await?;
+ /// while let Some(slice) = slices.try_next().await? {
+ /// schedule(slice);
+ /// }
+ /// ```
+ pub async fn get_file_slices_stream(
+ &self,
+ options: &ReadOptions,
+ ) -> Result<futures::stream::BoxStream<'static, Result<FileSlice>>> {
+ use futures::StreamExt;
+ use futures::stream;
+
+ let prepared = self.prepare_reader_options(options)?;
+ let base_file_only = self.is_base_file_only(&prepared)?;
+ match prepared.query_type()? {
+ QueryType::Snapshot => {
+ let Some(timestamp) = prepared.end_timestamp() else {
+ return Ok(stream::empty().boxed());
+ };
+ self.get_file_slices_stream_inner(timestamp,
&prepared.filters, base_file_only)
+ .await
+ }
+ QueryType::Incremental => {
+ let (Some(start), Some(end)) =
+ (prepared.start_timestamp(), prepared.end_timestamp())
+ else {
+ return Ok(stream::empty().boxed());
+ };
+ self.get_file_slices_between_stream_inner(
+ start,
+ end,
+ &prepared.filters,
+ base_file_only,
+ )
+ .await
+ }
+ }
+ }
+
+ async fn get_file_slices_between_stream_inner(
+ &self,
+ start_timestamp: &str,
+ end_timestamp: &str,
+ filters: &[Filter],
+ base_file_only: bool,
+ ) -> Result<futures::stream::BoxStream<'static, Result<FileSlice>>> {
+ use futures::future;
+ use futures::{StreamExt, TryStreamExt};
+
+ let estimator = self.get_or_init_estimator(end_timestamp).await;
+ let file_groups = self
+ .timeline
+ .get_file_groups_between(Some(start_timestamp),
Some(end_timestamp), estimator)
+ .await?;
+ let mut touched: HashMap<String, HashSet<String>> = HashMap::new();
+ for file_group in file_groups {
+ touched
+ .entry(file_group.partition_path)
+ .or_default()
+ .insert(file_group.file_id);
+ }
+
+ let stream = self
+ .get_file_slices_stream_inner(end_timestamp, filters,
base_file_only)
+ .await?;
+ Ok(stream
+ .try_filter(move |slice| {
+ future::ready(
+ touched
+ .get(&slice.partition_path)
+ .is_some_and(|file_ids|
file_ids.contains(slice.file_id())),
+ )
+ })
+ .boxed())
+ }
+
+ async fn get_file_slices_stream_inner(
Review Comment:
non-blocking: the setup up to `metadata_listing` duplicates
`get_file_slices_inner`. Could you pull it into a shared helper so a later fix
to the metadata-table fallback lands in both paths?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
