Method write

crates/paimon/src/arrow/format/blob.rs:622–714

(&mut self, batch: &RecordBatch)

#[async_trait]
impl FormatFileWriter for BlobFormatWriter {
    async fn write(&mut self, batch: &RecordBatch) -> crate::Result<()> {
        if batch.num_rows() == 0 {
            return Ok(());
        }

        let col = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow_array::BinaryArray>()
            .ok_or_else(|| Error::DataInvalid {
                message: "BlobFormatWriter expects a single Binary column".to_string(),
                source: None,
            })?;

        for row_idx in 0..col.len() {
            if col.is_null(row_idx) {
                self.lengths.push(-1);
                continue;
            }

            let value = col.value(row_idx);

            if BlobDescriptor::is_blob_descriptor(value) {
                let desc = BlobDescriptor::deserialize(value)?;
                let payload_len = desc.length() as u64;
                let entry_length = (payload_len + BLOB_ENTRY_OVERHEAD) as i64;
                self.lengths.push(entry_length);

                let file_io = self.file_io.as_ref().ok_or_else(|| Error::DataInvalid {
                    message:
                        "BlobFormatWriter received a BlobDescriptor but has no FileIO to resolve it"
                            .to_string(),
                    source: None,
                })?;
                let input = file_io.new_input(desc.uri())?;
                let reader = input.reader().await?;

                let mut hasher = crc32fast::Hasher::new();

                hasher.update(&BLOB_MAGIC_NUMBER_BYTES);
                self.writer
                    .write(Bytes::copy_from_slice(&BLOB_MAGIC_NUMBER_BYTES))
                    .await?;

                // Stream payload in chunks to avoid loading entire blob into memory
                let start = desc.offset() as u64;
                let end = start + payload_len;
                let mut pos = start;
                while pos < end {
                    let chunk_end = (pos + BLOB_WRITE_BUFFER_SIZE).min(end);
                    let chunk = reader.read(pos..chunk_end).await?;
                    hasher.update(&chunk);
                    self.writer.write(chunk).await?;
                    pos = chunk_end;
                }

                let entry_length_bytes = entry_length.to_le_bytes();
                hasher.update(&entry_length_bytes);