yihua commented on code in PR #688:
URL: https://github.com/apache/hudi-rs/pull/688#discussion_r3910876010
##########
crates/core/src/file_group/reader_v2/merge_iterator.rs:
##########
@@ -487,22 +531,23 @@ impl Iterator for FileGroupMergeIterator {
Err((e, merge_ms, build_ms)) => {
self.done = true;
self.record_chunk_timing(merge_ms, build_ms);
- Some(Err(core_to_arrow_err(e)))
+ Some(Err(e))
}
}
}
-}
-impl RecordBatchReader for FileGroupMergeIterator {
- fn schema(&self) -> SchemaRef {
- self.output_schema.clone()
+ /// Drive this merge as a `Stream`, for a caller that wants to compose it.
+ ///
+ /// Chunk-at-a-time and demand-driven: nothing is merged until the consumer
+ /// asks, so the extra memory over the merge map itself is one chunk.
+ pub fn into_stream(self) -> BoxStream<'static, Result<RecordBatch>> {
Review Comment:
non-blocking: `unfold` panics if polled again after it returns
`Ready(None)`, and this stream is the public surface of `read_stream`.
`next_chunk`'s sticky `done` already makes the extra poll semantically a no-op,
so a `.fuse()` before `.boxed()` would turn that panic into the `None` the
caller expects, for free.
##########
crates/core/src/file_group/reader_v2/merge_iterator.rs:
##########
@@ -376,92 +374,138 @@ impl Iterator for FileGroupMergeIterator {
// Ok(None) — stream exhausted
// Err((e, merge_ms, build_ms)) — terminal error
type Chunk = (RecordBatch, u64, u64);
- let produced: Result<Option<Chunk>, (CoreError, u64, u64)> = match
&mut self.source {
- MergeSource::Eager { source } => {
- // One source batch = one emitted chunk. The
- // source is a `RecordBatchReader` — lazy (`ParquetSyncReader`
- // doing block_on per row-group) or eager
(`RecordBatchIterator`
- // over a Vec for tests / instant-range-filter fallback). The
- // CoW / empty-FG no-merge path never materializes the whole
- // base file; it pulls one row-group at a time.
- match source.next() {
- None => Ok(None),
- // Eager has no merge work; build timing is the converter
only.
- Some(Ok(batch)) => Ok(Some((batch, 0, 0))),
- Some(Err(e)) => Err((CoreError::ArrowError(e), 0, 0)),
+ let produced: std::result::Result<Option<Chunk>, (CoreError, u64,
u64)> =
+ match &mut self.source {
+ MergeSource::Eager { source } => {
+ // One source batch = one emitted chunk. The no-merge path
+ // (CoW, or a file group with no log files) never
+ // materialises the whole base file; it pulls one row group
+ // at a time.
+ match source.next().await {
+ None => Ok(None),
+ // Eager has no merge work; build timing is the
converter only.
+ Some(Ok(batch)) => Ok(Some((batch, 0, 0))),
+ Some(Err(e)) => Err((e, 0, 0)),
+ }
}
- }
- MergeSource::Buffered {
- buffer,
- merge_schema,
- state,
- } => {
- // Vectorized merge path: pull the next merged base
- // batch, or drain log-only inserts, then hand it to the shared
- // timing / projection / exhaustion tail below (Step 2) so the
- // stream-stats timing is preserved. Loop to skip base
- // chunks fully eliminated by log deletes.
- let merge_start = Instant::now();
- let mut chunk_err: Option<CoreError> = None;
- let mut out_batch: Option<RecordBatch> = None;
- loop {
- match state {
- BufferedState::BaseScanning => {
- match buffer.next_merged_base_batch(merge_schema) {
- Ok(Some(b)) => {
- if b.num_rows() == 0 {
- // All base rows in this source batch
lost
- // to a log delete — pull the next.
+ MergeSource::Buffered {
+ buffer,
+ base_source,
+ merge_schema,
+ state,
+ } => {
+ // Vectorized merge path: pull the next base batch and
merge
+ // it, or drain log-only inserts, then hand the result to
the
+ // shared timing / projection / exhaustion tail below
+ // (Step 2). Loop to skip base batches fully eliminated by
+ // log deletes.
+ //
+ // Only the merge itself is timed. Timing the whole loop
+ // would fold the base file's read latency into
+ // `final_merge_ms`, which is meant to measure merging.
+ let mut merge_ms = 0u64;
+ let mut chunk_err: Option<CoreError> = None;
+ let mut out_batch: Option<RecordBatch> = None;
+ loop {
+ match state {
+ BufferedState::BaseScanning => {
+ // Pull first, merge second. Only the source
knows
+ // when the base is exhausted, so that is the
only
+ // thing that moves the state machine on to the
+ // drain — a merge that yields no rows just
means
+ // this batch contributed none, and must not be
+ // read as the end of the base file.
+ let pulled = match base_source.as_mut() {
+ None => None,
+ Some(source) => match source.next().await {
+ None => {
+ *base_source = None;
+ None
+ }
+ Some(b) => Some(b),
+ },
+ };
+ let base = match pulled {
+ None => {
+ // Base exhausted — drain log-only
inserts
+ // on the next loop turn.
+ *state =
BufferedState::DrainingLogInserts;
continue;
}
- out_batch = Some(b);
- break;
- }
- Ok(None) => {
- // Base source exhausted — drain log-only
- // inserts on the next loop turn.
- *state = BufferedState::DrainingLogInserts;
- continue;
+ Some(Ok(b)) => b,
+ Some(Err(e)) => {
+ // Drop the source: a failed read must
not
+ // be resumed as if it had merely
ended,
+ // which would truncate the output
+ // silently.
+ *base_source = None;
+ log::error!("[FileGroupMergeStream]
base file source: {e}");
+ chunk_err =
Some(CoreError::ReadFileSliceError(format!(
+ "base file source error: {e}"
Review Comment:
non-blocking: The Buffered path stringifies a base-source error into
`ReadFileSliceError` while the Eager path forwards the original `CoreError`, so
the same storage failure surfaces as a different variant depending on whether
the slice had log files. Forwarding `e` unchanged here (and in
`pull_and_merge_next_base_batch`) would keep the taxonomy uniform and preserve
the source chain for callers that classify retryable errors.
--
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]
