Method read_records

parquet/src/arrow/array_reader/cached_array_reader.rs:197–257

(&mut self, num_records: usize)

    }

    fn read_records(&mut self, num_records: usize) -> Result<usize> {
        let mut read = 0;
        while read < num_records {
            let batch_id = self.get_batch_id_from_position(self.outer_position);

            // Check local cache first
            let cached = if let Some(array) = self.local_cache.get(&batch_id) {
                Some(Arc::clone(array))
            } else {
                // If not in local cache, i.e., we are consumer, check shared cache
                let cache_content = self
                    .shared_cache
                    .read()
                    .unwrap()
                    .get(self.column_idx, batch_id);
                if let Some(array) = cache_content.as_ref() {
                    // Store in local cache for later use in consume_batch
                    self.local_cache.insert(batch_id, Arc::clone(array));
                }
                cache_content
            };

            match cached {
                Some(array) => {
                    let array_len = array.len();
                    if array_len + batch_id.val * self.batch_size > self.outer_position {
                        // the cache batch has some records that we can select
                        let v = array_len + batch_id.val * self.batch_size - self.outer_position;
                        let select_cnt = std::cmp::min(num_records - read, v);
                        read += select_cnt;
                        self.metrics.increment_cache_reads(select_cnt);
                        self.outer_position += select_cnt;
                        self.selections.append_n(select_cnt, true);
                    } else {
                        // this is last batch and we have used all records from it
                        break;
                    }
                }
                None => {
                    let read_from_inner = self.fetch_batch(batch_id)?;
                    // Reached end-of-file, no more records to read
                    if read_from_inner == 0 {
                        break;
                    }
                    self.metrics.increment_inner_reads(read_from_inner);
                    let select_from_this_batch = std::cmp::min(
                        num_records - read,
                        self.inner_position - self.outer_position,
                    );
                    read += select_from_this_batch;
                    self.outer_position += select_from_this_batch;
                    self.selections.append_n(select_from_this_batch, true);
                    if read_from_inner < self.batch_size {
                        // this is last batch from inner reader
                        break;
                    }
                }
            }