You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/04/17 13:47:57 UTC
(paimon-rust) branch main updated: feat: support reading blob files in data evolution flow (#259)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 9796ef7 feat: support reading blob files in data evolution flow (#259)
9796ef7 is described below
commit 9796ef764bd94c24e9ceeccd85edd66b0fe5bc86
Author: Zach <45...@users.noreply.github.com>
AuthorDate: Fri Apr 17 21:47:52 2026 +0800
feat: support reading blob files in data evolution flow (#259)
---
crates/blob_test_utils.rs | 97 ++
crates/paimon/src/arrow/format/blob.rs | 839 ++++++++++++
crates/paimon/src/arrow/format/mod.rs | 5 +-
crates/paimon/src/table/data_evolution_reader.rs | 1511 +++++++++++++++++++---
crates/paimon/testdata/blob/blob-basic.blob | Bin 0 -> 67 bytes
crates/paimon/testdata/blob/blob-part-1.blob | Bin 0 -> 28 bytes
crates/paimon/testdata/blob/blob-part-2.blob | Bin 0 -> 44 bytes
7 files changed, 2240 insertions(+), 212 deletions(-)
diff --git a/crates/blob_test_utils.rs b/crates/blob_test_utils.rs
new file mode 100644
index 0000000..24fcae0
--- /dev/null
+++ b/crates/blob_test_utils.rs
@@ -0,0 +1,97 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::fs;
+use std::path::Path;
+
+const BLOB_MAGIC_NUMBER_BYTES: [u8; 4] = 1481511375_i32.to_le_bytes();
+const BLOB_ENTRY_OVERHEAD: usize = 16;
+const BLOB_FORMAT_VERSION: u8 = 1;
+
+pub(crate) fn build_blob_file_bytes(rows: &[Option<&[u8]>]) -> Vec<u8> {
+ let mut file_bytes = Vec::new();
+ let mut lengths = Vec::with_capacity(rows.len());
+
+ for row in rows {
+ match row {
+ Some(payload) => {
+ let entry_length = payload
+ .len()
+ .checked_add(BLOB_ENTRY_OVERHEAD)
+ .and_then(|len| i64::try_from(len).ok())
+ .unwrap_or_else(|| {
+ panic!("Blob payload length {} exceeds test helper limits", payload.len())
+ });
+ lengths.push(entry_length);
+
+ file_bytes.extend_from_slice(&BLOB_MAGIC_NUMBER_BYTES);
+ file_bytes.extend_from_slice(payload);
+ let mut hasher = crc32fast::Hasher::new();
+ hasher.update(&BLOB_MAGIC_NUMBER_BYTES);
+ hasher.update(payload);
+
+ let entry_length_bytes = entry_length.to_le_bytes();
+ file_bytes.extend_from_slice(&entry_length_bytes);
+ hasher.update(&entry_length_bytes);
+ file_bytes.extend_from_slice(&hasher.finalize().to_le_bytes());
+ }
+ None => lengths.push(-1),
+ }
+ }
+
+ let index_bytes = encode_delta_varints(&lengths);
+ let index_length = i32::try_from(index_bytes.len()).unwrap_or_else(|_| {
+ panic!(
+ "Blob index length {} exceeds test helper limits",
+ index_bytes.len()
+ )
+ });
+ file_bytes.extend_from_slice(&index_bytes);
+ file_bytes.extend_from_slice(&index_length.to_le_bytes());
+ file_bytes.push(BLOB_FORMAT_VERSION);
+ file_bytes
+}
+
+pub(crate) fn write_blob_file(path: &Path, rows: &[Option<&[u8]>]) {
+ let file_bytes = build_blob_file_bytes(rows);
+ fs::write(path, file_bytes)
+ .unwrap_or_else(|e| panic!("Failed to write blob test file {path:?}: {e}"));
+}
+
+pub(crate) fn encode_delta_varints(values: &[i64]) -> Vec<u8> {
+ if values.is_empty() {
+ return Vec::new();
+ }
+
+ let mut encoded = Vec::new();
+ let mut previous = 0_i64;
+ for (idx, value) in values.iter().copied().enumerate() {
+ let delta = if idx == 0 { value } else { value - previous };
+ previous = value;
+ encode_varint(delta, &mut encoded);
+ }
+ encoded
+}
+
+fn encode_varint(value: i64, out: &mut Vec<u8>) {
+ let mut remaining = ((value << 1) ^ (value >> 63)) as u64;
+ while (remaining & !0x7f) != 0 {
+ out.push(((remaining & 0x7f) as u8) | 0x80);
+ remaining >>= 7;
+ }
+ out.push(remaining as u8);
+}
diff --git a/crates/paimon/src/arrow/format/blob.rs b/crates/paimon/src/arrow/format/blob.rs
new file mode 100644
index 0000000..d1bb5dd
--- /dev/null
+++ b/crates/paimon/src/arrow/format/blob.rs
@@ -0,0 +1,839 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::{FilePredicates, FormatFileReader};
+use crate::arrow::build_target_arrow_schema;
+use crate::io::FileRead;
+use crate::spec::{DataField, DataType};
+use crate::table::{ArrowRecordBatchStream, RowRange};
+use crate::Error;
+use arrow_array::builder::BinaryBuilder;
+use arrow_array::{ArrayRef, RecordBatch, RecordBatchOptions};
+use async_stream::try_stream;
+use async_trait::async_trait;
+use bytes::Bytes;
+use futures::{StreamExt, TryStreamExt};
+use std::ops::Range;
+use std::sync::Arc;
+
+pub(crate) struct BlobFormatReader;
+
+const BLOB_FOOTER_SIZE: u64 = 5;
+const BLOB_FORMAT_VERSION: u8 = 1;
+const BLOB_INLINE_HEADER_SIZE: u64 = 4;
+const BLOB_TRAILER_SIZE: u64 = 12;
+const BLOB_ENTRY_OVERHEAD: u64 = BLOB_INLINE_HEADER_SIZE + BLOB_TRAILER_SIZE;
+const DEFAULT_BATCH_SIZE: usize = 128;
+const BLOB_READ_CONCURRENCY: usize = 8;
+
+#[async_trait]
+impl FormatFileReader for BlobFormatReader {
+ async fn read_batch_stream(
+ &self,
+ reader: Box<dyn FileRead>,
+ file_size: u64,
+ read_fields: &[DataField],
+ _predicates: Option<&FilePredicates>,
+ batch_size: Option<usize>,
+ row_selection: Option<Vec<RowRange>>,
+ ) -> crate::Result<ArrowRecordBatchStream> {
+ validate_read_fields(read_fields)?;
+
+ let target_schema = build_target_arrow_schema(read_fields)?;
+ let batch_size = batch_size.unwrap_or(DEFAULT_BATCH_SIZE);
+ let blob_index = BlobFileIndex::load(reader.as_ref(), file_size).await?;
+ let mut selection = RowSelectionCursor::new(blob_index.num_rows(), row_selection)?;
+ let project_values = !read_fields.is_empty();
+
+ Ok(try_stream! {
+ while let Some(positions) = selection.next_batch(batch_size) {
+ let batch = read_blob_batch(
+ reader.as_ref(),
+ &blob_index,
+ &target_schema,
+ &positions,
+ project_values,
+ ).await?;
+ yield batch;
+ }
+ }
+ .boxed())
+ }
+}
+
+fn validate_read_fields(read_fields: &[DataField]) -> crate::Result<()> {
+ if read_fields.len() > 1 {
+ return Err(Error::DataInvalid {
+ message: format!(
+ ".blob format only supports reading at most one projected column, got {}",
+ read_fields.len()
+ ),
+ source: None,
+ });
+ }
+
+ if let Some(field) = read_fields.first() {
+ match field.data_type() {
+ DataType::Blob(_) => Ok(()),
+ other => Err(Error::DataInvalid {
+ message: format!(
+ ".blob format requires a Blob field, got {:?} for column '{}'",
+ other,
+ field.name()
+ ),
+ source: None,
+ }),
+ }?;
+ }
+
+ Ok(())
+}
+
+async fn read_blob_batch(
+ reader: &dyn FileRead,
+ blob_index: &BlobFileIndex,
+ target_schema: &Arc<arrow_schema::Schema>,
+ positions: &[usize],
+ project_values: bool,
+) -> crate::Result<RecordBatch> {
+ if !project_values {
+ return RecordBatch::try_new_with_options(
+ target_schema.clone(),
+ Vec::new(),
+ &RecordBatchOptions::new().with_row_count(Some(positions.len())),
+ )
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build empty blob RecordBatch: {e}"),
+ source: Some(Box::new(e)),
+ });
+ }
+
+ let planned_reads = plan_blob_reads(blob_index, positions)?;
+ let values = fetch_blob_values(reader, planned_reads).await?;
+ let mut builder = BinaryBuilder::new();
+ for value in values {
+ match value {
+ BlobValue::Null => builder.append_null(),
+ BlobValue::Inline(bytes) => builder.append_value(bytes.as_ref()),
+ }
+ }
+
+ let columns: Vec<ArrayRef> = vec![Arc::new(builder.finish())];
+ RecordBatch::try_new(target_schema.clone(), columns).map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build blob RecordBatch: {e}"),
+ source: Some(Box::new(e)),
+ })
+}
+
+fn plan_blob_reads(
+ blob_index: &BlobFileIndex,
+ positions: &[usize],
+) -> crate::Result<Vec<PlannedBlobRead>> {
+ positions
+ .iter()
+ .map(|&position| {
+ let entry = blob_index
+ .entry(position)
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "Blob row selection referenced out-of-range position {position} for {} rows",
+ blob_index.num_rows()
+ ),
+ source: None,
+ })?;
+
+ Ok(match entry.inline_data_range() {
+ Some(range) if range.start == range.end => PlannedBlobRead::Empty,
+ Some(range) => PlannedBlobRead::Read(range),
+ None => PlannedBlobRead::Null,
+ })
+ })
+ .collect()
+}
+
+async fn fetch_blob_values(
+ reader: &dyn FileRead,
+ planned_reads: Vec<PlannedBlobRead>,
+) -> crate::Result<Vec<BlobValue>> {
+ futures::stream::iter(planned_reads.into_iter().map(|planned_read| async move {
+ match planned_read {
+ PlannedBlobRead::Null => Ok(BlobValue::Null),
+ PlannedBlobRead::Empty => Ok(BlobValue::Inline(Bytes::new())),
+ PlannedBlobRead::Read(range) => reader.read(range).await.map(BlobValue::Inline),
+ }
+ }))
+ .buffered(BLOB_READ_CONCURRENCY)
+ .try_collect()
+ .await
+}
+
+#[derive(Debug, Clone)]
+enum PlannedBlobRead {
+ Null,
+ Empty,
+ Read(Range<u64>),
+}
+
+#[derive(Debug, Clone)]
+enum BlobValue {
+ Null,
+ Inline(Bytes),
+}
+
+#[derive(Debug, Clone)]
+struct BlobFileIndex {
+ entries: Vec<BlobEntry>,
+}
+
+impl BlobFileIndex {
+ async fn load(reader: &dyn FileRead, file_size: u64) -> crate::Result<Self> {
+ if file_size < BLOB_FOOTER_SIZE {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob file is too small: expected at least {BLOB_FOOTER_SIZE} bytes, got {file_size}"
+ ),
+ source: None,
+ });
+ }
+
+ let footer = reader
+ .read(file_size - BLOB_FOOTER_SIZE..file_size)
+ .await
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to read blob footer: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+
+ let footer_bytes: [u8; BLOB_FOOTER_SIZE as usize] =
+ footer.as_ref().try_into().map_err(|_| Error::DataInvalid {
+ message: "Blob footer should be exactly 5 bytes".to_string(),
+ source: None,
+ })?;
+ let index_length = i32::from_le_bytes(footer_bytes[..4].try_into().unwrap());
+ if index_length < 0 {
+ return Err(Error::DataInvalid {
+ message: format!("Blob footer contains a negative index length: {index_length}"),
+ source: None,
+ });
+ }
+ if footer_bytes[4] != BLOB_FORMAT_VERSION {
+ return Err(Error::Unsupported {
+ message: format!(
+ "unsupported .blob footer version: expected {BLOB_FORMAT_VERSION}, got {}",
+ footer_bytes[4]
+ ),
+ });
+ }
+
+ let index_length = index_length as u64;
+ if index_length > file_size - BLOB_FOOTER_SIZE {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob footer index length {index_length} exceeds file payload size {}",
+ file_size - BLOB_FOOTER_SIZE
+ ),
+ source: None,
+ });
+ }
+
+ let index_start = file_size - BLOB_FOOTER_SIZE - index_length;
+ let data_region_end = index_start;
+ let index_bytes = reader
+ .read(index_start..index_start + index_length)
+ .await
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to read blob index bytes: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+
+ let lengths = decode_delta_varints(index_bytes.as_ref())?;
+ let entries = BlobEntry::build_all(&lengths, data_region_end)?;
+ Ok(Self { entries })
+ }
+
+ fn num_rows(&self) -> usize {
+ self.entries.len()
+ }
+
+ fn entry(&self, position: usize) -> Option<&BlobEntry> {
+ self.entries.get(position)
+ }
+}
+
+#[derive(Debug, Clone)]
+struct BlobEntry {
+ data_offset: Option<u64>,
+ data_length: u64,
+}
+
+impl BlobEntry {
+ fn build_all(lengths: &[i64], data_region_end: u64) -> crate::Result<Vec<Self>> {
+ let mut entries = Vec::with_capacity(lengths.len());
+ let mut next_offset = 0_u64;
+
+ for &entry_length in lengths {
+ if entry_length == -1 {
+ entries.push(Self {
+ data_offset: None,
+ data_length: 0,
+ });
+ continue;
+ }
+
+ let entry_length = u64::try_from(entry_length).map_err(|e| Error::DataInvalid {
+ message: format!("Blob entry length must be positive or -1, got {entry_length}"),
+ source: Some(Box::new(e)),
+ })?;
+
+ if entry_length < BLOB_ENTRY_OVERHEAD {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob entry length {entry_length} is smaller than minimum overhead {BLOB_ENTRY_OVERHEAD}"
+ ),
+ source: None,
+ });
+ }
+
+ let entry_end =
+ next_offset
+ .checked_add(entry_length)
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!("Blob entry length overflow at offset {next_offset}"),
+ source: None,
+ })?;
+ if entry_end > data_region_end {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob entry range [{next_offset}, {entry_end}) exceeds data region end {data_region_end}"
+ ),
+ source: None,
+ });
+ }
+
+ entries.push(Self {
+ data_offset: Some(next_offset + BLOB_INLINE_HEADER_SIZE),
+ data_length: entry_length - BLOB_ENTRY_OVERHEAD,
+ });
+ next_offset = entry_end;
+ }
+
+ Ok(entries)
+ }
+
+ fn inline_data_range(&self) -> Option<Range<u64>> {
+ self.data_offset
+ .map(|offset| offset..offset + self.data_length)
+ }
+}
+
+#[derive(Debug, Clone)]
+struct RowSelectionCursor {
+ state: RowSelectionState,
+}
+
+#[derive(Debug, Clone)]
+enum RowSelectionState {
+ All {
+ next: usize,
+ total_rows: usize,
+ },
+ Ranges {
+ total_rows: usize,
+ ranges: Vec<RowRange>,
+ range_idx: usize,
+ next_in_range: i64,
+ },
+}
+
+impl RowSelectionCursor {
+ fn new(total_rows: usize, row_selection: Option<Vec<RowRange>>) -> crate::Result<Self> {
+ let state = match row_selection {
+ None => RowSelectionState::All {
+ next: 0,
+ total_rows,
+ },
+ Some(ranges) => {
+ for range in &ranges {
+ if range.from() < 0 {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob row selection must be non-negative, got [{}..={}]",
+ range.from(),
+ range.to()
+ ),
+ source: None,
+ });
+ }
+ let to = usize::try_from(range.to()).map_err(|e| Error::DataInvalid {
+ message: format!(
+ "Blob row selection upper bound {} is out of range",
+ range.to()
+ ),
+ source: Some(Box::new(e)),
+ })?;
+ if to >= total_rows && total_rows != 0 {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob row selection [{}..={}] exceeds available rows {}",
+ range.from(),
+ range.to(),
+ total_rows
+ ),
+ source: None,
+ });
+ }
+ }
+
+ let next_in_range = ranges.first().map_or(0, RowRange::from);
+ RowSelectionState::Ranges {
+ total_rows,
+ ranges,
+ range_idx: 0,
+ next_in_range,
+ }
+ }
+ };
+
+ Ok(Self { state })
+ }
+
+ fn next_batch(&mut self, batch_size: usize) -> Option<Vec<usize>> {
+ if batch_size == 0 {
+ return None;
+ }
+
+ match &mut self.state {
+ RowSelectionState::All { next, total_rows } => {
+ if *next >= *total_rows {
+ return None;
+ }
+
+ let end = (*next + batch_size).min(*total_rows);
+ let batch: Vec<usize> = (*next..end).collect();
+ *next = end;
+ Some(batch)
+ }
+ RowSelectionState::Ranges {
+ total_rows,
+ ranges,
+ range_idx,
+ next_in_range,
+ } => {
+ if *range_idx >= ranges.len() || *total_rows == 0 {
+ return None;
+ }
+
+ let mut batch = Vec::with_capacity(batch_size);
+ while batch.len() < batch_size && *range_idx < ranges.len() {
+ let range = &ranges[*range_idx];
+ if *next_in_range > range.to() {
+ *range_idx += 1;
+ if *range_idx < ranges.len() {
+ *next_in_range = ranges[*range_idx].from();
+ }
+ continue;
+ }
+
+ batch.push(*next_in_range as usize);
+ *next_in_range += 1;
+ }
+
+ if batch.is_empty() {
+ None
+ } else {
+ Some(batch)
+ }
+ }
+ }
+ }
+}
+
+fn decode_delta_varints(bytes: &[u8]) -> crate::Result<Vec<i64>> {
+ let mut values = Vec::new();
+ let mut cursor = 0usize;
+ let mut previous = 0_i64;
+
+ while cursor < bytes.len() {
+ let (delta, consumed) = decode_varint(&bytes[cursor..])?;
+ cursor += consumed;
+
+ let value = if values.is_empty() {
+ delta
+ } else {
+ previous
+ .checked_add(delta)
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "Blob delta-varint index overflow after previous value {previous}"
+ ),
+ source: None,
+ })?
+ };
+ values.push(value);
+ previous = value;
+ }
+
+ Ok(values)
+}
+
+fn decode_varint(bytes: &[u8]) -> crate::Result<(i64, usize)> {
+ let mut value = 0_u64;
+ let mut shift = 0_u32;
+
+ for (idx, byte) in bytes.iter().copied().enumerate() {
+ value |= u64::from(byte & 0x7f) << shift;
+ if (byte & 0x80) == 0 {
+ let decoded = ((value >> 1) as i64) ^ (-((value & 1) as i64));
+ return Ok((decoded, idx + 1));
+ }
+
+ shift += 7;
+ if shift > 63 {
+ return Err(Error::DataInvalid {
+ message: "Blob delta-varint index overflow".to_string(),
+ source: None,
+ });
+ }
+ }
+
+ Err(Error::DataInvalid {
+ message: "Unexpected end of blob delta-varint index".to_string(),
+ source: None,
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::btree::test_util::BytesFileRead;
+ use crate::spec::BlobType;
+ use arrow_array::Array;
+ use bytes::Bytes;
+ use futures::TryStreamExt;
+ use std::sync::atomic::{AtomicUsize, Ordering};
+ use std::time::Duration;
+
+ #[allow(dead_code)]
+ mod blob_test_utils {
+ include!(concat!(
+ env!("CARGO_MANIFEST_DIR"),
+ "/../blob_test_utils.rs"
+ ));
+ }
+
+ #[tokio::test]
+ async fn test_blob_reader_reads_inline_bytes_and_selection() {
+ let read_fields = vec![DataField::new(
+ 0,
+ "payload".to_string(),
+ DataType::Blob(BlobType::new()),
+ )];
+ let reader = BlobFormatReader;
+ let file_bytes = load_blob_fixture("blob-basic.blob");
+
+ let stream = reader
+ .read_batch_stream(
+ Box::new(BytesFileRead(Bytes::from(file_bytes.clone()))),
+ file_bytes.len() as u64,
+ &read_fields,
+ None,
+ Some(2),
+ None,
+ )
+ .await
+ .unwrap();
+ let batches = stream.try_collect::<Vec<_>>().await.unwrap();
+
+ assert_eq!(batches.len(), 2);
+ assert_eq!(
+ collect_binary_values(&batches[0]),
+ vec![Some(b"hello".to_vec()), None]
+ );
+ assert_eq!(
+ collect_binary_values(&batches[1]),
+ vec![Some(b"world".to_vec()), Some(Vec::new())]
+ );
+
+ let selected = reader
+ .read_batch_stream(
+ Box::new(BytesFileRead(Bytes::from(file_bytes.clone()))),
+ file_bytes.len() as u64,
+ &read_fields,
+ None,
+ Some(8),
+ Some(vec![RowRange::new(2, 3)]),
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(selected.len(), 1);
+ assert_eq!(
+ collect_binary_values(&selected[0]),
+ vec![Some(b"world".to_vec()), Some(Vec::new())]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_blob_reader_reads_payloads_with_bounded_parallelism() {
+ let read_fields = vec![DataField::new(
+ 0,
+ "payload".to_string(),
+ DataType::Blob(BlobType::new()),
+ )];
+ let file_bytes = load_blob_fixture("blob-basic.blob");
+ let reader = TrackingFileRead::new(Bytes::from(file_bytes.clone()));
+
+ let batches = BlobFormatReader
+ .read_batch_stream(
+ Box::new(reader.clone()),
+ file_bytes.len() as u64,
+ &read_fields,
+ None,
+ Some(8),
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(batches.len(), 1);
+ assert_eq!(
+ collect_binary_values(&batches[0]),
+ vec![
+ Some(b"hello".to_vec()),
+ None,
+ Some(b"world".to_vec()),
+ Some(Vec::new()),
+ ]
+ );
+ assert!(reader.max_in_flight() > 1);
+ }
+
+ #[test]
+ fn test_blob_reader_test_helper_matches_java_fixture() {
+ let generated = blob_test_utils::build_blob_file_bytes(&basic_blob_rows());
+
+ assert_eq!(generated, load_blob_fixture("blob-basic.blob"));
+ }
+
+ #[tokio::test]
+ async fn test_blob_reader_supports_empty_projection() {
+ let reader = BlobFormatReader;
+ let file_bytes = load_blob_fixture("blob-basic.blob");
+
+ let batches = reader
+ .read_batch_stream(
+ Box::new(BytesFileRead(Bytes::from(file_bytes.clone()))),
+ file_bytes.len() as u64,
+ &[],
+ None,
+ Some(2),
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(batches.len(), 2);
+ assert!(batches[0].columns().is_empty());
+ assert_eq!(batches[0].num_rows(), 2);
+ assert!(batches[1].columns().is_empty());
+ assert_eq!(batches[1].num_rows(), 2);
+ }
+
+ #[tokio::test]
+ async fn test_blob_reader_rejects_out_of_range_selection() {
+ let reader = BlobFormatReader;
+ let file_bytes = load_blob_fixture("blob-basic.blob");
+ let read_fields = vec![DataField::new(
+ 0,
+ "payload".to_string(),
+ DataType::Blob(BlobType::new()),
+ )];
+
+ let result = reader
+ .read_batch_stream(
+ Box::new(BytesFileRead(Bytes::from(file_bytes.clone()))),
+ file_bytes.len() as u64,
+ &read_fields,
+ None,
+ None,
+ Some(vec![RowRange::new(0, 4)]),
+ )
+ .await;
+
+ assert!(
+ matches!(result, Err(Error::DataInvalid { message, .. }) if message.contains("exceeds available rows"))
+ );
+ }
+
+ #[tokio::test]
+ async fn test_blob_reader_rejects_wrong_field_family() {
+ let reader = BlobFormatReader;
+ let file_bytes = load_blob_fixture("blob-basic.blob");
+ let read_fields = vec![DataField::new(
+ 0,
+ "payload".to_string(),
+ DataType::Int(crate::spec::IntType::new()),
+ )];
+
+ let result = reader
+ .read_batch_stream(
+ Box::new(BytesFileRead(Bytes::from(file_bytes.clone()))),
+ file_bytes.len() as u64,
+ &read_fields,
+ None,
+ None,
+ None,
+ )
+ .await;
+
+ assert!(
+ matches!(result, Err(Error::DataInvalid { message, .. }) if message.contains("Blob field"))
+ );
+ }
+
+ #[tokio::test]
+ async fn test_blob_reader_rejects_unsupported_version() {
+ let mut file_bytes = blob_test_utils::build_blob_file_bytes(&basic_blob_rows());
+ let last = file_bytes.len() - 1;
+ file_bytes[last] = 2;
+
+ let result = BlobFormatReader
+ .read_batch_stream(
+ Box::new(BytesFileRead(Bytes::from(file_bytes.clone()))),
+ file_bytes.len() as u64,
+ &[DataField::new(
+ 0,
+ "payload".to_string(),
+ DataType::Blob(BlobType::new()),
+ )],
+ None,
+ None,
+ None,
+ )
+ .await;
+
+ assert!(
+ matches!(result, Err(Error::Unsupported { message }) if message.contains("footer version"))
+ );
+ }
+
+ #[tokio::test]
+ async fn test_blob_reader_rejects_truncated_entry() {
+ let mut file_bytes = blob_test_utils::build_blob_file_bytes(&basic_blob_rows());
+ let footer_start = file_bytes.len() - BLOB_FOOTER_SIZE as usize;
+ let index_length = i32::from_le_bytes(
+ file_bytes[footer_start..footer_start + 4]
+ .try_into()
+ .unwrap(),
+ ) as usize;
+ let index_start = footer_start - index_length;
+ let lengths = decode_delta_varints(&file_bytes[index_start..footer_start]).unwrap();
+ let mut replacement_lengths = lengths.clone();
+ replacement_lengths[0] = 15;
+ let replacement = blob_test_utils::encode_delta_varints(&replacement_lengths);
+ file_bytes.splice(index_start..footer_start, replacement.iter().copied());
+ let footer_start = file_bytes.len() - BLOB_FOOTER_SIZE as usize;
+ file_bytes[footer_start..footer_start + 4]
+ .copy_from_slice(&(replacement.len() as i32).to_le_bytes());
+
+ let result = BlobFormatReader
+ .read_batch_stream(
+ Box::new(BytesFileRead(Bytes::from(file_bytes.clone()))),
+ file_bytes.len() as u64,
+ &[DataField::new(
+ 0,
+ "payload".to_string(),
+ DataType::Blob(BlobType::new()),
+ )],
+ None,
+ None,
+ None,
+ )
+ .await;
+
+ assert!(!lengths.is_empty());
+ assert!(
+ matches!(result, Err(Error::DataInvalid { message, .. }) if message.contains("minimum overhead"))
+ );
+ }
+
+ fn basic_blob_rows() -> [Option<&'static [u8]>; 4] {
+ [
+ Some(&b"hello"[..]),
+ None,
+ Some(&b"world"[..]),
+ Some(&b""[..]),
+ ]
+ }
+
+ fn collect_binary_values(batch: &RecordBatch) -> Vec<Option<Vec<u8>>> {
+ let array = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<arrow_array::BinaryArray>()
+ .unwrap();
+ (0..array.len())
+ .map(|idx| (!array.is_null(idx)).then(|| array.value(idx).to_vec()))
+ .collect()
+ }
+
+ fn load_blob_fixture(name: &str) -> Vec<u8> {
+ let path = format!("{}/testdata/blob/{name}", env!("CARGO_MANIFEST_DIR"));
+ std::fs::read(&path).unwrap_or_else(|e| panic!("Failed to read {path}: {e}"))
+ }
+
+ #[derive(Clone)]
+ struct TrackingFileRead {
+ bytes: Bytes,
+ in_flight: Arc<AtomicUsize>,
+ max_in_flight: Arc<AtomicUsize>,
+ }
+
+ impl TrackingFileRead {
+ fn new(bytes: Bytes) -> Self {
+ Self {
+ bytes,
+ in_flight: Arc::new(AtomicUsize::new(0)),
+ max_in_flight: Arc::new(AtomicUsize::new(0)),
+ }
+ }
+
+ fn max_in_flight(&self) -> usize {
+ self.max_in_flight.load(Ordering::SeqCst)
+ }
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for TrackingFileRead {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ let in_flight = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
+ self.max_in_flight.fetch_max(in_flight, Ordering::SeqCst);
+ tokio::time::sleep(Duration::from_millis(10)).await;
+ self.in_flight.fetch_sub(1, Ordering::SeqCst);
+ Ok(self.bytes.slice(range.start as usize..range.end as usize))
+ }
+ }
+}
diff --git a/crates/paimon/src/arrow/format/mod.rs b/crates/paimon/src/arrow/format/mod.rs
index 454e621..63c2f85 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -16,6 +16,7 @@
// under the License.
mod avro;
+mod blob;
mod orc;
mod parquet;
@@ -88,6 +89,8 @@ pub(crate) trait FormatFileWriter: Send {
pub(crate) fn create_format_reader(path: &str) -> crate::Result<Box<dyn FormatFileReader>> {
if path.to_ascii_lowercase().ends_with(".parquet") {
Ok(Box::new(parquet::ParquetFormatReader))
+ } else if path.to_ascii_lowercase().ends_with(".blob") {
+ Ok(Box::new(blob::BlobFormatReader))
} else if path.to_ascii_lowercase().ends_with(".orc") {
Ok(Box::new(orc::OrcFormatReader))
} else if path.to_ascii_lowercase().ends_with(".avro") {
@@ -95,7 +98,7 @@ pub(crate) fn create_format_reader(path: &str) -> crate::Result<Box<dyn FormatFi
} else {
Err(Error::Unsupported {
message: format!(
- "unsupported file format: expected .parquet, .orc, or .avro, got: {path}"
+ "unsupported file format: expected .parquet, .blob, .orc, or .avro, got: {path}"
),
})
}
diff --git a/crates/paimon/src/table/data_evolution_reader.rs b/crates/paimon/src/table/data_evolution_reader.rs
index 4ea2b87..616a0d2 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -21,13 +21,12 @@ use super::data_file_reader::{
};
use crate::arrow::build_target_arrow_schema;
use crate::io::FileIO;
-use crate::spec::{DataField, DataFileMeta, ROW_ID_FIELD_NAME};
+use crate::spec::{DataField, DataFileMeta, DataType, ROW_ID_FIELD_NAME};
use crate::table::schema_manager::SchemaManager;
use crate::table::ArrowRecordBatchStream;
use crate::table::RowRange;
use crate::{DataSplit, Error};
-use arrow_array::{Array, Int64Array, RecordBatch};
-
+use arrow_array::{Int64Array, RecordBatch};
use async_stream::try_stream;
use futures::StreamExt;
use std::collections::HashMap;
@@ -59,7 +58,7 @@ fn is_raw_convertible(files: &[DataFileMeta]) -> bool {
}
/// Reads data files in data evolution mode, merging columns from files
-/// that share the same row ID range.
+/// that share the same logical row range.
pub(crate) struct DataEvolutionReader {
file_io: FileIO,
schema_manager: SchemaManager,
@@ -101,10 +100,6 @@ impl DataEvolutionReader {
}
/// Read data files in data evolution mode.
- ///
- /// Each DataSplit contains files grouped by `first_row_id`. Files within a split may contain
- /// different columns for the same logical rows. This method reads each file and merges them
- /// column-wise, respecting `max_sequence_number` for conflict resolution.
pub fn read(self, data_splits: &[DataSplit]) -> crate::Result<ArrowRecordBatchStream> {
let splits: Vec<DataSplit> = data_splits.to_vec();
@@ -123,12 +118,14 @@ impl DataEvolutionReader {
if is_raw_convertible(split.data_files()) {
for file_meta in split.data_files().to_vec() {
- let data_fields: Option<Vec<DataField>> = if file_meta.schema_id != self.table_schema_id {
- let data_schema = self.schema_manager.schema(file_meta.schema_id).await?;
- Some(data_schema.fields().to_vec())
- } else {
- None
- };
+ let data_fields: Option<Vec<DataField>> =
+ if file_meta.schema_id != self.table_schema_id {
+ let data_schema =
+ self.schema_manager.schema(file_meta.schema_id).await?;
+ Some(data_schema.fields().to_vec())
+ } else {
+ None
+ };
let has_row_id = file_meta.first_row_id.is_some();
let effective_row_ranges = if has_row_id { row_ranges.clone() } else { None };
@@ -175,56 +172,39 @@ impl DataEvolutionReader {
}
}
} else {
- let files = split.data_files();
- if !files.iter().all(|f| f.first_row_id.is_some()) {
- Err(Error::UnexpectedError {
- message: "All files in a field merge split should have first_row_id".to_string(),
- source: None,
- })?;
- }
- if !files.iter().all(|f| f.row_count == files[0].row_count) {
- Err(Error::UnexpectedError {
- message: "All files in a field merge split should have the same row count".to_string(),
- source: None,
- })?;
- }
- if !files.iter().all(|f| f.first_row_id == files[0].first_row_id) {
- Err(Error::UnexpectedError {
- message: "All files in a field merge split should have the same first row id".to_string(),
- source: None,
- })?;
- }
-
- let group_base_row_id = files[0].first_row_id;
- let has_group_row_id = group_base_row_id.is_some();
- let group_row_count = files[0].row_count;
- let effective_row_ranges = if has_group_row_id { row_ranges.clone() } else { None };
+ let prepared_group = PreparedMergeGroup::new(split.data_files())?;
+ let effective_row_ranges = row_ranges.clone();
+ let expected_output_rows = count_selected_rows(
+ prepared_group.first_row_id,
+ prepared_group.logical_row_count,
+ effective_row_ranges.as_deref(),
+ )?;
- let selected_row_ids = if self.row_id_index.is_some() && has_group_row_id {
+ let selected_row_ids = if self.row_id_index.is_some() {
effective_row_ranges.as_ref().map(|ranges| {
expand_selected_row_ids(
- group_base_row_id.unwrap(),
- group_row_count,
+ prepared_group.first_row_id,
+ prepared_group.logical_row_count,
ranges,
)
})
} else {
None
};
- let mut row_id_cursor = group_base_row_id.unwrap_or(0);
+ let mut row_id_cursor = prepared_group.first_row_id;
let mut row_id_offset: usize = 0;
let mut merge_stream = self.merge_files_by_columns(
&split,
+ &prepared_group,
effective_row_ranges,
+ expected_output_rows,
)?;
while let Some(batch) = merge_stream.next().await {
let batch = batch?;
let num_rows = batch.num_rows();
if let Some(idx) = self.row_id_index {
- if !has_group_row_id {
- yield append_null_row_id_column(batch, idx, &self.output_schema)?;
- } else if let Some(ref ids) = selected_row_ids {
+ if let Some(ref ids) = selected_row_ids {
yield attach_row_id(batch, idx, ids, &mut row_id_offset, &self.output_schema)?;
} else {
let row_ids: Vec<i64> = (row_id_cursor..row_id_cursor + num_rows as i64).collect();
@@ -242,17 +222,19 @@ impl DataEvolutionReader {
.boxed())
}
- /// Merge multiple files column-wise for data evolution, streaming with bounded memory.
+ /// Merge multiple logical sources column-wise for data evolution.
///
- /// Uses field IDs (not column names) to resolve which file provides which column,
- /// ensuring correctness across schema evolution (column rename, add, drop).
+ /// 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.
fn merge_files_by_columns(
&self,
split: &DataSplit,
+ prepared_group: &PreparedMergeGroup,
row_ranges: Option<Vec<RowRange>>,
+ expected_output_rows: usize,
) -> crate::Result<ArrowRecordBatchStream> {
- let data_files = split.data_files();
- if data_files.is_empty() {
+ if prepared_group.files.is_empty() {
return Ok(futures::stream::empty().boxed());
}
@@ -260,93 +242,35 @@ impl DataEvolutionReader {
let schema_manager = self.schema_manager.clone();
let table_schema_id = self.table_schema_id;
let split = split.clone();
- let data_files: Vec<DataFileMeta> = data_files.to_vec();
+ let prepared_group = prepared_group.clone();
let read_type = self.file_read_type.clone();
let table_fields = self.table_fields.clone();
// Batch size for column-merge output. Matches the default Parquet reader batch size.
const MERGE_BATCH_SIZE: usize = 1024;
- let output_batch_size: usize = MERGE_BATCH_SIZE;
let target_schema = build_target_arrow_schema(&read_type)?;
Ok(try_stream! {
- // Pre-load schemas and collect field IDs + data_fields per file.
- let mut file_info: HashMap<usize, (Vec<i32>, Option<Vec<DataField>>)> = HashMap::new();
-
- for (file_idx, file_meta) in data_files.iter().enumerate() {
- let (field_ids, data_fields) = if file_meta.schema_id != table_schema_id {
- let file_schema = schema_manager.schema(file_meta.schema_id).await?;
- let file_fields = file_schema.fields();
-
- let ids: Vec<i32> = if let Some(ref wc) = file_meta.write_cols {
- wc.iter()
- .filter_map(|name| file_fields.iter().find(|f| f.name() == name).map(|f| f.id()))
- .collect()
- } else {
- file_fields.iter().map(|f| f.id()).collect()
- };
-
- (ids, Some(file_fields.to_vec()))
- } else {
- let ids: Vec<i32> = if let Some(ref wc) = file_meta.write_cols {
- wc.iter()
- .filter_map(|name| table_fields.iter().find(|f| f.name() == name).map(|f| f.id()))
- .collect()
- } else {
- table_fields.iter().map(|f| f.id()).collect()
- };
-
- (ids, None)
- };
+ 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)?;
- file_info.insert(file_idx, (field_ids, data_fields));
- }
-
- // Determine which file provides each field ID, resolving conflicts by max_sequence_number.
- let mut field_id_source: HashMap<i32, (usize, i64)> = HashMap::new();
- for (file_idx, file_meta) in data_files.iter().enumerate() {
- let (ref field_ids, _) = file_info[&file_idx];
- for &fid in field_ids {
- let entry = field_id_source
- .entry(fid)
- .or_insert((file_idx, i64::MIN));
- if file_meta.max_sequence_number > entry.1 {
- *entry = (file_idx, file_meta.max_sequence_number);
- }
- }
- }
-
- // For each projected field, determine which file provides it (by field ID).
- let mut file_read_columns: HashMap<usize, Vec<String>> = HashMap::new();
- for field in &read_type {
- if let Some(&(file_idx, _)) = field_id_source.get(&field.id()) {
- file_read_columns
- .entry(file_idx)
- .or_default()
- .push(field.name().to_string());
- }
- }
-
- let column_plan: Vec<(Option<usize>, String)> = read_type
+ let active_source_indices: Vec<usize> = source_plan
+ .sources
.iter()
- .map(|field| {
- let file_idx = field_id_source.get(&field.id()).map(|&(idx, _)| idx);
- (file_idx, field.name().to_string())
- })
+ .enumerate()
+ .filter_map(|(idx, source)| (!source.read_fields().is_empty()).then_some(idx))
.collect();
- let active_file_indices: Vec<usize> = file_read_columns.keys().copied().collect();
-
// Edge case: no file provides any projected column.
- if active_file_indices.is_empty() {
- let first_row_id = data_files[0].first_row_id.unwrap_or(0);
- let file_row_count = data_files[0].row_count;
- let total_rows = match &row_ranges {
- Some(ranges) => expand_selected_row_ids(first_row_id, file_row_count, ranges).len(),
- None => file_row_count as usize,
- };
- let mut emitted = 0;
- while emitted < total_rows {
- let rows_to_emit = (total_rows - emitted).min(output_batch_size);
+ 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()
@@ -368,113 +292,1278 @@ impl DataEvolutionReader {
emitted += rows_to_emit;
yield batch;
}
- } else {
- // Open a stream for each active file via DataFileReader.
- let mut file_streams: HashMap<usize, ArrowRecordBatchStream> = HashMap::new();
- for &file_idx in &active_file_indices {
- let file_cols = file_read_columns.get(&file_idx).cloned().unwrap_or_default();
- let file_read_type: Vec<DataField> = file_cols
- .iter()
- .filter_map(|col_name| read_type.iter().find(|f| f.name() == col_name).cloned())
- .collect();
-
- let (_, ref data_fields) = file_info[&file_idx];
-
- let file_reader = DataFileReader::new(
- file_io.clone(),
- schema_manager.clone(),
- table_schema_id,
- table_fields.clone(),
- file_read_type,
- Vec::new(),
- );
- let stream = file_reader.read_single_file_stream(
- &split,
- data_files[file_idx].clone(),
- data_fields.clone(),
- None,
- row_ranges.clone(),
- )?;
- file_streams.insert(file_idx, stream);
- }
+ return;
+ }
- // Per-file cursor: current batch + offset within it.
- let mut file_cursors: HashMap<usize, (RecordBatch, usize)> = HashMap::new();
+ let mut source_streams: Vec<Option<ArrowRecordBatchStream>> = source_plan
+ .sources
+ .iter()
+ .map(|source| {
+ if source.read_fields().is_empty() {
+ Ok(None)
+ } else {
+ open_source_stream(
+ &split,
+ source,
+ row_ranges.clone(),
+ file_io.clone(),
+ schema_manager.clone(),
+ table_schema_id,
+ table_fields.clone(),
+ )
+ .map(Some)
+ }
+ })
+ .collect::<crate::Result<_>>()?;
+ let mut source_cursors: Vec<Option<(RecordBatch, usize)>> = source_plan
+ .sources
+ .iter()
+ .map(|_| None)
+ .collect();
+ let mut emitted_rows = 0usize;
- loop {
- for &file_idx in &active_file_indices {
- let needs_next = match file_cursors.get(&file_idx) {
- None => true,
- Some((batch, offset)) => *offset >= batch.num_rows(),
- };
- if needs_next {
- file_cursors.remove(&file_idx);
- if let Some(stream) = file_streams.get_mut(&file_idx) {
- if let Some(batch_result) = stream.next().await {
- let batch = batch_result?;
- if batch.num_rows() > 0 {
- file_cursors.insert(file_idx, (batch, 0));
- }
+ loop {
+ for &source_idx in &active_source_indices {
+ let needs_next = match source_cursors[source_idx].as_ref() {
+ None => true,
+ Some((batch, offset)) => *offset >= batch.num_rows(),
+ };
+ if needs_next {
+ source_cursors[source_idx] = None;
+ if let Some(stream) = source_streams[source_idx].as_mut() {
+ while let Some(batch_result) = stream.next().await {
+ let batch = batch_result?;
+ if batch.num_rows() == 0 {
+ continue;
}
+ source_cursors[source_idx] = Some((batch, 0));
+ break;
}
}
}
+ }
- // All files in a merge group have the same row count (validated above),
- // so any file stream exhausting means all streams are done.
- if active_file_indices.iter().any(|idx| !file_cursors.contains_key(idx)) {
- break;
- }
-
- let remaining: usize = active_file_indices
- .iter()
- .map(|idx| {
- let (batch, offset) = file_cursors.get(idx).unwrap();
- batch.num_rows() - offset
- })
- .min()
- .unwrap_or(0);
-
- if remaining == 0 {
+ let finished_sources = active_source_indices
+ .iter()
+ .filter(|&&idx| source_cursors[idx].is_none())
+ .count();
+ if finished_sources > 0 {
+ if finished_sources == active_source_indices.len() {
+ if emitted_rows != expected_output_rows {
+ Err(Error::DataInvalid {
+ message: format!(
+ "Merged data evolution sources produced {emitted_rows} rows but expected {expected_output_rows}"
+ ),
+ source: None,
+ })?;
+ }
break;
}
- let rows_to_emit = remaining.min(output_batch_size);
+ Err(Error::DataInvalid {
+ message: "Data evolution sources exhausted at different row counts".to_string(),
+ source: None,
+ })?;
+ }
- let mut columns: Vec<Arc<dyn arrow_array::Array>> =
- Vec::with_capacity(column_plan.len());
+ let remaining = active_source_indices
+ .iter()
+ .map(|&idx| {
+ let (batch, offset) = source_cursors[idx].as_ref().unwrap();
+ batch.num_rows() - offset
+ })
+ .min()
+ .unwrap_or(0);
+ if remaining == 0 {
+ Err(Error::UnexpectedError {
+ message: "Data evolution source cursor reached an empty batch".to_string(),
+ source: None,
+ })?;
+ }
- for (i, (file_idx_opt, col_name)) in column_plan.iter().enumerate() {
- let target_field = &target_schema.fields()[i];
- let col = file_idx_opt
- .and_then(|file_idx| file_cursors.get(&file_idx))
- .and_then(|(batch, offset)| {
- batch
- .schema()
- .index_of(col_name)
- .ok()
- .map(|col_idx| batch.column(col_idx).slice(*offset, rows_to_emit))
- });
+ let rows_to_emit = remaining.min(MERGE_BATCH_SIZE);
+ let mut columns: Vec<Arc<dyn arrow_array::Array>> =
+ Vec::with_capacity(source_plan.column_plan.len());
- columns.push(col.unwrap_or_else(|| {
+ for (idx, provider) in source_plan.column_plan.iter().enumerate() {
+ let target_field = &target_schema.fields()[idx];
+ let array = provider
+ .and_then(|(source_idx, field_offset)| {
+ source_cursors[source_idx].as_ref().map(|(batch, offset)| {
+ batch.column(field_offset).slice(*offset, rows_to_emit)
+ })
+ })
+ .unwrap_or_else(|| {
arrow_array::new_null_array(target_field.data_type(), rows_to_emit)
- }));
+ });
+ columns.push(array);
+ }
+
+ for &source_idx in &active_source_indices {
+ if let Some((_, offset)) = source_cursors[source_idx].as_mut() {
+ *offset += rows_to_emit;
}
+ }
- for &file_idx in &active_file_indices {
- if let Some((_, ref mut offset)) = file_cursors.get_mut(&file_idx) {
- *offset += rows_to_emit;
+ emitted_rows += rows_to_emit;
+ let merged =
+ RecordBatch::try_new(target_schema.clone(), columns).map_err(|e| {
+ Error::UnexpectedError {
+ message: format!("Failed to build merged RecordBatch: {e}"),
+ source: Some(Box::new(e)),
}
+ })?;
+ yield merged;
+ }
+ }
+ .boxed())
+ }
+}
+
+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>,
+) -> crate::Result<ArrowRecordBatchStream> {
+ let file_reader = DataFileReader::new(
+ file_io,
+ schema_manager,
+ table_schema_id,
+ table_fields,
+ source.read_fields().to_vec(),
+ Vec::new(),
+ );
+
+ 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())
+ }
+ }
+}
- let merged = RecordBatch::try_new(target_schema.clone(), columns).map_err(|e| Error::UnexpectedError {
- message: format!("Failed to build merged RecordBatch: {e}"),
- source: Some(Box::new(e)),
- })?;
- yield merged;
+#[derive(Debug, Clone)]
+struct PreparedMergeGroup {
+ files: Vec<DataFileMeta>,
+ logical_row_count: i64,
+ first_row_id: i64,
+}
+
+impl PreparedMergeGroup {
+ fn new(files: &[DataFileMeta]) -> crate::Result<Self> {
+ let files = normalize_merge_group(files.to_vec())?;
+ if files.is_empty() {
+ return Ok(Self {
+ files,
+ logical_row_count: 0,
+ first_row_id: 0,
+ });
+ }
+
+ let data_files: Vec<&DataFileMeta> = files
+ .iter()
+ .filter(|file| !is_blob_file_name(&file.file_name))
+ .collect();
+ if data_files.is_empty() {
+ return Err(Error::DataInvalid {
+ message: "Field merge split containing .blob files requires at least one non-blob data file".to_string(),
+ source: None,
+ });
+ }
+
+ let first_data_file = data_files[0];
+ let first_row_id = first_data_file
+ .first_row_id
+ .ok_or_else(|| Error::DataInvalid {
+ message: "All files in a field merge split should have first_row_id".to_string(),
+ source: None,
+ })?;
+ let logical_row_count = first_data_file.row_count;
+
+ for file in data_files.iter().skip(1) {
+ if file.first_row_id != Some(first_row_id) || file.row_count != logical_row_count {
+ return Err(Error::DataInvalid {
+ message: "All non-blob files in a field merge split should have the same row id range".to_string(),
+ source: None,
+ });
+ }
+ }
+
+ Ok(Self {
+ files,
+ logical_row_count,
+ first_row_id,
+ })
+ }
+}
+
+#[derive(Debug, Clone)]
+struct ResolvedFileInfo {
+ field_ids: Vec<i32>,
+ data_fields: Option<Vec<DataField>>,
+}
+
+async fn load_file_infos(
+ schema_manager: &SchemaManager,
+ table_schema_id: i64,
+ table_fields: &[DataField],
+ files: &[DataFileMeta],
+) -> crate::Result<Vec<ResolvedFileInfo>> {
+ let mut infos = Vec::with_capacity(files.len());
+
+ for file in files {
+ if file.schema_id == table_schema_id {
+ infos.push(ResolvedFileInfo {
+ field_ids: resolve_field_ids(file, table_fields)?,
+ data_fields: None,
+ });
+ } else {
+ let data_schema = schema_manager.schema(file.schema_id).await?;
+ let data_fields = data_schema.fields().to_vec();
+ infos.push(ResolvedFileInfo {
+ field_ids: resolve_field_ids(file, &data_fields)?,
+ data_fields: Some(data_fields),
+ });
+ }
+ }
+
+ Ok(infos)
+}
+
+fn resolve_field_ids(file: &DataFileMeta, fields: &[DataField]) -> crate::Result<Vec<i32>> {
+ match &file.write_cols {
+ Some(write_cols) => write_cols
+ .iter()
+ .map(|name| {
+ fields
+ .iter()
+ .find(|field| field.name() == name)
+ .map(|field| field.id())
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "Failed to resolve write column '{}' in file '{}'",
+ name, file.file_name
+ ),
+ source: None,
+ })
+ })
+ .collect(),
+ None => Ok(fields.iter().map(|field| field.id()).collect()),
+ }
+}
+
+#[derive(Debug, Clone)]
+struct SourcePlan {
+ sources: Vec<FieldSource>,
+ column_plan: Vec<Option<(usize, usize)>>,
+}
+
+fn build_source_plan(
+ prepared_group: &PreparedMergeGroup,
+ file_infos: &[ResolvedFileInfo],
+ read_type: &[DataField],
+) -> crate::Result<SourcePlan> {
+ let mut sources = Vec::new();
+ let mut normal_source_indices: HashMap<usize, usize> = HashMap::new();
+ let mut blob_source_indices: HashMap<i32, usize> = HashMap::new();
+ let mut expected_blob_row_count: Option<i64> = None;
+
+ for (file_idx, file) in prepared_group.files.iter().enumerate() {
+ let info = &file_infos[file_idx];
+ if is_blob_file_name(&file.file_name) {
+ let field_id = resolve_blob_field_id(file, info)?;
+ let expected_row_count = expected_blob_row_count.ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ "Blob file '{}' must be ordered after a non-blob data file",
+ file.file_name
+ ),
+ source: None,
+ })?;
+
+ let source_idx = if let Some(&existing_idx) = blob_source_indices.get(&field_id) {
+ existing_idx
+ } else {
+ let source_idx = sources.len();
+ sources.push(FieldSource::BlobBunch {
+ bunch: BlobBunch::new(expected_row_count),
+ data_fields: info.data_fields.clone(),
+ read_fields: Vec::new(),
+ });
+ blob_source_indices.insert(field_id, source_idx);
+ source_idx
+ };
+
+ sources[source_idx]
+ .blob_bunch_mut()
+ .unwrap()
+ .add(file.clone())?;
+ } else {
+ expected_blob_row_count = Some(file.row_count);
+ let source_idx = sources.len();
+ sources.push(FieldSource::DataFile {
+ file: Box::new(file.clone()),
+ data_fields: info.data_fields.clone(),
+ read_fields: Vec::new(),
+ });
+ normal_source_indices.insert(file_idx, source_idx);
+ }
+ }
+
+ let mut column_plan = Vec::with_capacity(read_type.len());
+ for field in read_type {
+ let source_idx = if matches!(field.data_type(), DataType::Blob(_)) {
+ blob_source_indices.get(&field.id()).copied()
+ } else {
+ select_normal_provider(
+ &prepared_group.files,
+ file_infos,
+ &normal_source_indices,
+ field.id(),
+ )
+ };
+
+ if let Some(source_idx) = source_idx {
+ let field_offset = sources[source_idx].add_read_field(field.clone());
+ column_plan.push(Some((source_idx, field_offset)));
+ } else {
+ column_plan.push(None);
+ }
+ }
+
+ for source in &sources {
+ if let FieldSource::BlobBunch {
+ bunch, read_fields, ..
+ } = source
+ {
+ if !read_fields.is_empty() && bunch.row_count() != prepared_group.logical_row_count {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob bunch row count {} does not match logical row count {}",
+ bunch.row_count(),
+ prepared_group.logical_row_count
+ ),
+ source: None,
+ });
+ }
+ }
+ }
+
+ Ok(SourcePlan {
+ sources,
+ column_plan,
+ })
+}
+
+fn select_normal_provider(
+ files: &[DataFileMeta],
+ file_infos: &[ResolvedFileInfo],
+ normal_source_indices: &HashMap<usize, usize>,
+ field_id: i32,
+) -> Option<usize> {
+ files.iter().enumerate().find_map(|(file_idx, file)| {
+ if is_blob_file_name(&file.file_name) {
+ return None;
+ }
+
+ file_infos[file_idx]
+ .field_ids
+ .contains(&field_id)
+ .then(|| normal_source_indices.get(&file_idx).copied())
+ .flatten()
+ })
+}
+
+fn resolve_blob_field_id(file: &DataFileMeta, info: &ResolvedFileInfo) -> crate::Result<i32> {
+ if info.field_ids.len() != 1 {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob file '{}' should resolve to exactly one write column, got {}",
+ file.file_name,
+ info.field_ids.len()
+ ),
+ source: None,
+ });
+ }
+
+ Ok(info.field_ids[0])
+}
+
+#[derive(Debug, Clone)]
+enum FieldSource {
+ DataFile {
+ file: Box<DataFileMeta>,
+ data_fields: Option<Vec<DataField>>,
+ read_fields: Vec<DataField>,
+ },
+ BlobBunch {
+ bunch: BlobBunch,
+ data_fields: Option<Vec<DataField>>,
+ read_fields: Vec<DataField>,
+ },
+}
+
+impl FieldSource {
+ fn read_fields(&self) -> &[DataField] {
+ match self {
+ FieldSource::DataFile { read_fields, .. }
+ | FieldSource::BlobBunch { read_fields, .. } => read_fields,
+ }
+ }
+
+ fn add_read_field(&mut self, field: DataField) -> usize {
+ let read_fields = match self {
+ FieldSource::DataFile { read_fields, .. }
+ | FieldSource::BlobBunch { read_fields, .. } => read_fields,
+ };
+ if let Some(offset) = read_fields
+ .iter()
+ .position(|existing| existing.id() == field.id())
+ {
+ return offset;
+ }
+
+ read_fields.push(field);
+ read_fields.len() - 1
+ }
+
+ fn blob_bunch_mut(&mut self) -> Option<&mut BlobBunch> {
+ match self {
+ FieldSource::BlobBunch { bunch, .. } => Some(bunch),
+ FieldSource::DataFile { .. } => None,
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct BlobBunch {
+ files: Vec<DataFileMeta>,
+ expected_row_count: i64,
+ latest_first_row_id: i64,
+ expected_next_first_row_id: i64,
+ latest_max_sequence_number: i64,
+ row_count: i64,
+}
+
+impl BlobBunch {
+ fn new(expected_row_count: i64) -> Self {
+ Self {
+ files: Vec::new(),
+ expected_row_count,
+ latest_first_row_id: -1,
+ expected_next_first_row_id: -1,
+ latest_max_sequence_number: -1,
+ row_count: 0,
+ }
+ }
+
+ fn add(&mut self, file: DataFileMeta) -> crate::Result<()> {
+ if !is_blob_file_name(&file.file_name) {
+ return Err(Error::DataInvalid {
+ message: "Only blob file can be added to a blob bunch.".to_string(),
+ source: None,
+ });
+ }
+
+ let first_row_id = file.first_row_id.ok_or_else(|| Error::DataInvalid {
+ message: format!("Blob file '{}' is missing first_row_id", file.file_name),
+ source: None,
+ })?;
+
+ if first_row_id == self.latest_first_row_id {
+ if file.max_sequence_number >= self.latest_max_sequence_number {
+ return Err(Error::DataInvalid {
+ message:
+ "Blob file with same first row id should have decreasing sequence number."
+ .to_string(),
+ source: None,
+ });
+ }
+ return Ok(());
+ }
+
+ if !self.files.is_empty() {
+ if first_row_id < self.expected_next_first_row_id {
+ if file.max_sequence_number >= self.latest_max_sequence_number {
+ return Err(Error::DataInvalid {
+ message:
+ "Blob file with overlapping row id should have decreasing sequence number."
+ .to_string(),
+ source: None,
+ });
+ }
+ return Ok(());
+ } else if first_row_id > self.expected_next_first_row_id {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob file first row id should be continuous, expect {} but got {}",
+ self.expected_next_first_row_id, first_row_id
+ ),
+ source: None,
+ });
+ }
+
+ if !self.files.is_empty() {
+ let first_file = &self.files[0];
+ if file.schema_id != first_file.schema_id {
+ return Err(Error::DataInvalid {
+ message: "All files in a blob bunch should have the same schema id."
+ .to_string(),
+ source: None,
+ });
+ }
+ if file.write_cols != first_file.write_cols {
+ return Err(Error::DataInvalid {
+ message: "All files in a blob bunch should have the same write columns."
+ .to_string(),
+ source: None,
+ });
}
}
}
- .boxed())
+
+ self.row_count += file.row_count;
+ if self.row_count > self.expected_row_count {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob files row count {} exceed the expected {}",
+ self.row_count, self.expected_row_count
+ ),
+ source: None,
+ });
+ }
+ self.latest_max_sequence_number = file.max_sequence_number;
+ self.latest_first_row_id = first_row_id;
+ self.expected_next_first_row_id = first_row_id + file.row_count;
+ self.files.push(file);
+ Ok(())
+ }
+
+ fn row_count(&self) -> i64 {
+ self.row_count
+ }
+}
+
+fn normalize_merge_group(files: Vec<DataFileMeta>) -> crate::Result<Vec<DataFileMeta>> {
+ let mut data_files = Vec::new();
+ let mut blob_files = Vec::new();
+
+ for file in files {
+ if is_blob_file_name(&file.file_name) {
+ blob_files.push(file);
+ } else {
+ data_files.push(file);
+ }
+ }
+
+ data_files.sort_by(|left, right| right.max_sequence_number.cmp(&left.max_sequence_number));
+ if let Some(first) = data_files.first() {
+ let first_row_id = first.first_row_id.ok_or_else(|| Error::DataInvalid {
+ message: "All data files in a field merge split should have first_row_id".to_string(),
+ source: None,
+ })?;
+ let first_row_count = first.row_count;
+ for file in data_files.iter().skip(1) {
+ if file.first_row_id != Some(first_row_id) || file.row_count != first_row_count {
+ return Err(Error::DataInvalid {
+ message:
+ "All data files in a field merge split should have the same row id range."
+ .to_string(),
+ source: None,
+ });
+ }
+ }
+ }
+
+ blob_files.sort_by(|left, right| {
+ let left_first_row_id = left.first_row_id.unwrap_or(i64::MIN);
+ let right_first_row_id = right.first_row_id.unwrap_or(i64::MIN);
+ left_first_row_id
+ .cmp(&right_first_row_id)
+ .then_with(|| right.max_sequence_number.cmp(&left.max_sequence_number))
+ });
+
+ if blob_files.iter().any(|file| file.first_row_id.is_none()) {
+ return Err(Error::DataInvalid {
+ message: "All blob files in a field merge split should have first_row_id".to_string(),
+ source: None,
+ });
+ }
+
+ data_files.extend(blob_files);
+ Ok(data_files)
+}
+
+fn is_blob_file_name(file_name: &str) -> bool {
+ file_name.to_ascii_lowercase().ends_with(".blob")
+}
+
+fn count_selected_rows(
+ first_row_id: i64,
+ row_count: i64,
+ row_ranges: Option<&[RowRange]>,
+) -> crate::Result<usize> {
+ match row_ranges {
+ Some(ranges) => Ok(expand_selected_row_ids(first_row_id, row_count, ranges).len()),
+ None => usize::try_from(row_count).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid logical row count {row_count}"),
+ source: Some(Box::new(e)),
+ }),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::catalog::Identifier;
+ use crate::io::FileIOBuilder;
+ use crate::spec::stats::BinaryTableStats;
+ use crate::spec::{BinaryRow, BlobType, IntType, Schema, TableSchema};
+ use crate::table::{DataSplitBuilder, Table, TableRead};
+ use arrow_array::{Array, BinaryArray, Int32Array, RecordBatch};
+ use futures::TryStreamExt;
+ use std::fs;
+ use std::path::{Path, PathBuf};
+ use tempfile::tempdir;
+
+ mod blob_test_utils {
+ include!(concat!(
+ env!("CARGO_MANIFEST_DIR"),
+ "/../blob_test_utils.rs"
+ ));
+ }
+
+ #[allow(dead_code)]
+ mod test_utils {
+ include!(concat!(env!("CARGO_MANIFEST_DIR"), "/../test_utils.rs"));
+ }
+
+ use blob_test_utils::write_blob_file;
+ use test_utils::{local_file_path, write_int_parquet_file};
+
+ #[test]
+ fn test_normalize_merge_group_orders_blob_files_after_data_files() {
+ let files = vec![
+ data_file("file1.parquet", 1, 10, 1, None),
+ data_file("file2.blob", 1, 1, 1, Some(vec!["payload"])),
+ data_file("file3.blob", 1, 1, 3, Some(vec!["payload"])),
+ data_file("file4.blob", 2, 9, 1, Some(vec!["payload"])),
+ data_file("file7.parquet", 1, 10, 3, None),
+ ];
+
+ let normalized = normalize_merge_group(files).unwrap();
+ let file_names: Vec<&str> = normalized
+ .iter()
+ .map(|file| file.file_name.as_str())
+ .collect();
+ assert_eq!(
+ file_names,
+ vec![
+ "file7.parquet",
+ "file1.parquet",
+ "file3.blob",
+ "file2.blob",
+ "file4.blob",
+ ]
+ );
+ }
+
+ #[test]
+ fn test_blob_bunch_ignores_same_first_row_id_with_lower_sequence() {
+ let mut bunch = BlobBunch::new(1000);
+ bunch
+ .add(data_file(
+ "blob-high.blob",
+ 0,
+ 100,
+ 3,
+ Some(vec!["payload"]),
+ ))
+ .unwrap();
+ bunch
+ .add(data_file("blob-low.blob", 0, 100, 2, Some(vec!["payload"])))
+ .unwrap();
+
+ assert_eq!(bunch.row_count(), 100);
+ assert_eq!(bunch.files.len(), 1);
+ assert_eq!(bunch.files[0].file_name, "blob-high.blob");
+ }
+
+ #[test]
+ fn test_blob_bunch_rejects_same_first_row_id_with_higher_sequence() {
+ let mut bunch = BlobBunch::new(1000);
+ bunch
+ .add(data_file("blob-low.blob", 0, 100, 2, Some(vec!["payload"])))
+ .unwrap();
+
+ let err = bunch
+ .add(data_file(
+ "blob-high.blob",
+ 0,
+ 100,
+ 3,
+ Some(vec!["payload"]),
+ ))
+ .unwrap_err();
+
+ assert!(
+ matches!(err, Error::DataInvalid { message, .. } if message.contains("same first row id"))
+ );
+ }
+
+ #[test]
+ fn test_blob_bunch_rejects_overlapping_higher_sequence_file() {
+ let mut bunch = BlobBunch::new(1000);
+ bunch
+ .add(data_file("blob1.blob", 0, 100, 1, Some(vec!["payload"])))
+ .unwrap();
+
+ let err = bunch
+ .add(data_file("blob2.blob", 50, 150, 2, Some(vec!["payload"])))
+ .unwrap_err();
+
+ assert!(
+ matches!(err, Error::DataInvalid { message, .. } if message.contains("overlapping row id"))
+ );
+ }
+
+ #[test]
+ fn test_blob_bunch_rejects_non_continuous_first_row_id() {
+ let mut bunch = BlobBunch::new(1000);
+ bunch
+ .add(data_file("blob1.blob", 0, 100, 3, Some(vec!["payload"])))
+ .unwrap();
+
+ let err = bunch
+ .add(data_file("blob2.blob", 150, 100, 2, Some(vec!["payload"])))
+ .unwrap_err();
+
+ assert!(
+ matches!(err, Error::DataInvalid { message, .. } if message.contains("continuous"))
+ );
+ }
+
+ #[test]
+ fn test_blob_bunch_rejects_mixed_write_columns() {
+ let mut bunch = BlobBunch::new(200);
+ bunch
+ .add(data_file("blob1.blob", 0, 100, 3, Some(vec!["payload"])))
+ .unwrap();
+
+ let err = bunch
+ .add(data_file("blob2.blob", 100, 100, 2, Some(vec!["payload2"])))
+ .unwrap_err();
+
+ assert!(
+ matches!(err, Error::DataInvalid { message, .. } if message.contains("same write columns"))
+ );
+ }
+
+ #[test]
+ fn test_blob_bunch_rejects_mixed_schema_ids() {
+ let mut bunch = BlobBunch::new(200);
+ bunch
+ .add(data_file("blob1.blob", 0, 100, 3, Some(vec!["payload"])))
+ .unwrap();
+
+ let mut mixed_schema = data_file("blob2.blob", 100, 100, 2, Some(vec!["payload"]));
+ mixed_schema.schema_id = 1;
+ let err = bunch.add(mixed_schema).unwrap_err();
+
+ assert!(
+ matches!(err, Error::DataInvalid { message, .. } if message.contains("same schema id"))
+ );
+ }
+
+ #[test]
+ fn test_blob_bunch_rejects_row_count_exceeding_expected() {
+ let mut bunch = BlobBunch::new(100);
+ bunch
+ .add(data_file("blob1.blob", 0, 60, 3, Some(vec!["payload"])))
+ .unwrap();
+
+ let err = bunch
+ .add(data_file("blob2.blob", 60, 50, 2, Some(vec!["payload"])))
+ .unwrap_err();
+
+ assert!(
+ matches!(err, Error::DataInvalid { message, .. } if message.contains("exceed the expected"))
+ );
+ }
+
+ #[test]
+ fn test_build_source_plan_picks_latest_blob_segments() {
+ let files = vec![
+ data_file("others.parquet", 0, 1000, 1, None),
+ data_file("blob1.blob", 0, 1000, 1, Some(vec!["payload"])),
+ data_file("blob2.blob", 0, 500, 2, Some(vec!["payload"])),
+ data_file("blob3.blob", 500, 250, 2, Some(vec!["payload"])),
+ data_file("blob4.blob", 750, 250, 2, Some(vec!["payload"])),
+ data_file("blob5.blob", 0, 100, 3, Some(vec!["payload"])),
+ data_file("blob6.blob", 100, 400, 3, Some(vec!["payload"])),
+ data_file("blob7.blob", 750, 100, 3, Some(vec!["payload"])),
+ data_file("blob8.blob", 850, 150, 3, Some(vec!["payload"])),
+ data_file("blob9.blob", 100, 650, 4, Some(vec!["payload"])),
+ ];
+ let prepared_group = PreparedMergeGroup::new(&files).unwrap();
+ let file_infos: Vec<ResolvedFileInfo> = prepared_group
+ .files
+ .iter()
+ .map(|file| {
+ if is_blob_file_name(&file.file_name) {
+ resolved_info(vec![2])
+ } else {
+ resolved_info(vec![1])
+ }
+ })
+ .collect();
+
+ let read_type = vec![
+ DataField::new(1, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(2, "payload".to_string(), DataType::Blob(BlobType::new())),
+ ];
+ let source_plan = build_source_plan(&prepared_group, &file_infos, &read_type).unwrap();
+
+ assert_eq!(source_plan.sources.len(), 2);
+ assert_eq!(source_plan.column_plan, vec![Some((0, 0)), Some((1, 0))]);
+
+ match &source_plan.sources[1] {
+ FieldSource::BlobBunch { bunch, .. } => {
+ let file_names: Vec<&str> = bunch
+ .files
+ .iter()
+ .map(|file| file.file_name.as_str())
+ .collect();
+ assert_eq!(
+ file_names,
+ vec!["blob5.blob", "blob9.blob", "blob7.blob", "blob8.blob"]
+ );
+ }
+ FieldSource::DataFile { .. } => panic!("expected blob bunch source"),
+ }
+ }
+
+ #[test]
+ fn test_build_source_plan_prefers_latest_normal_file_provider() {
+ let files = vec![
+ data_file("base-v1.parquet", 0, 4, 1, None),
+ data_file("base-v2.parquet", 0, 4, 2, None),
+ data_file("payload.blob", 0, 4, 2, Some(vec!["payload"])),
+ ];
+ let prepared_group = PreparedMergeGroup::new(&files).unwrap();
+ let file_infos = vec![
+ resolved_info(vec![1]),
+ resolved_info(vec![1]),
+ resolved_info(vec![2]),
+ ];
+ let read_type = vec![
+ DataField::new(1, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(2, "payload".to_string(), DataType::Blob(BlobType::new())),
+ ];
+
+ let source_plan = build_source_plan(&prepared_group, &file_infos, &read_type).unwrap();
+
+ assert_eq!(source_plan.column_plan, vec![Some((0, 0)), Some((2, 0))]);
+ }
+
+ #[test]
+ fn test_build_source_plan_groups_multiple_blob_columns() {
+ let files = vec![
+ data_file("others.parquet", 0, 1000, 1, None),
+ data_file("blob5.blob", 0, 100, 3, Some(vec!["payload"])),
+ data_file("blob9.blob", 100, 650, 4, Some(vec!["payload"])),
+ data_file("blob7.blob", 750, 100, 3, Some(vec!["payload"])),
+ data_file("blob8.blob", 850, 150, 3, Some(vec!["payload"])),
+ data_file("blob15.blob", 0, 100, 3, Some(vec!["payload2"])),
+ data_file("blob19.blob", 100, 650, 4, Some(vec!["payload2"])),
+ data_file("blob17.blob", 750, 100, 3, Some(vec!["payload2"])),
+ data_file("blob18.blob", 850, 150, 3, Some(vec!["payload2"])),
+ ];
+ let prepared_group = PreparedMergeGroup::new(&files).unwrap();
+ let file_infos: Vec<ResolvedFileInfo> = prepared_group
+ .files
+ .iter()
+ .map(
+ |file| match file.write_cols.as_ref().and_then(|cols| cols.first()) {
+ Some(name) if name == "payload" => resolved_info(vec![2]),
+ Some(name) if name == "payload2" => resolved_info(vec![3]),
+ _ => resolved_info(vec![1]),
+ },
+ )
+ .collect();
+
+ let read_type = vec![
+ DataField::new(1, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(2, "payload".to_string(), DataType::Blob(BlobType::new())),
+ DataField::new(3, "payload2".to_string(), DataType::Blob(BlobType::new())),
+ ];
+ let source_plan = build_source_plan(&prepared_group, &file_infos, &read_type).unwrap();
+
+ assert_eq!(source_plan.sources.len(), 3);
+ assert_eq!(
+ source_plan.column_plan,
+ vec![Some((0, 0)), Some((1, 0)), Some((2, 0))]
+ );
+
+ match &source_plan.sources[1] {
+ FieldSource::BlobBunch { bunch, .. } => {
+ let file_names: Vec<&str> = bunch
+ .files
+ .iter()
+ .map(|file| file.file_name.as_str())
+ .collect();
+ assert_eq!(
+ file_names,
+ vec!["blob5.blob", "blob9.blob", "blob7.blob", "blob8.blob"]
+ );
+ }
+ FieldSource::DataFile { .. } => panic!("expected blob bunch source"),
+ }
+
+ match &source_plan.sources[2] {
+ FieldSource::BlobBunch { bunch, .. } => {
+ let file_names: Vec<&str> = bunch
+ .files
+ .iter()
+ .map(|file| file.file_name.as_str())
+ .collect();
+ assert_eq!(
+ file_names,
+ vec!["blob15.blob", "blob19.blob", "blob17.blob", "blob18.blob"]
+ );
+ }
+ FieldSource::DataFile { .. } => panic!("expected blob bunch source"),
+ }
+ }
+
+ #[tokio::test]
+ async fn test_table_read_merges_parquet_and_java_rolling_blob_files() {
+ let tempdir = tempdir().unwrap();
+ let table_path = local_file_path(tempdir.path());
+ let bucket_dir = tempdir.path().join("bucket-0");
+ fs::create_dir_all(&bucket_dir).unwrap();
+
+ let parquet_path = bucket_dir.join("data.parquet");
+ write_int_parquet_file(&parquet_path, vec![("id", vec![1, 2, 3, 4])], None);
+
+ let blob_part1_path = bucket_dir.join("blob-part-1.blob");
+ let blob_part2_path = bucket_dir.join("blob-part-2.blob");
+ copy_blob_fixture("blob-part-1.blob", &blob_part1_path);
+ copy_blob_fixture("blob-part-2.blob", &blob_part2_path);
+
+ let file_io = FileIOBuilder::new("file").build().unwrap();
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Blob(BlobType::new()))
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap(),
+ );
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "blob_t"),
+ table_path,
+ table_schema,
+ None,
+ );
+
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(local_file_path(&bucket_dir))
+ .with_total_buckets(1)
+ .with_data_files(vec![
+ data_file_meta_with_path(
+ "data.parquet",
+ 0,
+ 4,
+ 1,
+ parquet_path.metadata().unwrap().len() as i64,
+ Some(vec!["id"]),
+ ),
+ data_file_meta_with_path(
+ "blob-part-1.blob",
+ 0,
+ 2,
+ 1,
+ blob_part1_path.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ ),
+ data_file_meta_with_path(
+ "blob-part-2.blob",
+ 2,
+ 2,
+ 1,
+ blob_part2_path.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ ),
+ ])
+ .with_raw_convertible(false)
+ .build()
+ .unwrap();
+
+ let read = TableRead::new(&table, table.schema().fields().to_vec(), Vec::new());
+ let batches = read
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(collect_int_values(&batches, "id"), vec![1, 2, 3, 4]);
+ assert_eq!(
+ collect_binary_values(&batches, "payload"),
+ vec![
+ Some(b"hello".to_vec()),
+ None,
+ Some(b"world".to_vec()),
+ Some(Vec::new()),
+ ]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_table_read_merges_multiple_blob_columns_with_row_ranges() {
+ let tempdir = tempdir().unwrap();
+ let table_path = local_file_path(tempdir.path());
+ let bucket_dir = tempdir.path().join("bucket-0");
+ fs::create_dir_all(&bucket_dir).unwrap();
+
+ let parquet_path = bucket_dir.join("data.parquet");
+ write_int_parquet_file(&parquet_path, vec![("id", vec![1, 2, 3, 4])], None);
+
+ let payload_a_1 = bucket_dir.join("payload-a-1.blob");
+ let payload_a_2 = bucket_dir.join("payload-a-2.blob");
+ let payload_b_1 = bucket_dir.join("payload-b-1.blob");
+ let payload_b_2 = bucket_dir.join("payload-b-2.blob");
+ write_blob_file(&payload_a_1, &[Some(&b"a1"[..]), Some(&b"a2"[..])]);
+ write_blob_file(&payload_a_2, &[Some(&b"a3"[..]), Some(&b"a4"[..])]);
+ write_blob_file(&payload_b_1, &[Some(&b"b1"[..]), Some(&b"b2"[..])]);
+ write_blob_file(&payload_b_2, &[Some(&b"b3"[..]), Some(&b"b4"[..])]);
+
+ let file_io = FileIOBuilder::new("file").build().unwrap();
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Blob(BlobType::new()))
+ .column("payload2", DataType::Blob(BlobType::new()))
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap(),
+ );
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "blob_multi_t"),
+ table_path,
+ table_schema,
+ None,
+ );
+
+ let split = DataSplitBuilder::new()
+ .with_snapshot(1)
+ .with_partition(BinaryRow::new(0))
+ .with_bucket(0)
+ .with_bucket_path(local_file_path(&bucket_dir))
+ .with_total_buckets(1)
+ .with_data_files(vec![
+ data_file_meta_with_path(
+ "data.parquet",
+ 0,
+ 4,
+ 1,
+ parquet_path.metadata().unwrap().len() as i64,
+ Some(vec!["id"]),
+ ),
+ data_file_meta_with_path(
+ "payload-a-1.blob",
+ 0,
+ 2,
+ 1,
+ payload_a_1.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ ),
+ data_file_meta_with_path(
+ "payload-a-2.blob",
+ 2,
+ 2,
+ 1,
+ payload_a_2.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ ),
+ data_file_meta_with_path(
+ "payload-b-1.blob",
+ 0,
+ 2,
+ 1,
+ payload_b_1.metadata().unwrap().len() as i64,
+ Some(vec!["payload2"]),
+ ),
+ data_file_meta_with_path(
+ "payload-b-2.blob",
+ 2,
+ 2,
+ 1,
+ payload_b_2.metadata().unwrap().len() as i64,
+ Some(vec!["payload2"]),
+ ),
+ ])
+ .with_raw_convertible(false)
+ .with_row_ranges(vec![RowRange::new(1, 2)])
+ .build()
+ .unwrap();
+
+ let read = TableRead::new(&table, table.schema().fields().to_vec(), Vec::new());
+ let batches = read
+ .to_arrow(&[split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(collect_int_values(&batches, "id"), vec![2, 3]);
+ assert_eq!(
+ collect_binary_values(&batches, "payload"),
+ vec![Some(b"a2".to_vec()), Some(b"a3".to_vec())]
+ );
+ assert_eq!(
+ collect_binary_values(&batches, "payload2"),
+ vec![Some(b"b2".to_vec()), Some(b"b3".to_vec())]
+ );
+ }
+
+ fn resolved_info(field_ids: Vec<i32>) -> ResolvedFileInfo {
+ ResolvedFileInfo {
+ field_ids,
+ data_fields: None,
+ }
+ }
+
+ fn data_file(
+ file_name: &str,
+ first_row_id: i64,
+ row_count: i64,
+ max_sequence_number: i64,
+ write_cols: Option<Vec<&str>>,
+ ) -> DataFileMeta {
+ DataFileMeta {
+ file_name: file_name.to_string(),
+ file_size: 0,
+ row_count,
+ min_key: Vec::new(),
+ max_key: Vec::new(),
+ key_stats: BinaryTableStats::new(Vec::new(), Vec::new(), Vec::new()),
+ value_stats: BinaryTableStats::new(Vec::new(), Vec::new(), Vec::new()),
+ min_sequence_number: 0,
+ max_sequence_number,
+ schema_id: 0,
+ level: 0,
+ extra_files: Vec::new(),
+ creation_time: None,
+ delete_row_count: None,
+ embedded_index: None,
+ file_source: None,
+ value_stats_cols: None,
+ external_path: None,
+ first_row_id: Some(first_row_id),
+ write_cols: write_cols.map(|cols| cols.into_iter().map(str::to_string).collect()),
+ }
+ }
+
+ fn data_file_meta_with_path(
+ file_name: &str,
+ first_row_id: i64,
+ row_count: i64,
+ max_sequence_number: i64,
+ file_size: i64,
+ write_cols: Option<Vec<&str>>,
+ ) -> DataFileMeta {
+ let mut file = data_file(
+ file_name,
+ first_row_id,
+ row_count,
+ max_sequence_number,
+ write_cols,
+ );
+ file.file_size = file_size;
+ file
+ }
+
+ fn copy_blob_fixture(name: &str, destination: &Path) {
+ let source = blob_fixture_path(name);
+ fs::copy(&source, destination).unwrap_or_else(|e| {
+ panic!("Failed to copy blob fixture {source:?} -> {destination:?}: {e}")
+ });
+ }
+
+ fn blob_fixture_path(name: &str) -> PathBuf {
+ PathBuf::from(env!("CARGO_MANIFEST_DIR")).join(format!("testdata/blob/{name}"))
+ }
+
+ fn collect_int_values(batches: &[RecordBatch], column_name: &str) -> Vec<i32> {
+ batches
+ .iter()
+ .flat_map(|batch| {
+ let idx = batch.schema().index_of(column_name).unwrap();
+ let array = batch
+ .column(idx)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ (0..array.len())
+ .map(|row| array.value(row))
+ .collect::<Vec<_>>()
+ })
+ .collect()
+ }
+
+ fn collect_binary_values(batches: &[RecordBatch], column_name: &str) -> Vec<Option<Vec<u8>>> {
+ batches
+ .iter()
+ .flat_map(|batch| {
+ let idx = batch.schema().index_of(column_name).unwrap();
+ let array = batch
+ .column(idx)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ (0..array.len())
+ .map(|row| (!array.is_null(row)).then(|| array.value(row).to_vec()))
+ .collect::<Vec<_>>()
+ })
+ .collect()
}
}
diff --git a/crates/paimon/testdata/blob/blob-basic.blob b/crates/paimon/testdata/blob/blob-basic.blob
new file mode 100644
index 0000000..c896202
Binary files /dev/null and b/crates/paimon/testdata/blob/blob-basic.blob differ
diff --git a/crates/paimon/testdata/blob/blob-part-1.blob b/crates/paimon/testdata/blob/blob-part-1.blob
new file mode 100644
index 0000000..af7eb84
Binary files /dev/null and b/crates/paimon/testdata/blob/blob-part-1.blob differ
diff --git a/crates/paimon/testdata/blob/blob-part-2.blob b/crates/paimon/testdata/blob/blob-part-2.blob
new file mode 100644
index 0000000..40549d9
Binary files /dev/null and b/crates/paimon/testdata/blob/blob-part-2.blob differ