Function calculate_range

datafusion/datasource/src/mod.rs:418–462

Calculates an appropriate byte range for reading from an object based on the provided metadata. This asynchronous function examines the [`PartitionedFile`] of an object in an object store and determines the range of bytes to be read. The range calculation may adjust the start and end points to align with meaningful data boundaries (like newlines). Returns a `Result` wrapping a [`RangeCalculation

(
    file: &PartitionedFile,
    store: &Arc<dyn ObjectStore>,
    terminator: Option<u8>,
)

///
/// Returns an `Error` if any part of the range calculation fails, such as issues in reading from the object store or invalid range boundaries.
pub async fn calculate_range(
    file: &PartitionedFile,
    store: &Arc<dyn ObjectStore>,
    terminator: Option<u8>,
) -> Result<RangeCalculation> {
    let location = &file.object_meta.location;
    let file_size = file.object_meta.size;
    let newline = terminator.unwrap_or(b'\n');

    match file.range {
        None => Ok(RangeCalculation::Range(None)),
        Some(FileRange { start, end }) => {
            let start: u64 = start.try_into().map_err(|_| {
                exec_datafusion_err!("Expect start range to fit in u64, got {start}")
            })?;
            let end: u64 = end.try_into().map_err(|_| {
                exec_datafusion_err!("Expect end range to fit in u64, got {end}")
            })?;

            let start_delta = if start != 0 {
                find_first_newline(store, location, start - 1, file_size, newline).await?
            } else {
                0
            };

            if start + start_delta > end {
                return Ok(RangeCalculation::TerminateEarly);
            }

            let end_delta = if end != file_size {
                find_first_newline(store, location, end - 1, file_size, newline).await?
            } else {
                0
            };

            let range = start + start_delta..end + end_delta;

            if range.start >= range.end {
                return Ok(RangeCalculation::TerminateEarly);
            }

            Ok(RangeCalculation::Range(Some(range)))
        }
    }
}

/// Asynchronously finds the position of the first newline character in a specified byte range
/// within an object, such as a file, in an object store.