goutamadwant commented on code in PR #701:
URL: https://github.com/apache/hudi-rs/pull/701#discussion_r4409590375
##########
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:
Updated discovery to list root directories once, exclude metadata
directories, and lazily stream each remaining subtree. Added regression
coverage for the exact listing prefixes and lazy opening of subsequent subtrees.
##########
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:
Removed both caps, the byte accounting, and cap-specific tests. Updated the
README and reader spec accordingly. Deduplication and bounded partition-listing
concurrency remain.
##########
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:
Restored main's version of the row. It now correctly documents snapshot-only
estimates and None for incremental queries.
--
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]
