Method read_batch_stream

crates/paimon/src/arrow/format/parquet.rs:138–200

(
        &self,
        reader: Box<dyn FileRead>,
        file_size: u64,
        read_fields: &[DataField],
        predicates: Option<&FilePredicates>,
        batch_size: Option<usize>,
        r

#[async_trait]
impl FormatFileReader for ParquetFormatReader {
    async fn read_batch_stream(
        &self,
        reader: Box<dyn FileRead>,
        file_size: u64,
        read_fields: &[DataField],
        predicates: Option<&FilePredicates>,
        batch_size: Option<usize>,
        row_selection: Option<Vec<RowRange>>,
    ) -> crate::Result<ArrowRecordBatchStream> {
        let arrow_file_reader = ArrowFileReader::new(file_size, reader);

        let mut batch_stream_builder =
            ParquetRecordBatchStreamBuilder::new(arrow_file_reader).await?;

        let parquet_schema = batch_stream_builder.parquet_schema().clone();
        let root_schema = parquet_schema.root_schema();
        let root_indices: Vec<usize> = read_fields
            .iter()
            .filter_map(|f| {
                root_schema
                    .get_fields()
                    .iter()
                    .position(|pf| pf.name() == f.name())
            })
            .collect();

        let mask = ProjectionMask::roots(&parquet_schema, root_indices);
        batch_stream_builder = batch_stream_builder.with_projection(mask);

        let empty_predicates = Vec::new();
        let (preds, file_fields): (&[Predicate], &[DataField]) = match predicates {
            Some(fp) => (&fp.predicates, &fp.file_fields),
            None => (&empty_predicates, &[]),
        };

        let parquet_row_filter = build_parquet_row_filter(&parquet_schema, preds, file_fields)?;
        if let Some(f) = parquet_row_filter {
            batch_stream_builder = batch_stream_builder.with_row_filter(f);
        }

        let predicate_row_selection = build_predicate_row_selection(
            batch_stream_builder.metadata().row_groups(),
            preds,
            file_fields,
        )?;
        let mut combined_selection = predicate_row_selection;

        if let Some(ref ranges) = row_selection {
            let range_selection =
                build_row_ranges_selection(batch_stream_builder.metadata().row_groups(), ranges);
            combined_selection =
                intersect_optional_row_selections(combined_selection, Some(range_selection));
        }
        if let Some(sel) = combined_selection {
            batch_stream_builder = batch_stream_builder.with_row_selection(sel);
        }
        if let Some(size) = batch_size {
            batch_stream_builder = batch_stream_builder.with_batch_size(size);