Function open_source_stream

crates/paimon/src/table/data_evolution_reader.rs:480–533

(
    split: &DataSplit,
    source: &FieldSource,
    row_ranges: Option<Vec<RowRange>>,
    file_io: FileIO,
    schema_manager: SchemaManager,
    table_schema_id: i64,
    table_fields: Vec<DataFi

#[allow(clippy::too_many_arguments)]
fn open_source_stream(
    split: &DataSplit,
    source: &FieldSource,
    row_ranges: Option<Vec<RowRange>>,
    file_io: FileIO,
    schema_manager: SchemaManager,
    table_schema_id: i64,
    table_fields: Vec<DataField>,
    blob_as_descriptor: bool,
) -> crate::Result<ArrowRecordBatchStream> {
    let file_reader = DataFileReader::new(
        file_io,
        schema_manager,
        table_schema_id,
        table_fields,
        source.read_fields().to_vec(),
        Vec::new(),
    )
    .with_blob_as_descriptor(blob_as_descriptor);

    match source {
        FieldSource::DataFile {
            file, data_fields, ..
        } => file_reader.read_single_file_stream(
            split,
            file.as_ref().clone(),
            data_fields.clone(),
            None,
            row_ranges,
        ),
        FieldSource::BlobBunch {
            bunch, data_fields, ..
        } => {
            let split = split.clone();
            let files = bunch.files.clone();
            let data_fields = data_fields.clone();
            Ok(try_stream! {
                for file in files {
                    let mut stream = file_reader.read_single_file_stream(
                        &split,
                        file,
                        data_fields.clone(),
                        None,
                        row_ranges.clone(),
                    )?;
                    while let Some(batch) = stream.next().await {
                        yield batch?;
                    }
                }
            }
            .boxed())
        }
    }
}

#[derive(Debug, Clone)]
struct PreparedMergeGroup {