goutamadwant commented on code in PR #701:
URL: https://github.com/apache/hudi-rs/pull/701#discussion_r4409591544
##########
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:
You're right. Replaced the directory-order assumption with explicit
stream::iter inputs. Tests now cover discovery errors and real
malformed-partition loading in both valid-then-invalid and invalid-first order.
##########
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:
Extracted shared planning and metadata setup for both paths, preserving
their existing estimator initialization order. Added coverage for metadata
initialization fallback and verified that corrupt rollback metadata still fails
both APIs.
--
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]
