| 217 | type Item = object_store::Result<Bytes>; |
| 218 | |
| 219 | fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { |
| 220 | let this = self.get_mut(); |
| 221 | |
| 222 | loop { |
| 223 | match this.phase { |
| 224 | Phase::Done => return Poll::Ready(None), |
| 225 | |
| 226 | Phase::ScanningFirstTerminator => { |
| 227 | // Find the first terminator and skip everything up to |
| 228 | // and including it. Store any remainder in `pending` |
| 229 | // so `FetchingChunks` can apply end-boundary logic to it. |
| 230 | match this.inner.poll_next_unpin(cx) { |
| 231 | Poll::Pending => return Poll::Pending, |
| 232 | Poll::Ready(None) => { |
| 233 | this.phase = Phase::Done; |
| 234 | return Poll::Ready(None); |
| 235 | } |
| 236 | Poll::Ready(Some(Err(e))) => { |
| 237 | this.phase = Phase::Done; |
| 238 | return Poll::Ready(Some(Err(e))); |
| 239 | } |
| 240 | Poll::Ready(Some(Ok(chunk))) => { |
| 241 | this.bytes_consumed += chunk.len() as u64; |
| 242 | match chunk.iter().position(|&b| b == this.terminator) { |
| 243 | Some(pos) => { |
| 244 | let remainder = chunk.slice((pos + 1)..); |
| 245 | // The aligned start position is where |
| 246 | // data begins after the newline. |
| 247 | let aligned_start = |
| 248 | this.abs_pos() - remainder.len() as u64; |
| 249 | if aligned_start >= this.end { |
| 250 | // Start alignment landed at or past |
| 251 | // the end boundary — no complete |
| 252 | // lines in this partition's range. |
| 253 | this.phase = Phase::Done; |
| 254 | return Poll::Ready(None); |
| 255 | } |
| 256 | if !remainder.is_empty() { |
| 257 | this.pending = Some(remainder); |
| 258 | } |
| 259 | this.phase = Phase::FetchingChunks; |
| 260 | continue; |
| 261 | } |
| 262 | None => continue, |
| 263 | } |
| 264 | } |
| 265 | } |
| 266 | } |
| 267 | |
| 268 | Phase::FetchingChunks => { |
| 269 | // Get the next chunk: pending remainder or inner stream. |
| 270 | let chunk = if let Some(pending) = this.pending.take() { |
| 271 | pending |
| 272 | } else { |
| 273 | match this.inner.poll_next_unpin(cx) { |
| 274 | Poll::Pending => return Poll::Pending, |
| 275 | Poll::Ready(None) => { |
| 276 | this.phase = Phase::Done; |