Digest this code:


pub fn index_from_path(path: &str) -> io::Result<csi::Index> {
    let bai_path = format!("{}.bai", path);
    let csi_path = format!("{}.csi", path);
    let index = if Path::new(&bai_path).exists() {
        bam::bai::read(bai_path)?
    } else if Path::new(&csi_path).exists() {
        csi::read(csi_path)?
    } else {
        panic!("Could not find a .bai or .csi index file for the given BAM file.");
    };
    Ok(index)
}

/// A BAM reader.
pub struct BamReader<R> {
    reader: bam::Reader<bgzf::Reader<R>>,
    header: sam::Header,
    index: csi::Index,
}

impl BamReader<BufReader<File>> {
    /// Creates a BAM reader from a given file path.
    pub fn new_from_path(path: &str) -> std::io::Result<Self> {
        let index = index_from_path(path)?;
        let file = std::fs::File::open(path)?;
        let buf_file = std::io::BufReader::with_capacity(1024 * 1024, file);
        let mut reader = bam::Reader::new(buf_file);
        let header = reader.read_header()?;
        Ok(Self {
            reader,
            header,
            index,
        })
    }
}

and the check this one:

impl<R: Read + Seek> BamReader<R> {
    /// Creates a BAM reader.
    pub fn new(read: R, index: csi::Index) -> std::io::Result<Self> {
        let mut reader = bam::Reader::new(read);
        let header = reader.read_header()?;
        Ok(Self {
            reader,
            header,
            index,
        })
    }
    pub fn records_to_record_batch(
        &mut self,
        region: Option<&str>,
    ) -> Result<
        arrow::record_batch::RecordBatchIterator<
            impl Iterator<Item = Result<RecordBatch, arrow::error::ArrowError>> + 'static,
        >,
        ArrowError,
    > {
        let batch_builder = BamBatchBuilder::new(1024, &self.header)?;
        if let Some(region) = region {
            let region: Region = region.parse().unwrap();
            let query = self
                .reader
                .query(&self.header, &self.index, &region)
                .map_err(|e| ArrowError::ExternalError(e.into()))?
                .map(|i| i.map_err(|e| ArrowError::ExternalError(e.into())));

            let batches = get_record_batches(query, batch_builder)?;
            let schema = batches.schema();

            let reader =
                RecordBatchIterator::new(vec![batches].into_iter().map(Ok), schema.clone());

            return Ok(reader);
        }
        let records = self
            .reader
            .records(&self.header)
            .map(|i| i.map_err(|e| ArrowError::ExternalError(e.into())));

        let batches = get_record_batches(records, batch_builder)?;
        let schema = batches.schema();

        let reader = RecordBatchIterator::new(vec![batches].into_iter().map(Ok), schema.clone());

        Ok(reader)
    }
where get_record_batches is defined in another file called batch_builder.rs:

pub trait BatchBuilder {
    type Record<'a>;
    fn push(&mut self, record: Self::Record<'_>);
    fn finish(self) -> Result<RecordBatch, ArrowError>;
}

pub fn get_record_batches<T>(
    records: impl Iterator<Item = Result<T, ArrowError>>,
    mut batch_builder: impl for<'a> BatchBuilder<Record<'a> = &'a T>,
) -> Result<RecordBatch, ArrowError> {
    for record in records {
        let record = record?;
        batch_builder.push(&record)
    }

    Ok(batch_builder.finish()?)
}

Ok, after digesting and understanding deeply all that code, I need you to help me achieve something. I have this idea in mind to instead of doing this:

let reader =
                RecordBatchIterator::new(vec![batches].into_iter().map(Ok), schema.clone());


where I force batches to be an iterator (right now batches only holds 1 RecordBatch because the reader reads the whole BAM file once), to have an iterator of RecordBatch. In order to do that, I want to read the file in chunks of a given size. Could you guide please? Be detailed and ensure that everything is correct.