Method create_kv_writer

crates/paimon/src/table/table_write.rs:651–694

Create a key-value writer for PK tables with normal buckets.

(
        &mut self,
        partition_path: String,
        bucket: i32,
        partition_bytes: &[u8],
    )

    /// Create a key-value writer for PK tables with normal buckets.
    async fn create_kv_writer(
        &mut self,
        partition_path: String,
        bucket: i32,
        partition_bytes: &[u8],
    ) -> Result<FileWriter> {
        // Lazily scan partition sequence numbers on first writer creation per partition.
        // Overwrite mode skips this — old data will be replaced, so seq starts at 0.
        if !self.is_overwrite && !self.partition_seq_cache.contains_key(partition_bytes) {
            let bucket_seq =
                Self::scan_partition_sequence_numbers(&self.table, partition_bytes).await?;
            self.partition_seq_cache
                .insert(partition_bytes.to_vec(), bucket_seq);
        }
        let next_seq = self
            .partition_seq_cache
            .get(partition_bytes)
            .and_then(|m| m.get(&bucket))
            .copied()
            .unwrap_or(0);

        Ok(FileWriter::KeyValue(KeyValueFileWriter::new(
            self.table.file_io().clone(),
            KeyValueWriteConfig {
                table_name: self.table.identifier().full_name(),
                table_options: self.table.schema().options().clone(),
                table_location: self.table.location().to_string(),
                partition_path,
                bucket,
                schema_id: self.schema_id,
                file_compression: self.file_compression.clone(),
                file_compression_zstd_level: self.file_compression_zstd_level,
                write_buffer_size: self.write_buffer_size,
                file_format: self.file_format.clone(),
                primary_key_indices: self.primary_key_indices.clone(),
                primary_key_types: self.primary_key_types.clone(),
                sequence_field_indices: self.sequence_field_indices.clone(),
                merge_engine: self.merge_engine,
                deletion_vectors_enabled: CoreOptions::new(self.table.schema().options())
                    .deletion_vectors_enabled(),
            },
            next_seq,
        )?))
    }
}

#[cfg(test)]