Method merge_files_by_columns

crates/paimon/src/table/data_evolution_reader.rs:242–440

Merge multiple logical sources column-wise for data evolution. Normal partial-column files remain one source per file. Rolling `.blob` files are first grouped into a logical BlobBunch source per field, then source streams are merged by projected field position.

(
        &self,
        split: &DataSplit,
        prepared_group: &PreparedMergeGroup,
        row_ranges: Option<Vec<RowRange>>,
        expected_output_rows: usize,
    )

    /// files are first grouped into a logical BlobBunch source per field, then
    /// source streams are merged by projected field position.
    fn merge_files_by_columns(
        &self,
        split: &DataSplit,
        prepared_group: &PreparedMergeGroup,
        row_ranges: Option<Vec<RowRange>>,
        expected_output_rows: usize,
    ) -> crate::Result<ArrowRecordBatchStream> {
        if prepared_group.files.is_empty() {
            return Ok(futures::stream::empty().boxed());
        }

        let file_io = self.file_io.clone();
        let schema_manager = self.schema_manager.clone();
        let table_schema_id = self.table_schema_id;
        let split = split.clone();
        let prepared_group = prepared_group.clone();
        let read_type = self.file_read_type.clone();
        let table_fields = self.table_fields.clone();
        let blob_descriptor_fields = self.blob_descriptor_fields.clone();
        let blob_as_descriptor = self.blob_as_descriptor;
        // Batch size for column-merge output. Matches the default Parquet reader batch size.
        const MERGE_BATCH_SIZE: usize = 1024;
        let target_schema = build_target_arrow_schema(&read_type)?;

        Ok(try_stream! {
            let file_infos = load_file_infos(
                &schema_manager,
                table_schema_id,
                &table_fields,
                &prepared_group.files,
            )
            .await?;
            let source_plan = build_source_plan(&prepared_group, &file_infos, &read_type, &blob_descriptor_fields)?;

            let active_source_indices: Vec<usize> = source_plan
                .sources
                .iter()
                .enumerate()
                .filter_map(|(idx, source)| (!source.read_fields().is_empty()).then_some(idx))
                .collect();

            // Edge case: no file provides any projected column.
            if active_source_indices.is_empty() {
                let mut emitted = 0usize;
                while emitted < expected_output_rows {
                    let rows_to_emit = (expected_output_rows - emitted).min(MERGE_BATCH_SIZE);
                    let columns: Vec<Arc<dyn arrow_array::Array>> = target_schema
                        .fields()
                        .iter()
                        .map(|f| arrow_array::new_null_array(f.data_type(), rows_to_emit))
                        .collect();
                    let batch = if columns.is_empty() {
                        RecordBatch::try_new_with_options(
                            target_schema.clone(),
                            columns,
                            &arrow_array::RecordBatchOptions::new().with_row_count(Some(rows_to_emit)),
                        )
                    } else {