yihua commented on code in PR #689:
URL: https://github.com/apache/hudi-rs/pull/689#discussion_r3910930893
##########
crates/core/src/storage/reader.rs:
##########
@@ -240,62 +223,113 @@ impl StorageReader {
LogBlockFetcher::new(self.object_store.clone(), self.location.clone())
}
- /// Refill the window if the cursor has moved outside it. No-op past the
end.
- fn ensure_window(&mut self) -> Result<()> {
- if self.pos >= self.file_len {
- return Ok(());
- }
- let in_window = self.pos >= self.window_start
- && self.pos < self.window_start + self.window.len() as u64;
- if in_window {
- return Ok(());
- }
- let start = self.pos;
- let end = (start + self.window_size).min(self.file_len);
- self.window = get_range_blocking(&self.object_store, &self.location,
start, end - start)?;
- self.window_start = start;
+ /// Length of the whole file, known without reading any of it.
+ pub fn file_len(&self) -> u64 {
+ self.file_len
+ }
+
+ /// Where the cursor sits.
+ pub fn position(&self) -> u64 {
+ self.pos
+ }
+
+ /// Move the cursor. No I/O, and a position past the end of the file is
+ /// allowed: it fails at the next read, which is what lets a caller skip
past
+ /// a block's content without paying for it.
+ pub fn seek_to(&mut self, pos: u64) {
+ self.pos = pos;
+ }
+
+ /// Whether `[start, end)` lies inside the resident window.
+ fn window_covers(&self, start: u64, end: u64) -> bool {
+ start >= self.window_start && end <= self.window_start +
self.window.len() as u64
+ }
+
+ /// Fetch a window starting at the cursor.
+ async fn fill_window(&mut self) -> Result<()> {
Review Comment:
non-blocking: Two fetch patterns here look worth a measurement before the
merge-down: on a file whose blocks are smaller than the window, the walk's
refills sweep essentially the whole file and Pass 3 then re-fetches the
admitted content, so total bytes read are ~2x the old whole-file GET; and for a
block larger than the window, `is_block_corrupted`'s trailing-pointer probe
evicts and re-fetches the header-side window (seek to trailing → refill → seek
back → refill again), costing ~2 extra windows per large block. Have you
considered a bytes-fetched counter in the object-storage bench you already
plan, or restoring/keeping the walk window across the corruption probe? The
memory win is measured and real — this is only about knowing what it costs in
bandwidth where round trips were the justification.
##########
crates/core/src/file_group/reader_v2/merge_iterator.rs:
##########
@@ -487,22 +554,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 +397,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]
