This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 a811c2f fix(blob): support placeholder fallback reads (#530)
a811c2f is described below
commit a811c2ff41ca0b7dba5505b2fa843ee88ff4f74d
Author: QuakeWang <[email protected]>
AuthorDate: Sat Jul 18 10:21:36 2026 +0800
fix(blob): support placeholder fallback reads (#530)
---
crates/blob_test_utils.rs | 29 +-
crates/paimon/src/arrow/format/blob.rs | 304 ++++---
crates/paimon/src/table/data_evolution_reader.rs | 897 ++++++++++++++++++---
.../table/data_evolution_reader/blob_fallback.rs | 432 ++++++++++
crates/paimon/testdata/blob/blob-placeholder.blob | Bin 0 -> 33 bytes
5 files changed, 1420 insertions(+), 242 deletions(-)
diff --git a/crates/blob_test_utils.rs b/crates/blob_test_utils.rs
index 24fcae0..e878783 100644
--- a/crates/blob_test_utils.rs
+++ b/crates/blob_test_utils.rs
@@ -22,13 +22,31 @@ const BLOB_MAGIC_NUMBER_BYTES: [u8; 4] =
1481511375_i32.to_le_bytes();
const BLOB_ENTRY_OVERHEAD: usize = 16;
const BLOB_FORMAT_VERSION: u8 = 1;
+#[derive(Clone, Copy)]
+pub(crate) enum BlobFixtureValue<'a> {
+ Value(&'a [u8]),
+ Null,
+ Placeholder,
+}
+
pub(crate) fn build_blob_file_bytes(rows: &[Option<&[u8]>]) -> Vec<u8> {
+ let values = rows
+ .iter()
+ .map(|row| match row {
+ Some(payload) => BlobFixtureValue::Value(payload),
+ None => BlobFixtureValue::Null,
+ })
+ .collect::<Vec<_>>();
+ build_blob_file_bytes_with_values(&values)
+}
+
+pub(crate) fn build_blob_file_bytes_with_values(rows: &[BlobFixtureValue<'_>])
-> Vec<u8> {
let mut file_bytes = Vec::new();
let mut lengths = Vec::with_capacity(rows.len());
for row in rows {
match row {
- Some(payload) => {
+ BlobFixtureValue::Value(payload) => {
let entry_length = payload
.len()
.checked_add(BLOB_ENTRY_OVERHEAD)
@@ -49,7 +67,8 @@ pub(crate) fn build_blob_file_bytes(rows: &[Option<&[u8]>])
-> Vec<u8> {
hasher.update(&entry_length_bytes);
file_bytes.extend_from_slice(&hasher.finalize().to_le_bytes());
}
- None => lengths.push(-1),
+ BlobFixtureValue::Null => lengths.push(-1),
+ BlobFixtureValue::Placeholder => lengths.push(-2),
}
}
@@ -72,6 +91,12 @@ pub(crate) fn write_blob_file(path: &Path, rows:
&[Option<&[u8]>]) {
.unwrap_or_else(|e| panic!("Failed to write blob test file {path:?}:
{e}"));
}
+pub(crate) fn write_blob_file_with_values(path: &Path, rows:
&[BlobFixtureValue<'_>]) {
+ let file_bytes = build_blob_file_bytes_with_values(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();
diff --git a/crates/paimon/src/arrow/format/blob.rs
b/crates/paimon/src/arrow/format/blob.rs
index c401a06..f66108c 100644
--- a/crates/paimon/src/arrow/format/blob.rs
+++ b/crates/paimon/src/arrow/format/blob.rs
@@ -44,6 +44,53 @@ impl BlobFormatReader {
}
}
+pub(crate) struct IndexedBlobReader {
+ reader: Box<dyn FileRead>,
+ index: BlobFileIndex,
+ descriptor_mode: bool,
+ file_path: String,
+}
+
+impl IndexedBlobReader {
+ pub(crate) async fn open(
+ reader: Box<dyn FileRead>,
+ file_size: u64,
+ file_path: String,
+ descriptor_mode: bool,
+ ) -> crate::Result<Self> {
+ let index = BlobFileIndex::load(reader.as_ref(), file_size).await?;
+ Ok(Self {
+ reader,
+ index,
+ descriptor_mode,
+ file_path,
+ })
+ }
+
+ pub(crate) fn num_rows(&self) -> usize {
+ self.index.num_rows()
+ }
+
+ pub(crate) async fn read_positions(
+ &self,
+ positions: &[usize],
+ ) -> crate::Result<Vec<BlobReadValue>> {
+ if self.descriptor_mode {
+ build_descriptor_values(&self.index, positions, &self.file_path)
+ } else {
+ let planned_reads = plan_blob_reads(&self.index, positions)?;
+ fetch_blob_values(self.reader.as_ref(), planned_reads).await
+ }
+ }
+}
+
+#[derive(Debug)]
+pub(crate) enum BlobReadValue {
+ Value(Bytes),
+ Null,
+ Placeholder,
+}
+
const BLOB_FOOTER_SIZE: u64 = 5;
const BLOB_FORMAT_VERSION: u8 = 1;
const BLOB_INLINE_HEADER_SIZE: u64 = 4;
@@ -67,46 +114,36 @@ impl FormatFileReader for BlobFormatReader {
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 blob_reader = IndexedBlobReader::open(
+ reader,
+ file_size,
+ self.file_path.clone(),
+ self.descriptor_mode,
+ )
+ .await?;
+ let mut selection = RowSelectionCursor::new(blob_reader.num_rows(),
row_selection)?;
let project_values = !read_fields.is_empty();
- if self.descriptor_mode {
- let file_path = self.file_path.clone();
- Ok(try_stream! {
- while let Some(positions) = selection.next_batch(batch_size) {
- let batch = if project_values {
- build_descriptor_batch(&blob_index, &target_schema,
&positions, &file_path)?
- } else {
- 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)),
- })?
- };
- yield batch;
- }
- }
- .boxed())
- } else {
- 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;
- }
+ Ok(try_stream! {
+ while let Some(positions) = selection.next_batch(batch_size) {
+ let batch = if project_values {
+ let values = blob_reader.read_positions(&positions).await?;
+ build_blob_batch(&target_schema, values)?
+ } else {
+ 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)),
+ })?
+ };
+ yield batch;
}
- .boxed())
}
+ .boxed())
}
}
@@ -138,70 +175,49 @@ fn validate_read_fields(read_fields: &[DataField]) ->
crate::Result<()> {
Ok(())
}
-fn build_descriptor_batch(
+fn build_descriptor_values(
blob_index: &BlobFileIndex,
- target_schema: &Arc<arrow_schema::Schema>,
positions: &[usize],
file_path: &str,
-) -> crate::Result<RecordBatch> {
- let mut builder = BinaryBuilder::new();
- for &position in positions {
- 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,
- })?;
-
- match entry.inline_data_range() {
- None => builder.append_null(),
- Some(range) => {
- let descriptor = BlobDescriptor::new(
- file_path.to_string(),
- range.start as i64,
- (range.end - range.start) as i64,
- );
- builder.append_value(descriptor.serialize());
- }
- }
- }
+) -> crate::Result<Vec<BlobReadValue>> {
+ 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,
+ })?;
- 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 descriptor blob RecordBatch: {e}"),
- source: Some(Box::new(e)),
- })
+ Ok(match entry {
+ BlobEntry::Value(range) => {
+ let descriptor = BlobDescriptor::new(
+ file_path.to_string(),
+ range.start as i64,
+ (range.end - range.start) as i64,
+ );
+ BlobReadValue::Value(Bytes::from(descriptor.serialize()))
+ }
+ BlobEntry::Null => BlobReadValue::Null,
+ BlobEntry::Placeholder => BlobReadValue::Placeholder,
+ })
+ })
+ .collect()
}
-async fn read_blob_batch(
- reader: &dyn FileRead,
- blob_index: &BlobFileIndex,
+fn build_blob_batch(
target_schema: &Arc<arrow_schema::Schema>,
- positions: &[usize],
- project_values: bool,
+ values: Vec<BlobReadValue>,
) -> 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()),
+ BlobReadValue::Value(bytes) =>
builder.append_value(bytes.as_ref()),
+ BlobReadValue::Null | BlobReadValue::Placeholder =>
builder.append_null(),
}
}
@@ -229,10 +245,11 @@ fn plan_blob_reads(
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,
+ Ok(match entry {
+ BlobEntry::Value(range) if range.start == range.end =>
PlannedBlobRead::Empty,
+ BlobEntry::Value(range) =>
PlannedBlobRead::Read(range.clone()),
+ BlobEntry::Null => PlannedBlobRead::Null,
+ BlobEntry::Placeholder => PlannedBlobRead::Placeholder,
})
})
.collect()
@@ -241,12 +258,13 @@ fn plan_blob_reads(
async fn fetch_blob_values(
reader: &dyn FileRead,
planned_reads: Vec<PlannedBlobRead>,
-) -> crate::Result<Vec<BlobValue>> {
+) -> crate::Result<Vec<BlobReadValue>> {
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),
+ PlannedBlobRead::Null => Ok(BlobReadValue::Null),
+ PlannedBlobRead::Placeholder => Ok(BlobReadValue::Placeholder),
+ PlannedBlobRead::Empty => Ok(BlobReadValue::Value(Bytes::new())),
+ PlannedBlobRead::Read(range) =>
reader.read(range).await.map(BlobReadValue::Value),
}
}))
.buffered(BLOB_READ_CONCURRENCY)
@@ -257,16 +275,11 @@ async fn fetch_blob_values(
#[derive(Debug, Clone)]
enum PlannedBlobRead {
Null,
+ Placeholder,
Empty,
Read(Range<u64>),
}
-#[derive(Debug, Clone)]
-enum BlobValue {
- Null,
- Inline(Bytes),
-}
-
#[derive(Debug, Clone)]
struct BlobFileIndex {
entries: Vec<BlobEntry>,
@@ -348,9 +361,10 @@ impl BlobFileIndex {
}
#[derive(Debug, Clone)]
-struct BlobEntry {
- data_offset: Option<u64>,
- data_length: u64,
+enum BlobEntry {
+ Value(Range<u64>),
+ Null,
+ Placeholder,
}
impl BlobEntry {
@@ -359,16 +373,22 @@ impl BlobEntry {
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;
+ match entry_length {
+ -1 => {
+ entries.push(Self::Null);
+ continue;
+ }
+ -2 => {
+ entries.push(Self::Placeholder);
+ 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}"),
+ message: format!(
+ "Blob entry length must be positive, -1, or -2, got
{entry_length}"
+ ),
source: Some(Box::new(e)),
})?;
@@ -397,20 +417,14 @@ impl BlobEntry {
});
}
- entries.push(Self {
- data_offset: Some(next_offset + BLOB_INLINE_HEADER_SIZE),
- data_length: entry_length - BLOB_ENTRY_OVERHEAD,
- });
+ let data_offset = next_offset + BLOB_INLINE_HEADER_SIZE;
+ let data_length = entry_length - BLOB_ENTRY_OVERHEAD;
+ entries.push(Self::Value(data_offset..data_offset + data_length));
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)]
@@ -915,6 +929,38 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_blob_reader_treats_java_placeholders_as_null() {
+ let read_fields = vec![DataField::new(
+ 0,
+ "payload".to_string(),
+ DataType::Blob(BlobType::new()),
+ )];
+ let file_bytes = load_blob_fixture("blob-placeholder.blob");
+
+ let batches = BlobFormatReader::new(String::new(), false)
+ .read_batch_stream(
+ Box::new(BytesFileRead(Bytes::from(file_bytes.clone()))),
+ file_bytes.len() as u64,
+ &read_fields,
+ None,
+ Some(2),
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(batches.len(), 2);
+ assert_eq!(collect_binary_values(&batches[0]), vec![None, None]);
+ assert_eq!(
+ collect_binary_values(&batches[1]),
+ vec![Some(b"latest-3".to_vec()), None]
+ );
+ }
+
#[tokio::test]
async fn test_blob_reader_reads_payloads_with_bounded_parallelism() {
let read_fields = vec![DataField::new(
@@ -960,6 +1006,20 @@ mod tests {
assert_eq!(generated, load_blob_fixture("blob-basic.blob"));
}
+ #[test]
+ fn test_blob_reader_test_helper_matches_java_placeholder_fixture() {
+ use blob_test_utils::BlobFixtureValue::{Null, Placeholder, Value};
+
+ let generated = blob_test_utils::build_blob_file_bytes_with_values(&[
+ Placeholder,
+ Null,
+ Value(b"latest-3"),
+ Placeholder,
+ ]);
+
+ assert_eq!(generated, load_blob_fixture("blob-placeholder.blob"));
+ }
+
#[tokio::test]
async fn test_blob_reader_supports_empty_projection() {
let reader = BlobFormatReader::new(String::new(), false);
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 1f0e9af..1d5a130 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+mod blob_fallback;
+
use super::blob_resolver::{BlobReadLimiter, BLOB_DESCRIPTOR_READ_CONCURRENCY};
use super::data_file_reader::{
append_null_row_id_column, attach_row_id, expand_selected_row_ids,
insert_column_at,
@@ -36,7 +38,8 @@ use arrow_array::{Array, BinaryArray, Int64Array,
RecordBatch};
use async_stream::try_stream;
use futures::{StreamExt, TryStreamExt};
use roaring::RoaringBitmap;
-use std::collections::{HashMap, HashSet};
+use std::cmp::Reverse;
+use std::collections::{BTreeMap, HashMap, HashSet};
use std::future::Future;
use std::sync::Arc;
@@ -1109,6 +1112,21 @@ fn open_source_stream(
blob_as_descriptor: bool,
anchor_deletion_vector: Option<&DeletionVectorContext>,
) -> crate::Result<ArrowRecordBatchStream> {
+ if let FieldSource::BlobBunch { bunch, read_fields } = source {
+ // A single sequence group has no fallback work, so keep per-file lazy
streaming.
+ if !bunch.can_read_sequentially() {
+ return blob_fallback::read(
+ split,
+ bunch.clone(),
+ read_fields.clone(),
+ row_ranges,
+ file_io,
+ blob_as_descriptor,
+ anchor_deletion_vector.cloned(),
+ );
+ }
+ }
+
let file_reader = DataFileReader::new(
file_io,
schema_manager,
@@ -1132,16 +1150,26 @@ fn open_source_stream(
row_ranges,
)
}
- FieldSource::BlobBunch {
- bunch, data_fields, ..
- } => read_bunch_files_stream(
- file_reader,
- split,
- bunch.files.clone(),
- data_fields.clone(),
- row_ranges,
- anchor_deletion_vector.cloned(),
- ),
+ FieldSource::BlobBunch { bunch, .. } => {
+ let selected_ranges = selected_absolute_row_ranges_for_file(
+ bunch.expected_first_row_id,
+ bunch.expected_row_count,
+ row_ranges.as_deref(),
+ anchor_deletion_vector.map(|context|
context.deletion_vector.as_ref()),
+ )?;
+ let files = match selected_ranges.as_deref() {
+ Some(ranges) => bunch.files_overlapping(ranges)?,
+ None => bunch.files.clone(),
+ };
+ read_bunch_files_stream(
+ file_reader,
+ split,
+ files,
+ None,
+ row_ranges,
+ anchor_deletion_vector.cloned(),
+ )
+ }
FieldSource::VectorBunch {
bunch, data_fields, ..
} => read_bunch_files_stream(
@@ -1550,8 +1578,7 @@ fn build_source_plan(
} else {
let source_idx = sources.len();
sources.push(FieldSource::BlobBunch {
- bunch: BlobBunch::new(expected_row_count),
- data_fields: info.data_fields.clone(),
+ bunch: BlobBunch::new(prepared_group.first_row_id,
expected_row_count),
read_fields: Vec::new(),
});
blob_source_indices.insert(field_id, source_idx);
@@ -1667,15 +1694,8 @@ fn build_source_plan(
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,
- });
+ if !read_fields.is_empty() {
+ bunch.validate_logical_range()?;
}
}
}
@@ -1733,7 +1753,6 @@ enum FieldSource {
},
BlobBunch {
bunch: BlobBunch,
- data_fields: Option<Vec<DataField>>,
read_fields: Vec<DataField>,
},
}
@@ -1779,25 +1798,20 @@ impl FieldSource {
}
}
+/// All physical BLOB files for one field, including overlapping older
sequence groups.
#[derive(Debug, Clone)]
struct BlobBunch {
files: Vec<DataFileMeta>,
+ expected_first_row_id: i64,
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 {
+ fn new(expected_first_row_id: i64, expected_row_count: i64) -> Self {
Self {
files: Vec::new(),
+ expected_first_row_id,
expected_row_count,
- latest_first_row_id: -1,
- expected_next_first_row_id: -1,
- latest_max_sequence_number: -1,
- row_count: 0,
}
}
@@ -1809,85 +1823,166 @@ impl BlobBunch {
});
}
- 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 {
+ let range = blob_file_row_range(&file)?;
+ if let Some(first_file) = self.files.first() {
+ if file.write_cols != first_file.write_cols {
return Err(Error::DataInvalid {
- message:
- "Blob file with same first row id should have
decreasing sequence number."
- .to_string(),
+ message: "All files in a blob bunch should have the same
write columns."
+ .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 {
+ for existing in self
+ .files
+ .iter()
+ .filter(|existing| existing.max_sequence_number ==
file.max_sequence_number)
+ {
+ let existing_range = blob_file_row_range(existing)?;
+ if range.overlaps_inclusive(existing_range.from(),
existing_range.to()) {
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
+ "Blob files '{}' and '{}' in the same max sequence
group overlap",
+ existing.file_name, file.file_name
),
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,
- });
- }
- }
}
- self.row_count += file.row_count;
- if self.row_count > self.expected_row_count {
+ let mut ranges = self.logical_ranges();
+ ranges.push(range);
+ let row_count = crate::table::merge_row_ranges(ranges)
+ .iter()
+ .map(RowRange::count)
+ .sum::<i64>();
+ if 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
+ "Blob files logical row count {row_count} exceeds the
expected {}",
+ 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
+ self.logical_ranges().iter().map(RowRange::count).sum()
+ }
+
+ fn logical_ranges(&self) -> Vec<RowRange> {
+ crate::table::merge_row_ranges(
+ self.files
+ .iter()
+ .map(|file| blob_file_row_range(file).expect("validated blob
file range"))
+ .collect(),
+ )
+ }
+
+ fn expected_range(&self) -> crate::Result<RowRange> {
+ if self.expected_row_count <= 0 {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob bunch expected row count must be positive, got {}",
+ self.expected_row_count
+ ),
+ source: None,
+ });
+ }
+ let to = self
+ .expected_first_row_id
+ .checked_add(self.expected_row_count - 1)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Blob bunch expected row range overflows
i64".to_string(),
+ source: None,
+ })?;
+ Ok(RowRange::new(self.expected_first_row_id, to))
+ }
+
+ fn validate_logical_range(&self) -> crate::Result<()> {
+ let ranges = self.logical_ranges();
+ let expected = self.expected_range()?;
+ if ranges.as_slice() != [expected.clone()] {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob bunch logical row ranges {ranges:?} ({} rows) do not
match expected range {expected:?}",
+ self.row_count()
+ ),
+ source: None,
+ });
+ }
+ Ok(())
+ }
+
+ fn sequence_groups(&self) -> Vec<Vec<DataFileMeta>> {
+ let mut groups: BTreeMap<Reverse<i64>, Vec<DataFileMeta>> =
BTreeMap::new();
+ for file in &self.files {
+ groups
+ .entry(Reverse(file.max_sequence_number))
+ .or_default()
+ .push(file.clone());
+ }
+ for files in groups.values_mut() {
+ files.sort_by_key(|file| file.first_row_id.expect("validated blob
first_row_id"));
+ }
+ groups.into_values().collect()
+ }
+
+ fn can_read_sequentially(&self) -> bool {
+ let Some(first_file) = self.files.first() else {
+ return false;
+ };
+ self.files
+ .iter()
+ .all(|file| file.max_sequence_number ==
first_file.max_sequence_number)
+ }
+
+ fn files_overlapping(&self, ranges: &[RowRange]) ->
crate::Result<Vec<DataFileMeta>> {
+ let mut files = Vec::new();
+ for file in &self.files {
+ let file_range = blob_file_row_range(file)?;
+ if row_range_overlaps_any(&file_range, ranges) {
+ files.push(file.clone());
+ }
+ }
+ Ok(files)
}
}
+fn row_range_overlaps_any(range: &RowRange, ranges: &[RowRange]) -> bool {
+ ranges
+ .iter()
+ .any(|selected| range.overlaps_inclusive(selected.from(),
selected.to()))
+}
+
+fn blob_file_row_range(file: &DataFileMeta) -> crate::Result<RowRange> {
+ 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 file.row_count <= 0 {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob file '{}' row count must be positive, got {}",
+ file.file_name, file.row_count
+ ),
+ source: None,
+ });
+ }
+ let last_row_id = first_row_id
+ .checked_add(file.row_count - 1)
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!("Blob file '{}' row range overflows i64",
file.file_name),
+ source: None,
+ })?;
+ Ok(RowRange::new(first_row_id, last_row_id))
+}
+
/// Aggregates rolled `.vector.<format>` segments belonging to one logical
vector
/// source, mirroring upstream `VectorFileBunch` non-pushdown semantics. Unlike
/// `BlobBunch`, the expected row count is taken directly from the prepared
group's
@@ -2101,10 +2196,11 @@ mod tests {
BinaryRow, BlobType, Datum, FloatType, IntType, PredicateBuilder,
Schema, TableSchema,
VectorType,
};
- use crate::table::{DataSplitBuilder, Table, TableRead};
+ use crate::table::{DataSplitBuilder, DeletionFile, Table, TableRead};
use arrow_array::{
Array, BinaryArray, FixedSizeListArray, Float32Array, Int32Array,
Int64Array, RecordBatch,
};
+ use bytes::Bytes;
use futures::TryStreamExt;
use std::fs;
use std::path::{Path, PathBuf};
@@ -2122,7 +2218,7 @@ mod tests {
include!(concat!(env!("CARGO_MANIFEST_DIR"), "/../test_utils.rs"));
}
- use blob_test_utils::write_blob_file;
+ use blob_test_utils::{write_blob_file, write_blob_file_with_values,
BlobFixtureValue};
use test_utils::{local_file_path, write_int_parquet_file};
#[tokio::test]
@@ -2421,8 +2517,8 @@ mod tests {
}
#[test]
- fn test_blob_bunch_ignores_same_first_row_id_with_lower_sequence() {
- let mut bunch = BlobBunch::new(1000);
+ fn test_blob_bunch_retains_same_range_from_older_sequence() {
+ let mut bunch = BlobBunch::new(0, 1000);
bunch
.add(data_file(
"blob-high.blob",
@@ -2437,8 +2533,9 @@ mod tests {
.unwrap();
assert_eq!(bunch.row_count(), 100);
- assert_eq!(bunch.files.len(), 1);
+ assert_eq!(bunch.files.len(), 2);
assert_eq!(bunch.files[0].file_name, "blob-high.blob");
+ assert_eq!(bunch.files[1].file_name, "blob-low.blob");
}
#[test]
@@ -2467,13 +2564,12 @@ mod tests {
}
#[test]
- fn test_blob_bunch_rejects_same_first_row_id_with_higher_sequence() {
- let mut bunch = BlobBunch::new(1000);
+ fn test_blob_bunch_groups_sequences_in_descending_order() {
+ let mut bunch = BlobBunch::new(0, 1000);
bunch
.add(data_file("blob-low.blob", 0, 100, 2, Some(vec!["payload"])))
.unwrap();
-
- let err = bunch
+ bunch
.add(data_file(
"blob-high.blob",
0,
@@ -2481,48 +2577,63 @@ mod tests {
3,
Some(vec!["payload"]),
))
- .unwrap_err();
+ .unwrap();
- assert!(
- matches!(err, Error::DataInvalid { message, .. } if
message.contains("same first row id"))
- );
+ let groups = bunch.sequence_groups();
+ assert_eq!(groups.len(), 2);
+ assert_eq!(groups[0][0].file_name, "blob-high.blob");
+ assert_eq!(groups[1][0].file_name, "blob-low.blob");
}
#[test]
- fn test_blob_bunch_rejects_overlapping_higher_sequence_file() {
- let mut bunch = BlobBunch::new(1000);
+ fn test_blob_bunch_retains_overlapping_ranges_across_sequences() {
+ let mut bunch = BlobBunch::new(0, 1000);
bunch
.add(data_file("blob1.blob", 0, 100, 1, Some(vec!["payload"])))
.unwrap();
+ bunch
+ .add(data_file("blob2.blob", 50, 150, 2, Some(vec!["payload"])))
+ .unwrap();
+
+ assert_eq!(bunch.files.len(), 2);
+ assert_eq!(bunch.row_count(), 200);
+ assert_eq!(bunch.logical_ranges(), vec![RowRange::new(0, 199)]);
+ }
+ #[test]
+ fn test_blob_bunch_rejects_overlapping_ranges_within_sequence() {
+ let mut bunch = BlobBunch::new(0, 1000);
+ bunch
+ .add(data_file("blob1.blob", 0, 100, 2, 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"))
+ matches!(err, Error::DataInvalid { message, .. } if
message.contains("same max sequence group"))
);
}
#[test]
- fn test_blob_bunch_rejects_non_continuous_first_row_id() {
- let mut bunch = BlobBunch::new(1000);
+ fn test_blob_bunch_rejects_non_contiguous_logical_range() {
+ let mut bunch = BlobBunch::new(0, 250);
bunch
.add(data_file("blob1.blob", 0, 100, 3, Some(vec!["payload"])))
.unwrap();
-
- let err = bunch
+ bunch
.add(data_file("blob2.blob", 150, 100, 2, Some(vec!["payload"])))
- .unwrap_err();
+ .unwrap();
+ let err = bunch.validate_logical_range().unwrap_err();
assert!(
- matches!(err, Error::DataInvalid { message, .. } if
message.contains("continuous"))
+ matches!(err, Error::DataInvalid { message, .. } if
message.contains("logical row ranges"))
);
}
#[test]
fn test_blob_bunch_rejects_mixed_write_columns() {
- let mut bunch = BlobBunch::new(200);
+ let mut bunch = BlobBunch::new(0, 200);
bunch
.add(data_file("blob1.blob", 0, 100, 3, Some(vec!["payload"])))
.unwrap();
@@ -2537,24 +2648,27 @@ mod tests {
}
#[test]
- fn test_blob_bunch_rejects_mixed_schema_ids() {
- let mut bunch = BlobBunch::new(200);
+ fn test_blob_bunch_accepts_mixed_schema_ids() {
+ let mut bunch = BlobBunch::new(0, 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"]));
+ let mut mixed_schema = data_file("blob2.blob", 100, 100, 3,
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"))
- );
+ bunch.add(mixed_schema).unwrap();
+
+ assert_eq!(bunch.files.len(), 2);
+ assert_eq!(bunch.files[0].schema_id, 0);
+ assert_eq!(bunch.files[1].schema_id, 1);
+ assert_eq!(bunch.row_count(), 200);
+ assert!(bunch.can_read_sequentially());
+ bunch.validate_logical_range().unwrap();
}
#[test]
fn test_blob_bunch_rejects_row_count_exceeding_expected() {
- let mut bunch = BlobBunch::new(100);
+ let mut bunch = BlobBunch::new(0, 100);
bunch
.add(data_file("blob1.blob", 0, 60, 3, Some(vec!["payload"])))
.unwrap();
@@ -2564,7 +2678,7 @@ mod tests {
.unwrap_err();
assert!(
- matches!(err, Error::DataInvalid { message, .. } if
message.contains("exceed the expected"))
+ matches!(err, Error::DataInvalid { message, .. } if
message.contains("exceeds the expected"))
);
}
@@ -2767,7 +2881,7 @@ mod tests {
}
#[test]
- fn test_build_source_plan_picks_latest_blob_segments() {
+ fn test_build_source_plan_retains_all_blob_sequence_groups() {
let files = vec![
data_file("others.parquet", 0, 1000, 1, None),
data_file("blob1.blob", 0, 1000, 1, Some(vec!["payload"])),
@@ -2812,7 +2926,17 @@ mod tests {
.collect();
assert_eq!(
file_names,
- vec!["blob5.blob", "blob9.blob", "blob7.blob",
"blob8.blob"]
+ vec![
+ "blob5.blob",
+ "blob2.blob",
+ "blob1.blob",
+ "blob9.blob",
+ "blob6.blob",
+ "blob3.blob",
+ "blob7.blob",
+ "blob4.blob",
+ "blob8.blob",
+ ]
);
}
FieldSource::DataFile { .. } | FieldSource::VectorBunch { .. } => {
@@ -3008,6 +3132,501 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_table_read_falls_back_across_blob_sequence_groups() {
+ use BlobFixtureValue::{Null, Placeholder, Value};
+
+ 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, 5,
6])], None);
+
+ let base_path = bucket_dir.join("blob-base.blob");
+ let middle_path = bucket_dir.join("blob-middle.blob");
+ let latest_path = bucket_dir.join("blob-latest.blob");
+ write_blob_file_with_values(
+ &base_path,
+ &[
+ Value(b"old-0"),
+ Value(b"old-1"),
+ Value(b"old-2"),
+ Value(b"old-3"),
+ Null,
+ Placeholder,
+ ],
+ );
+ write_blob_file_with_values(
+ &middle_path,
+ &[Placeholder, Value(b"middle-1"), Value(b"middle-2")],
+ );
+ copy_blob_fixture("blob-placeholder.blob", &latest_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.clone(),
+ Identifier::new("default", "blob_fallback_t"),
+ table_path,
+ table_schema,
+ None,
+ );
+
+ let files = vec![
+ data_file_meta_with_path(
+ "data.parquet",
+ 0,
+ 6,
+ 1,
+ parquet_path.metadata().unwrap().len() as i64,
+ Some(vec!["id"]),
+ ),
+ data_file_meta_with_path(
+ "blob-base.blob",
+ 0,
+ 6,
+ 1,
+ base_path.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ ),
+ data_file_meta_with_path(
+ "blob-middle.blob",
+ 0,
+ 3,
+ 2,
+ middle_path.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ ),
+ data_file_meta_with_path(
+ "blob-latest.blob",
+ 1,
+ 4,
+ 3,
+ latest_path.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ ),
+ ];
+ 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(files.clone())
+ .build()
+ .unwrap();
+
+ let read = TableRead::new(&table, table.schema().fields().to_vec(),
Vec::new());
+ let batches = read
+ .to_arrow(std::slice::from_ref(&split))
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(collect_int_values(&batches, "id"), vec![1, 2, 3, 4, 5, 6]);
+ assert_eq!(
+ collect_binary_values(&batches, "payload"),
+ vec![
+ Some(b"old-0".to_vec()),
+ Some(b"middle-1".to_vec()),
+ None,
+ Some(b"latest-3".to_vec()),
+ None,
+ None,
+ ]
+ );
+
+ let descriptor_table = table.copy_with_options(HashMap::from([(
+ "blob-as-descriptor".to_string(),
+ "true".to_string(),
+ )]));
+ let descriptor_read = TableRead::new(
+ &descriptor_table,
+ descriptor_table.schema().fields().to_vec(),
+ Vec::new(),
+ );
+ let descriptor_batches = descriptor_read
+ .to_arrow(std::slice::from_ref(&split))
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let descriptors = collect_binary_values(&descriptor_batches,
"payload");
+ assert!(
+ BlobDescriptor::deserialize(descriptors[0].as_deref().unwrap())
+ .unwrap()
+ .uri()
+ .ends_with("blob-base.blob")
+ );
+ assert!(
+ BlobDescriptor::deserialize(descriptors[1].as_deref().unwrap())
+ .unwrap()
+ .uri()
+ .ends_with("blob-middle.blob")
+ );
+ assert!(descriptors[2].is_none());
+ assert!(
+ BlobDescriptor::deserialize(descriptors[3].as_deref().unwrap())
+ .unwrap()
+ .uri()
+ .ends_with("blob-latest.blob")
+ );
+ assert!(descriptors[4].is_none());
+ assert!(descriptors[5].is_none());
+
+ let deletion_path = format!("{}/index/dv-0",
local_file_path(tempdir.path()));
+ let deletion_file = write_test_deletion_file(&file_io, &deletion_path,
&[2]).await;
+ let selected_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(files)
+ .with_data_deletion_files(vec![Some(deletion_file), None, None,
None])
+ .with_row_ranges(vec![RowRange::new(1, 5)])
+ .build()
+ .unwrap();
+ let selected = read
+ .to_arrow(&[selected_split])
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(collect_int_values(&selected, "id"), vec![2, 4, 5, 6]);
+ assert_eq!(
+ collect_binary_values(&selected, "payload"),
+ vec![
+ Some(b"middle-1".to_vec()),
+ Some(b"latest-3".to_vec()),
+ None,
+ None
+ ]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_blob_fallback_across_schema_ids_skips_unselected_files() {
+ use BlobFixtureValue::{Placeholder, Value};
+
+ 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 old_path = bucket_dir.join("blob-old-selected.blob");
+ let latest_path = bucket_dir.join("blob-latest-selected.blob");
+ write_blob_file_with_values(&old_path, &[Value(b"old-2"),
Value(b"old-3")]);
+ write_blob_file_with_values(&latest_path, &[Placeholder, Placeholder]);
+
+ let schema_v0 = 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 schema_v1 = TableSchema::new(
+ 1,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Blob(BlobType::new()))
+ .column("added", DataType::Int(IntType::new()))
+ .option("data-evolution.enabled", "true")
+ .build()
+ .unwrap(),
+ );
+ assert_eq!(schema_v0.fields()[1].id(), schema_v1.fields()[1].id());
+
+ let file_io = FileIOBuilder::new("file").build().unwrap();
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "blob_schema_fallback_t"),
+ table_path,
+ schema_v1.clone(),
+ None,
+ );
+ write_schema_file(&table, &schema_v0).await;
+ write_schema_file(&table, &schema_v1).await;
+
+ let mut anchor = data_file_meta_with_path(
+ "data.parquet",
+ 0,
+ 4,
+ 1,
+ parquet_path.metadata().unwrap().len() as i64,
+ Some(vec!["id"]),
+ );
+ anchor.schema_id = 1;
+
+ // These two files make each sequence group cover rows 0..=3, but are
+ // deliberately absent. A read restricted to rows 2..=3 must not open
them.
+ let mut latest_unselected = data_file_meta_with_path(
+ "blob-latest-unselected.blob",
+ 0,
+ 2,
+ 2,
+ 5,
+ Some(vec!["payload"]),
+ );
+ latest_unselected.schema_id = 1;
+ let old_unselected = data_file_meta_with_path(
+ "blob-old-unselected.blob",
+ 0,
+ 2,
+ 1,
+ 5,
+ Some(vec!["payload"]),
+ );
+ let mut latest_selected = data_file_meta_with_path(
+ "blob-latest-selected.blob",
+ 2,
+ 2,
+ 2,
+ latest_path.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ );
+ latest_selected.schema_id = 1;
+ let old_selected = data_file_meta_with_path(
+ "blob-old-selected.blob",
+ 2,
+ 2,
+ 1,
+ old_path.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ );
+
+ 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![
+ anchor,
+ latest_unselected,
+ old_unselected,
+ latest_selected,
+ old_selected,
+ ])
+ .with_row_ranges(vec![RowRange::new(2, 3)])
+ .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![3, 4]);
+ assert_eq!(
+ collect_binary_values(&batches, "payload"),
+ vec![Some(b"old-2".to_vec()), Some(b"old-3".to_vec())]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_single_blob_sequence_group_yields_before_opening_next_file()
{
+ 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 first_blob_path = bucket_dir.join("blob-first.blob");
+ write_blob_file(
+ &first_blob_path,
+ &[Some(&b"first-0"[..]), Some(&b"first-1"[..])],
+ );
+
+ 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_lazy_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-first.blob",
+ 0,
+ 2,
+ 1,
+ first_blob_path.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ ),
+ data_file_meta_with_path(
+ "blob-missing-next.blob",
+ 2,
+ 2,
+ 1,
+ 5,
+ Some(vec!["payload"]),
+ ),
+ ])
+ .build()
+ .unwrap();
+
+ let read = TableRead::new(&table, table.schema().fields().to_vec(),
Vec::new());
+ let mut stream = read.to_arrow(&[split]).unwrap();
+ let first_batch = stream.try_next().await.unwrap().unwrap();
+
+ assert_eq!(
+ collect_int_values(std::slice::from_ref(&first_batch), "id"),
+ vec![1, 2]
+ );
+ assert_eq!(
+ collect_binary_values(std::slice::from_ref(&first_batch),
"payload"),
+ vec![Some(b"first-0".to_vec()), Some(b"first-1".to_vec())]
+ );
+ }
+
+ #[tokio::test]
+ async fn test_blob_fallback_defers_later_files_until_their_batch() {
+ use BlobFixtureValue::{Placeholder, Value};
+
+ 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", (0..2048).collect::<Vec<_>>())],
+ None,
+ );
+
+ let latest_path = bucket_dir.join("blob-latest-first.blob");
+ let older_path = bucket_dir.join("blob-older-first.blob");
+ let latest_values = vec![Placeholder; 1024];
+ let older_values = vec![Value(b"old"); 1024];
+ write_blob_file_with_values(&latest_path, &latest_values);
+ write_blob_file_with_values(&older_path, &older_values);
+
+ 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_lazy_fallback_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,
+ 2048,
+ 1,
+ parquet_path.metadata().unwrap().len() as i64,
+ Some(vec!["id"]),
+ ),
+ data_file_meta_with_path(
+ "blob-latest-first.blob",
+ 0,
+ 1024,
+ 2,
+ latest_path.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ ),
+ data_file_meta_with_path(
+ "blob-older-first.blob",
+ 0,
+ 1024,
+ 1,
+ older_path.metadata().unwrap().len() as i64,
+ Some(vec!["payload"]),
+ ),
+ data_file_meta_with_path(
+ "blob-older-missing.blob",
+ 1024,
+ 1024,
+ 1,
+ 5,
+ Some(vec!["payload"]),
+ ),
+ ])
+ .build()
+ .unwrap();
+
+ let read = TableRead::new(&table, table.schema().fields().to_vec(),
Vec::new());
+ let mut stream = read.to_arrow(&[split]).unwrap();
+ let first_batch = stream.try_next().await.unwrap().unwrap();
+
+ assert_eq!(first_batch.num_rows(), 1024);
+ let ids = collect_int_values(std::slice::from_ref(&first_batch), "id");
+ assert_eq!(ids.first(), Some(&0));
+ assert_eq!(ids.last(), Some(&1023));
+ assert!(
+ collect_binary_values(std::slice::from_ref(&first_batch),
"payload")
+ .into_iter()
+ .all(|value| value.as_deref() == Some(&b"old"[..]))
+ );
+
+ assert!(stream.try_next().await.is_err());
+ }
+
#[tokio::test]
async fn test_table_read_merges_multiple_blob_columns_with_row_ranges() {
let tempdir = tempdir().unwrap();
@@ -4150,6 +4769,48 @@ mod tests {
file
}
+ async fn write_schema_file(table: &Table, schema: &TableSchema) {
+ let path = table.schema_manager().schema_path(schema.id());
+ let dir = path.rsplit_once('/').map(|(dir, _)| dir).unwrap();
+ table.file_io().mkdirs(dir).await.unwrap();
+ let json = serde_json::to_vec(schema).unwrap();
+ table
+ .file_io()
+ .new_output(&path)
+ .unwrap()
+ .write(Bytes::from(json))
+ .await
+ .unwrap();
+ }
+
+ async fn write_test_deletion_file(
+ file_io: &crate::io::FileIO,
+ path: &str,
+ deleted_rows: &[u32],
+ ) -> DeletionFile {
+ let mut bitmap = RoaringBitmap::new();
+ for row in deleted_rows {
+ bitmap.insert(*row);
+ }
+ let bytes = DeletionVector::from_bitmap(bitmap)
+ .serialize_to_bytes()
+ .unwrap();
+ let bitmap_length =
i32::from_be_bytes(bytes[0..4].try_into().unwrap());
+ file_io
+ .new_output(path)
+ .unwrap()
+ .write(Bytes::from(bytes))
+ .await
+ .unwrap();
+
+ DeletionFile::new(
+ path.to_string(),
+ 0,
+ bitmap_length as i64,
+ Some(deleted_rows.len() as i64),
+ )
+ }
+
fn copy_blob_fixture(name: &str, destination: &Path) {
let source = blob_fixture_path(name);
fs::copy(&source, destination).unwrap_or_else(|e| {
diff --git a/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
b/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
new file mode 100644
index 0000000..e34dbca
--- /dev/null
+++ b/crates/paimon/src/table/data_evolution_reader/blob_fallback.rs
@@ -0,0 +1,432 @@
+// 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::{
+ blob_file_row_range, row_range_overlaps_any,
selected_absolute_row_ranges_for_file, BlobBunch,
+ DeletionVectorContext,
+};
+use crate::arrow::build_target_arrow_schema;
+use crate::arrow::format::blob::{BlobReadValue, IndexedBlobReader};
+use crate::io::FileIO;
+use crate::spec::DataField;
+use crate::table::{ArrowRecordBatchStream, RowRange};
+use crate::{DataSplit, Error};
+use arrow_array::builder::BinaryBuilder;
+use arrow_array::RecordBatch;
+use async_stream::try_stream;
+use futures::StreamExt;
+use std::collections::VecDeque;
+use std::sync::Arc;
+
+const BATCH_SIZE: usize = 1024;
+
+struct LazyBlobFile {
+ range: RowRange,
+ file_name: String,
+ path: String,
+ file_size: i64,
+ row_count: i64,
+ reader: Option<IndexedBlobReader>,
+}
+
+impl LazyBlobFile {
+ async fn read_positions(
+ &mut self,
+ positions: &[usize],
+ file_io: &FileIO,
+ blob_as_descriptor: bool,
+ ) -> crate::Result<Vec<BlobReadValue>> {
+ if self.reader.is_none() {
+ let file_size = u64::try_from(self.file_size).map_err(|e|
Error::DataInvalid {
+ message: format!(
+ "Blob file '{}' has negative file size {}",
+ self.file_name, self.file_size
+ ),
+ source: Some(Box::new(e)),
+ })?;
+ let input = file_io.new_input(&self.path)?;
+ let reader = input.reader().await?;
+ let reader = IndexedBlobReader::open(
+ Box::new(reader),
+ file_size,
+ self.path.clone(),
+ blob_as_descriptor,
+ )
+ .await?;
+ let indexed_rows =
+ i64::try_from(reader.num_rows()).map_err(|e|
Error::DataInvalid {
+ message: format!(
+ "Blob file '{}' index row count {} exceeds i64",
+ self.file_name,
+ reader.num_rows()
+ ),
+ source: Some(Box::new(e)),
+ })?;
+ if indexed_rows != self.row_count {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Blob file '{}' index contains {indexed_rows} rows but
metadata declares {}",
+ self.file_name, self.row_count
+ ),
+ source: None,
+ });
+ }
+ self.reader = Some(reader);
+ }
+
+ self.reader
+ .as_ref()
+ .expect("blob reader is initialized above")
+ .read_positions(positions)
+ .await
+ }
+
+ fn release_reader(&mut self) {
+ self.reader = None;
+ }
+}
+
+pub(super) fn read(
+ split: &DataSplit,
+ bunch: BlobBunch,
+ read_fields: Vec<DataField>,
+ row_ranges: Option<Vec<RowRange>>,
+ file_io: FileIO,
+ blob_as_descriptor: bool,
+ anchor_deletion_vector: Option<DeletionVectorContext>,
+) -> crate::Result<ArrowRecordBatchStream> {
+ if read_fields.len() != 1 || !read_fields[0].data_type().is_blob_type() {
+ return Err(Error::DataInvalid {
+ message: "Blob bunch should provide exactly one BLOB
field".to_string(),
+ source: None,
+ });
+ }
+
+ let target_schema = build_target_arrow_schema(&read_fields)?;
+ let split = split.clone();
+
+ Ok(try_stream! {
+ bunch.validate_logical_range()?;
+ let expected_range = bunch.expected_range()?;
+ let selected_ranges = selected_absolute_row_ranges_for_file(
+ bunch.expected_first_row_id,
+ bunch.expected_row_count,
+ row_ranges.as_deref(),
+ anchor_deletion_vector
+ .as_ref()
+ .map(|context| context.deletion_vector.as_ref()),
+ )?
+ .unwrap_or_else(|| vec![expected_range]);
+
+ let mut sequence_groups = Vec::new();
+ for files in bunch.sequence_groups() {
+ let mut group = VecDeque::with_capacity(files.len());
+ for file in files {
+ let range = blob_file_row_range(&file)?;
+ if !row_range_overlaps_any(&range, &selected_ranges) {
+ continue;
+ }
+ let path = split.data_file_path(&file);
+ group.push_back(LazyBlobFile {
+ range,
+ file_name: file.file_name,
+ path,
+ file_size: file.file_size,
+ row_count: file.row_count,
+ reader: None,
+ });
+ }
+ if !group.is_empty() {
+ sequence_groups.push(group);
+ }
+ }
+
+ let mut row_cursor = RowIdBatchCursor::new(selected_ranges);
+ while let Some(row_ids) = row_cursor.next_batch(BATCH_SIZE) {
+ yield resolve_batch(
+ &mut sequence_groups,
+ &row_ids,
+ target_schema.clone(),
+ &file_io,
+ blob_as_descriptor,
+ ).await?;
+ }
+ }
+ .boxed())
+}
+
+async fn resolve_batch(
+ sequence_groups: &mut [VecDeque<LazyBlobFile>],
+ row_ids: &[i64],
+ target_schema: Arc<arrow_schema::Schema>,
+ file_io: &FileIO,
+ blob_as_descriptor: bool,
+) -> crate::Result<RecordBatch> {
+ let mut resolved = (0..row_ids.len())
+ .map(|_| BlobReadValue::Placeholder)
+ .collect::<Vec<_>>();
+ let mut unresolved_count = resolved.len();
+ let batch_from = row_ids[0];
+ let batch_to = *row_ids.last().expect("row id batch is non-empty");
+
+ for group in sequence_groups.iter_mut() {
+ while group
+ .front()
+ .is_some_and(|file| file.range.to() < batch_from)
+ {
+ group.pop_front();
+ }
+ }
+
+ // Groups are newest first. A missing row or placeholder leaves the row
unresolved;
+ // an explicit NULL or value stops fallback.
+ for group in sequence_groups.iter_mut() {
+ for file in group.iter_mut() {
+ if unresolved_count == 0 || file.range.from() > batch_to {
+ break;
+ }
+
+ let mut output_positions = Vec::new();
+ let mut file_positions = Vec::new();
+ for (output_position, row_id) in
row_ids.iter().copied().enumerate() {
+ if !matches!(&resolved[output_position],
BlobReadValue::Placeholder) {
+ continue;
+ }
+ if row_id < file.range.from() || row_id > file.range.to() {
+ continue;
+ }
+ output_positions.push(output_position);
+ file_positions.push(usize::try_from(row_id -
file.range.from()).map_err(|e| {
+ Error::DataInvalid {
+ message: format!(
+ "Blob row id {row_id} cannot be represented as a
file position"
+ ),
+ source: Some(Box::new(e)),
+ }
+ })?);
+ }
+
+ if !file_positions.is_empty() {
+ let values = file
+ .read_positions(&file_positions, file_io,
blob_as_descriptor)
+ .await?;
+ for (output_position, value) in
output_positions.into_iter().zip(values) {
+ if !matches!(&value, BlobReadValue::Placeholder) {
+ resolved[output_position] = value;
+ unresolved_count -= 1;
+ }
+ }
+ }
+
+ if file.range.to() <= batch_to {
+ file.release_reader();
+ }
+ }
+
+ if unresolved_count == 0 {
+ break;
+ }
+ }
+
+ for group in sequence_groups.iter_mut() {
+ while group
+ .front()
+ .is_some_and(|file| file.range.to() <= batch_to)
+ {
+ group.pop_front();
+ }
+ }
+
+ let mut builder = BinaryBuilder::new();
+ for value in resolved {
+ match value {
+ BlobReadValue::Value(bytes) => builder.append_value(bytes),
+ BlobReadValue::Null | BlobReadValue::Placeholder =>
builder.append_null(),
+ }
+ }
+ RecordBatch::try_new(target_schema,
vec![Arc::new(builder.finish())]).map_err(|e| {
+ Error::UnexpectedError {
+ message: format!("Failed to build blob fallback RecordBatch: {e}"),
+ source: Some(Box::new(e)),
+ }
+ })
+}
+
+struct RowIdBatchCursor {
+ ranges: Vec<RowRange>,
+ range_index: usize,
+ next_row_id: Option<i64>,
+}
+
+impl RowIdBatchCursor {
+ fn new(ranges: Vec<RowRange>) -> Self {
+ let next_row_id = ranges.first().map(RowRange::from);
+ Self {
+ ranges,
+ range_index: 0,
+ next_row_id,
+ }
+ }
+
+ fn next_batch(&mut self, batch_size: usize) -> Option<Vec<i64>> {
+ let mut row_ids = Vec::with_capacity(batch_size);
+ while row_ids.len() < batch_size {
+ let Some(row_id) = self.next_row_id else {
+ break;
+ };
+ row_ids.push(row_id);
+
+ let range = &self.ranges[self.range_index];
+ if row_id == range.to() {
+ self.range_index += 1;
+ self.next_row_id =
self.ranges.get(self.range_index).map(RowRange::from);
+ } else {
+ self.next_row_id = Some(row_id + 1);
+ }
+ }
+ (!row_ids.is_empty()).then_some(row_ids)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::io::FileRead;
+ use crate::spec::{BlobType, DataType};
+ use arrow_array::{Array, BinaryArray};
+ use bytes::Bytes;
+ use std::ops::Range;
+ use std::sync::atomic::{AtomicUsize, Ordering};
+
+ #[allow(dead_code)]
+ mod blob_test_utils {
+ include!(concat!(
+ env!("CARGO_MANIFEST_DIR"),
+ "/../blob_test_utils.rs"
+ ));
+ }
+
+ use blob_test_utils::{build_blob_file_bytes_with_values, BlobFixtureValue};
+
+ #[derive(Clone)]
+ struct TrackingFileRead {
+ bytes: Bytes,
+ reads: Arc<AtomicUsize>,
+ }
+
+ #[async_trait::async_trait]
+ impl FileRead for TrackingFileRead {
+ async fn read(&self, range: Range<u64>) -> crate::Result<Bytes> {
+ self.reads.fetch_add(1, Ordering::SeqCst);
+ Ok(self.bytes.slice(range.start as usize..range.end as usize))
+ }
+ }
+
+ async fn tracking_blob_file(
+ file_name: &str,
+ first_row_id: i64,
+ values: &[BlobFixtureValue<'_>],
+ ) -> (LazyBlobFile, Arc<AtomicUsize>) {
+ let bytes = Bytes::from(build_blob_file_bytes_with_values(values));
+ let file_size = bytes.len() as u64;
+ let reads = Arc::new(AtomicUsize::new(0));
+ let reader = IndexedBlobReader::open(
+ Box::new(TrackingFileRead {
+ bytes,
+ reads: reads.clone(),
+ }),
+ file_size,
+ file_name.to_string(),
+ false,
+ )
+ .await
+ .unwrap();
+ reads.store(0, Ordering::SeqCst);
+
+ let last_row_id = first_row_id + i64::try_from(values.len()).unwrap()
- 1;
+ (
+ LazyBlobFile {
+ range: RowRange::new(first_row_id, last_row_id),
+ file_name: file_name.to_string(),
+ path: file_name.to_string(),
+ file_size: i64::try_from(file_size).unwrap(),
+ row_count: i64::try_from(values.len()).unwrap(),
+ reader: Some(reader),
+ },
+ reads,
+ )
+ }
+
+ #[tokio::test]
+ async fn test_resolve_batch_skips_payloads_for_resolved_rows() {
+ use BlobFixtureValue::{Null, Placeholder, Value};
+
+ let (latest, latest_reads) =
+ tracking_blob_file("latest.blob", 0, &[Value(b"new-0"), Null,
Placeholder]).await;
+ let (older, older_reads) = tracking_blob_file(
+ "older.blob",
+ 0,
+ &[
+ Value(b"old-0"),
+ Value(b"old-1"),
+ Value(b"old-2"),
+ Value(b"old-3"),
+ ],
+ )
+ .await;
+ let (oldest, oldest_reads) = tracking_blob_file(
+ "oldest.blob",
+ 0,
+ &[
+ Value(b"ancient-0"),
+ Value(b"ancient-1"),
+ Value(b"ancient-2"),
+ Value(b"ancient-3"),
+ ],
+ )
+ .await;
+ let schema = build_target_arrow_schema(&[DataField::new(
+ 0,
+ "payload".to_string(),
+ DataType::Blob(BlobType::new()),
+ )])
+ .unwrap();
+
+ let mut groups = vec![
+ VecDeque::from([latest]),
+ VecDeque::from([older]),
+ VecDeque::from([oldest]),
+ ];
+ let file_io = crate::io::FileIOBuilder::new("file").build().unwrap();
+ let batch = resolve_batch(&mut groups, &[0, 1, 2, 3], schema,
&file_io, false)
+ .await
+ .unwrap();
+ let values = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+
+ assert_eq!(values.value(0), b"new-0");
+ assert!(values.is_null(1));
+ assert_eq!(values.value(2), b"old-2");
+ assert_eq!(values.value(3), b"old-3");
+ assert_eq!(latest_reads.load(Ordering::SeqCst), 1);
+ assert_eq!(older_reads.load(Ordering::SeqCst), 2);
+ assert_eq!(oldest_reads.load(Ordering::SeqCst), 0);
+ }
+}
diff --git a/crates/paimon/testdata/blob/blob-placeholder.blob
b/crates/paimon/testdata/blob/blob-placeholder.blob
new file mode 100644
index 0000000..49d169b
Binary files /dev/null and b/crates/paimon/testdata/blob/blob-placeholder.blob
differ
