Method prepare_commit

crates/paimon/src/table/cow_writer.rs:208–342

(self)

    /// Rewrite affected files and produce CommitMessages.
    #[must_use = "commit messages must be passed to TableCommit"]
    pub async fn prepare_commit(self) -> Result<Vec<CommitMessage>> {
        if self.affected_files.is_empty() {
            return Ok(Vec::new());
        }

        let schema = self.table.schema();
        let core_options = CoreOptions::new(schema.options());
        let partition_keys: Vec<String> = schema.partition_keys().to_vec();
        let partition_computer = PartitionComputer::new(
            &partition_keys,
            schema.fields(),
            core_options.partition_default_name(),
            core_options.legacy_partition_name(),
        )?;

        let target_file_size = core_options.target_file_size();
        let file_compression = core_options.file_compression().to_string();
        let file_compression_zstd_level = core_options.file_compression_zstd_level();
        let write_buffer_size = core_options.write_parquet_buffer_size();
        let file_format = core_options.file_format().to_string();
        let schema_id = schema.id();

        let update_columns = &self.update_columns;
        let update_batches = &self.update_batches;
        let file_index = &self.file_index;
        let table = &self.table;
        let partition_keys = &partition_keys;
        let partition_computer = &partition_computer;
        let file_compression = file_compression.as_str();
        let file_format = file_format.as_str();

        // Process each affected file in parallel
        let rewrite_futures: Vec<_> = self
            .affected_files
            .iter()
            .map(|(&file_idx, operations)| async move {
                let file_info = &file_index[file_idx];

                // Read the entire file
                let single_split = DataSplitBuilder::new()
                    .with_snapshot(file_info.snapshot_id)
                    .with_partition(BinaryRow::from_serialized_bytes(&file_info.partition)?)
                    .with_bucket(file_info.bucket)
                    .with_bucket_path(file_info.bucket_path.clone())
                    .with_total_buckets(file_info.total_buckets)
                    .with_data_files(vec![file_info.file_meta.clone()])
                    .build()?;

                let read = table.new_read_builder().new_read()?;
                let original_batches: Vec<RecordBatch> =
                    read.to_arrow(&[single_split])?.try_collect().await?;

                if original_batches.is_empty() {
                    return Ok::<_, crate::Error>(None);
                }

                let original = if original_batches.len() == 1 {
                    original_batches.into_iter().next().unwrap()