Method read_batch_stream

crates/paimon/src/arrow/format/vortex.rs:104–205

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

#[async_trait]
impl FormatFileReader for VortexFormatReader {
    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 session = VortexSession::default();

        let source = Arc::new(PaimonVortexReadAt {
            file_size,
            reader: Arc::from(reader),
        });

        let vortex_file = session
            .open_options()
            .with_file_size(file_size)
            .open(source)
            .await
            .map_err(|e| Error::DataInvalid {
                message: format!("Failed to open Vortex file: {e}"),
                source: None,
            })?;

        // Build the target Arrow schema for the projected fields.
        let target_schema = crate::arrow::build_target_arrow_schema(read_fields)?;

        if read_fields.is_empty() {
            let row_count = match &row_selection {
                Some(ranges) => ranges.iter().map(|r| r.count() as usize).sum(),
                None => vortex_file.row_count() as usize,
            };
            let batch = RecordBatch::try_new_with_options(
                target_schema,
                vec![],
                &arrow_array::RecordBatchOptions::new().with_row_count(Some(row_count)),
            )
            .map_err(|e| Error::DataInvalid {
                message: format!("Failed to build empty RecordBatch: {e}"),
                source: None,
            })?;
            return Ok(Box::pin(futures::stream::once(async { Ok(batch) })));
        }

        // Build projection expression for requested fields.
        let projected_names: Vec<&str> = read_fields.iter().map(|f| f.name()).collect();

        let mut scan_builder = vortex_file.scan().map_err(|e| Error::DataInvalid {
            message: format!("Failed to create Vortex scan: {e}"),
            source: None,
        })?;

        // Apply column projection.
        {
            use vortex::array::expr::{root, select};
            scan_builder = scan_builder.with_projection(select(projected_names, root()));