Method new

datafusion/datasource-json/src/boundary_stream.rs:151–208

Open a ranged byte stream from `store` and return a ready-to-poll `AlignedBoundaryStream`. Issues a single bounded `get_opts` call covering `[fetch_start, raw_end + END_SCAN_LOOKAHEAD)`. If the terminating newline is not found within that window, `ScanningLastTerminator` automatically issues additional `END_SCAN_LOOKAHEAD`-sized GETs via `store` until the newline is found or EOF is reached.

(
        store: Arc<dyn ObjectStore>,
        location: object_store::path::Path,
        raw_start: u64,
        raw_end: u64,
        file_size: u64,
        terminator: u8,
    )

    /// automatically issues additional `END_SCAN_LOOKAHEAD`-sized GETs
    /// via `store` until the newline is found or EOF is reached.
    pub async fn new(
        store: Arc<dyn ObjectStore>,
        location: object_store::path::Path,
        raw_start: u64,
        raw_end: u64,
        file_size: u64,
        terminator: u8,
    ) -> object_store::Result<Self> {
        if raw_start >= raw_end || raw_start >= file_size {
            return Ok(Self {
                inner: futures::stream::empty().boxed(),
                terminator,
                end: 0,
                bytes_consumed: 0,
                fetch_start: 0,
                phase: Phase::Done,
                pending: None,
                store,
                location,
                file_size,
            });
        }

        let (fetch_start, phase) = if raw_start == 0 {
            (0, Phase::FetchingChunks)
        } else {
            (raw_start - 1, Phase::ScanningFirstTerminator)
        };

        let initial_fetch_end = raw_end.saturating_add(END_SCAN_LOOKAHEAD).min(file_size);

        let inner = get_stream(
            Arc::clone(&store),
            location.clone(),
            fetch_start..initial_fetch_end,
        )
        .await?;

        // Last partition reads until EOF is reached — no end-boundary scanning needed.
        let end = if raw_end >= file_size {
            u64::MAX
        } else {
            raw_end
        };

        Ok(Self {
            inner,
            terminator,
            end,
            bytes_consumed: 0,
            fetch_start,
            phase,
            pending: None,
            store,
            location,
            file_size,
        })
    }