From 53225527b354e903a3a32e42eab6170ae6dbf68d Mon Sep 17 00:00:00 2001 From: QuakeWang Date: Tue, 14 Jul 2026 13:41:41 +0800 Subject: [PATCH] fix(blob): support unknown-length BlobDescriptor ranges Dedicated blob writers cast descriptor lengths to u64, while inline resolvers rejected negative lengths. This broke valid Java descriptors with length -1. Treat -1 as offset-to-EOF in both paths, validate ranges, and reject short reads before they can produce inconsistent blob data. Signed-off-by: QuakeWang --- .../datafusion/tests/blob_tests.rs | 179 +++++++++++++++++- crates/paimon/src/arrow/format/blob.rs | 116 ++++++++++-- crates/paimon/src/spec/blob_descriptor.rs | 112 +++++++++++ crates/paimon/src/table/blob_resolver.rs | 141 ++++++++++---- docs/src/sql.md | 4 + 5 files changed, 500 insertions(+), 52 deletions(-) diff --git a/crates/integrations/datafusion/tests/blob_tests.rs b/crates/integrations/datafusion/tests/blob_tests.rs index 84fad62a..b844f65e 100644 --- a/crates/integrations/datafusion/tests/blob_tests.rs +++ b/crates/integrations/datafusion/tests/blob_tests.rs @@ -22,7 +22,7 @@ mod common; use arrow_array::{Array, BinaryArray, Int32Array, RecordBatch, StringArray}; -use common::{create_sql_context, create_test_env, exec}; +use common::{assert_sql_error, create_sql_context, create_test_env, exec}; use paimon::catalog::Identifier; use paimon::spec::{BlobDescriptor, BlobViewStruct}; use paimon::table::BranchManager; @@ -577,6 +577,66 @@ async fn test_blob_resolve_descriptor_with_offset() { assert_eq!(rows[0], (1, "Partial".into(), Some(b"PAYLOAD".to_vec()))); } +#[tokio::test] +async fn test_blob_resolve_unknown_length_descriptor() { + let (tmp, sql_context) = setup(BLOB_TABLE_DDL).await; + + let source_data = b"HEADER_PAYLOAD_TRAILER"; + let source_path = tmp.path().join("blob_unknown_length.bin"); + std::fs::write(&source_path, source_data).unwrap(); + + let uri = format!("file://{}", source_path.display()); + let full_hex = to_hex(&BlobDescriptor::new(uri.clone(), 0, -1).serialize()); + let suffix_hex = to_hex(&BlobDescriptor::new(uri.clone(), 7, -1).serialize()); + let eof_hex = + to_hex(&BlobDescriptor::new(uri.clone(), source_data.len() as i64, -1).serialize()); + let past_eof_hex = + to_hex(&BlobDescriptor::new(uri, source_data.len() as i64 + 5, -1).serialize()); + + let sql = format!( + "INSERT INTO paimon.test_db.t (id, name, picture) VALUES \ + (1, 'Full', X'{full_hex}'), \ + (2, 'Suffix', X'{suffix_hex}'), \ + (3, 'Eof', X'{eof_hex}'), \ + (4, 'PastEof', X'{past_eof_hex}'), \ + (5, 'Raw', X'524157'), \ + (6, 'Null', NULL)" + ); + exec(&sql_context, &sql).await; + + let rows = query_id_name_picture( + &sql_context, + "SELECT id, name, picture FROM paimon.test_db.t ORDER BY id", + ) + .await; + assert_eq!( + rows, + vec![ + (1, "Full".into(), Some(source_data.to_vec())), + (2, "Suffix".into(), Some(b"PAYLOAD_TRAILER".to_vec())), + (3, "Eof".into(), Some(Vec::new())), + (4, "PastEof".into(), Some(Vec::new())), + (5, "Raw".into(), Some(b"RAW".to_vec())), + (6, "Null".into(), None), + ] + ); +} + +#[tokio::test] +async fn test_blob_descriptor_short_read_returns_error() { + let (tmp, sql_context) = setup(BLOB_TABLE_DDL).await; + + let source_path = tmp.path().join("blob_short_read.bin"); + std::fs::write(&source_path, b"short").unwrap(); + let uri = format!("file://{}", source_path.display()); + let desc_hex = to_hex(&BlobDescriptor::new(uri, 0, 6).serialize()); + let sql = format!( + "INSERT INTO paimon.test_db.t (id, name, picture) VALUES (1, 'Short', X'{desc_hex}')" + ); + + assert_sql_error(&sql_context, &sql, "Failed to read BlobDescriptor").await; +} + /// Blob files roll independently when `blob.target-file-size` is small. #[tokio::test] async fn test_blob_rolling() { @@ -709,6 +769,123 @@ async fn test_blob_descriptor_field_resolve_descriptor_value() { ); } +#[tokio::test] +async fn test_blob_descriptor_field_resolve_unknown_length_descriptor() { + let (tmp, sql_context) = setup( + "CREATE TABLE paimon.test_db.t (\ + id INT, \ + name STRING, \ + picture BLOB \ + ) WITH (\ + 'data-evolution.enabled' = 'true', \ + 'row-tracking.enabled' = 'true', \ + 'blob-descriptor-field' = 'picture'\ + )", + ) + .await; + + let source_data = b"HEADER_PAYLOAD_TRAILER"; + let source_path = tmp.path().join("descriptor_unknown_length.bin"); + std::fs::write(&source_path, source_data).unwrap(); + + let uri = format!("file://{}", source_path.display()); + let bounded_hex = to_hex(&BlobDescriptor::new(uri.clone(), 0, 6).serialize()); + let suffix_hex = to_hex(&BlobDescriptor::new(uri.clone(), 7, -1).serialize()); + let eof_hex = + to_hex(&BlobDescriptor::new(uri.clone(), source_data.len() as i64, -1).serialize()); + let past_eof_hex = + to_hex(&BlobDescriptor::new(uri, source_data.len() as i64 + 5, -1).serialize()); + let sql = format!( + "INSERT INTO paimon.test_db.t (id, name, picture) VALUES \ + (1, 'Bounded', X'{bounded_hex}'), \ + (2, 'Suffix', X'{suffix_hex}'), \ + (3, 'Eof', X'{eof_hex}'), \ + (4, 'PastEof', X'{past_eof_hex}'), \ + (5, 'Raw', X'524157'), \ + (6, 'Null', NULL)" + ); + exec(&sql_context, &sql).await; + + let rows = query_id_name_picture( + &sql_context, + "SELECT id, name, picture FROM paimon.test_db.t ORDER BY id", + ) + .await; + assert_eq!( + rows, + vec![ + (1, "Bounded".into(), Some(b"HEADER".to_vec())), + (2, "Suffix".into(), Some(b"PAYLOAD_TRAILER".to_vec())), + (3, "Eof".into(), Some(Vec::new())), + (4, "PastEof".into(), Some(Vec::new())), + (5, "Raw".into(), Some(b"RAW".to_vec())), + (6, "Null".into(), None), + ] + ); +} + +#[tokio::test] +async fn test_blob_descriptor_field_rejects_invalid_length() { + let (tmp, sql_context) = setup( + "CREATE TABLE paimon.test_db.t (\ + id INT, \ + name STRING, \ + picture BLOB \ + ) WITH (\ + 'data-evolution.enabled' = 'true', \ + 'row-tracking.enabled' = 'true', \ + 'blob-descriptor-field' = 'picture'\ + )", + ) + .await; + + let uri = format!("file://{}", tmp.path().join("unused.bin").display()); + let desc_hex = to_hex(&BlobDescriptor::new(uri, 0, -2).serialize()); + let sql = format!( + "INSERT INTO paimon.test_db.t (id, name, picture) VALUES (1, 'Invalid', X'{desc_hex}')" + ); + exec(&sql_context, &sql).await; + + assert_sql_error( + &sql_context, + "SELECT id, name, picture FROM paimon.test_db.t", + "length must be -1 or non-negative", + ) + .await; +} + +#[tokio::test] +async fn test_blob_descriptor_field_short_read_returns_error() { + let (tmp, sql_context) = setup( + "CREATE TABLE paimon.test_db.t (\ + id INT, \ + name STRING, \ + picture BLOB \ + ) WITH (\ + 'data-evolution.enabled' = 'true', \ + 'row-tracking.enabled' = 'true', \ + 'blob-descriptor-field' = 'picture'\ + )", + ) + .await; + + let source_path = tmp.path().join("descriptor_short_read.bin"); + std::fs::write(&source_path, b"short").unwrap(); + let uri = format!("file://{}", source_path.display()); + let desc_hex = to_hex(&BlobDescriptor::new(uri, 0, 6).serialize()); + let sql = format!( + "INSERT INTO paimon.test_db.t (id, name, picture) VALUES (1, 'Short', X'{desc_hex}')" + ); + exec(&sql_context, &sql).await; + + assert_sql_error( + &sql_context, + "SELECT id, name, picture FROM paimon.test_db.t", + "Failed to read BlobDescriptor", + ) + .await; +} + #[tokio::test] async fn test_blob_descriptor_filter_before_resolve_skips_filtered_bad_descriptor() { let (tmp, sql_context) = setup( diff --git a/crates/paimon/src/arrow/format/blob.rs b/crates/paimon/src/arrow/format/blob.rs index 136170ec..c401a06e 100644 --- a/crates/paimon/src/arrow/format/blob.rs +++ b/crates/paimon/src/arrow/format/blob.rs @@ -617,6 +617,23 @@ impl BlobFormatWriter { const BLOB_WRITE_BUFFER_SIZE: u64 = 8 * 1024 * 1024; // 8 MB +fn checked_blob_entry_length(payload_len: u64) -> crate::Result { + let entry_length = payload_len + .checked_add(BLOB_ENTRY_OVERHEAD) + .ok_or_else(|| Error::DataInvalid { + message: format!( + "Blob entry length overflows u64: payload_length={payload_len}, overhead={BLOB_ENTRY_OVERHEAD}" + ), + source: None, + })?; + i64::try_from(entry_length).map_err(|e| Error::DataInvalid { + message: format!( + "Blob entry length exceeds i64: payload_length={payload_len}, entry_length={entry_length}" + ), + source: Some(Box::new(e)), + }) +} + #[async_trait] impl FormatFileWriter for BlobFormatWriter { async fn write(&mut self, batch: &RecordBatch) -> crate::Result<()> { @@ -643,9 +660,7 @@ impl FormatFileWriter for BlobFormatWriter { if BlobDescriptor::is_blob_descriptor(value) { let desc = BlobDescriptor::deserialize(value)?; - let payload_len = desc.length() as u64; - let entry_length = (payload_len + BLOB_ENTRY_OVERHEAD) as i64; - self.lengths.push(entry_length); + let range = desc.range_spec()?; let file_io = self.file_io.as_ref().ok_or_else(|| Error::DataInvalid { message: @@ -654,7 +669,47 @@ impl FormatFileWriter for BlobFormatWriter { source: None, })?; let input = file_io.new_input(desc.uri())?; - let reader = input.reader().await?; + let offset = range.offset(); + let payload_len = match range.length() { + Some(length) => length, + None => input + .metadata() + .await + .map_err(|e| Error::UnexpectedError { + message: format!( + "Failed to read metadata for BlobDescriptor '{}': {e}", + desc.uri() + ), + source: Some(Box::new(e)), + })? + .size + .saturating_sub(offset), + }; + let end = offset + .checked_add(payload_len) + .ok_or_else(|| Error::DataInvalid { + message: format!( + "BlobDescriptor range overflows u64: offset={offset}, length={payload_len}" + ), + source: None, + })?; + let entry_length = checked_blob_entry_length(payload_len)?; + let entry_length_u64 = entry_length as u64; + let bytes_written = self + .bytes_written + .checked_add(entry_length_u64) + .ok_or_else(|| Error::DataInvalid { + message: format!( + "Blob file size overflows u64: current_size={}, entry_length={entry_length_u64}", + self.bytes_written + ), + source: None, + })?; + let reader = if payload_len == 0 { + None + } else { + Some(input.reader().await?) + }; let mut hasher = crc32fast::Hasher::new(); @@ -664,15 +719,34 @@ impl FormatFileWriter for BlobFormatWriter { .await?; // Stream payload in chunks to avoid loading entire blob into memory - let start = desc.offset() as u64; - let end = start + payload_len; - let mut pos = start; - while pos < end { - let chunk_end = (pos + BLOB_WRITE_BUFFER_SIZE).min(end); - let chunk = reader.read(pos..chunk_end).await?; - hasher.update(&chunk); - self.writer.write(chunk).await?; - pos = chunk_end; + if let Some(reader) = reader.as_ref() { + let mut pos = offset; + while pos < end { + let chunk_end = pos.saturating_add(BLOB_WRITE_BUFFER_SIZE).min(end); + let chunk = reader.read(pos..chunk_end).await.map_err(|e| { + Error::UnexpectedError { + message: format!( + "Failed to read BlobDescriptor '{}' range {pos}..{chunk_end}: {e}", + desc.uri() + ), + source: Some(Box::new(e)), + } + })?; + let actual_len = chunk.len() as u64; + let expected_len = chunk_end - pos; + if actual_len != expected_len { + return Err(Error::DataInvalid { + message: format!( + "Failed to read BlobDescriptor '{}': short read for range {pos}..{chunk_end}, expected={expected_len} bytes, actual={actual_len} bytes", + desc.uri() + ), + source: None, + }); + } + hasher.update(&chunk); + self.writer.write(chunk).await?; + pos = chunk_end; + } } let entry_length_bytes = entry_length.to_le_bytes(); @@ -685,7 +759,8 @@ impl FormatFileWriter for BlobFormatWriter { .write(Bytes::copy_from_slice(&hasher.finalize().to_le_bytes())) .await?; - self.bytes_written += entry_length as u64; + self.lengths.push(entry_length); + self.bytes_written = bytes_written; } else { let entry_length = (value.len() + BLOB_ENTRY_OVERHEAD as usize) as i64; self.lengths.push(entry_length); @@ -1050,6 +1125,19 @@ mod tests { assert_eq!(decoded, values); } + #[test] + fn test_checked_blob_entry_length() { + assert_eq!( + checked_blob_entry_length(0).unwrap(), + BLOB_ENTRY_OVERHEAD as i64 + ); + + let max_payload = i64::MAX as u64 - BLOB_ENTRY_OVERHEAD; + assert_eq!(checked_blob_entry_length(max_payload).unwrap(), i64::MAX); + assert!(checked_blob_entry_length(max_payload + 1).is_err()); + assert!(checked_blob_entry_length(u64::MAX).is_err()); + } + fn basic_blob_rows() -> [Option<&'static [u8]>; 4] { [ Some(&b"hello"[..]), diff --git a/crates/paimon/src/spec/blob_descriptor.rs b/crates/paimon/src/spec/blob_descriptor.rs index 8924ca6b..93c84491 100644 --- a/crates/paimon/src/spec/blob_descriptor.rs +++ b/crates/paimon/src/spec/blob_descriptor.rs @@ -28,6 +28,22 @@ pub struct BlobDescriptor { length: i64, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct BlobRangeSpec { + offset: u64, + length: Option, +} + +impl BlobRangeSpec { + pub(crate) fn offset(self) -> u64 { + self.offset + } + + pub(crate) fn length(self) -> Option { + self.length + } +} + impl BlobDescriptor { pub fn new(uri: String, offset: i64, length: i64) -> Self { Self { @@ -50,6 +66,46 @@ impl BlobDescriptor { self.length } + pub(crate) fn range_spec(&self) -> crate::Result { + if self.offset < 0 { + return Err(Error::DataInvalid { + message: format!( + "BlobDescriptor offset must be non-negative: {}", + self.offset + ), + source: None, + }); + } + if self.length < -1 { + return Err(Error::DataInvalid { + message: format!( + "BlobDescriptor length must be -1 or non-negative: {}", + self.length + ), + source: None, + }); + } + + let offset = self.offset as u64; + let length = if self.length == -1 { + None + } else { + Some(self.length as u64) + }; + if let Some(length) = length { + offset + .checked_add(length) + .ok_or_else(|| Error::DataInvalid { + message: format!( + "BlobDescriptor range overflows u64: offset={offset}, length={length}" + ), + source: None, + })?; + } + + Ok(BlobRangeSpec { offset, length }) + } + pub fn serialize(&self) -> Vec { let uri_bytes = self.uri.as_bytes(); let uri_length = uri_bytes.len(); @@ -198,4 +254,60 @@ mod tests { bytes[1] = 0xFF; assert!(BlobDescriptor::deserialize(&bytes).is_err()); } + + #[test] + fn test_range_spec_supports_unknown_length() { + let full = BlobDescriptor::new("x".to_string(), 0, -1) + .range_spec() + .unwrap(); + assert_eq!(full.offset(), 0); + assert_eq!(full.length(), None); + + let range = BlobDescriptor::new("x".to_string(), 7, -1) + .range_spec() + .unwrap(); + assert_eq!(range.offset(), 7); + assert_eq!(range.length(), None); + } + + #[test] + fn test_range_spec_supports_bounded_and_empty_ranges() { + let bounded = BlobDescriptor::new("x".to_string(), 7, 11) + .range_spec() + .unwrap(); + assert_eq!(bounded.offset(), 7); + assert_eq!(bounded.length(), Some(11)); + + let empty = BlobDescriptor::new("x".to_string(), 7, 0) + .range_spec() + .unwrap(); + assert_eq!(empty.offset(), 7); + assert_eq!(empty.length(), Some(0)); + } + + #[test] + fn test_range_spec_rejects_invalid_signed_values() { + let err = BlobDescriptor::new("x".to_string(), -1, 1) + .range_spec() + .unwrap_err(); + assert!( + matches!(err, Error::DataInvalid { message, .. } if message.contains("offset must be non-negative")) + ); + + let err = BlobDescriptor::new("x".to_string(), 0, -2) + .range_spec() + .unwrap_err(); + assert!( + matches!(err, Error::DataInvalid { message, .. } if message.contains("length must be -1 or non-negative")) + ); + } + + #[test] + fn test_range_spec_handles_largest_bounded_range() { + let range = BlobDescriptor::new("x".to_string(), i64::MAX, i64::MAX) + .range_spec() + .unwrap(); + assert_eq!(range.offset(), i64::MAX as u64); + assert_eq!(range.length(), Some(i64::MAX as u64)); + } } diff --git a/crates/paimon/src/table/blob_resolver.rs b/crates/paimon/src/table/blob_resolver.rs index bbbf24ba..d3f5a82f 100644 --- a/crates/paimon/src/table/blob_resolver.rs +++ b/crates/paimon/src/table/blob_resolver.rs @@ -46,7 +46,7 @@ pub(crate) async fn resolve_blob_column( } let mut cells = Vec::with_capacity(col.len()); - let mut requests_by_uri: HashMap> = HashMap::new(); + let mut requests_by_uri: HashMap> = HashMap::new(); let mut value_capacity = 0usize; for row in 0..col.len() { @@ -58,36 +58,14 @@ pub(crate) async fn resolve_blob_column( let value = col.value(row); if BlobDescriptor::is_blob_descriptor(value) { let desc = BlobDescriptor::deserialize(value)?; - let offset = u64::try_from(desc.offset()).map_err(|e| crate::Error::DataInvalid { - message: format!( - "BlobDescriptor offset must be non-negative: {}", - desc.offset() - ), - source: Some(Box::new(e)), - })?; - let length = u64::try_from(desc.length()).map_err(|e| crate::Error::DataInvalid { - message: format!( - "BlobDescriptor length must be non-negative: {}", - desc.length() - ), - source: Some(Box::new(e)), - })?; - offset - .checked_add(length) - .ok_or_else(|| crate::Error::DataInvalid { - message: format!( - "BlobDescriptor range overflows u64: offset={offset}, length={length}" - ), - source: None, - })?; - value_capacity = value_capacity.saturating_add(length as usize); + let range = desc.range_spec()?; requests_by_uri .entry(desc.uri().to_string()) .or_default() - .push(BlobReadRequest { + .push(BlobReadRequestSpec { row, - offset, - length, + offset: range.offset(), + length: range.length(), }); cells.push(ResolvedBlobCell::Null); } else { @@ -96,19 +74,101 @@ pub(crate) async fn resolve_blob_column( } } - let mut readers: HashMap> = HashMap::new(); for (uri, requests) in requests_by_uri { - if !readers.contains_key(&uri) { - let input = file_io.new_input(&uri)?; - let reader = input.reader().await?; - readers.insert(uri.clone(), Box::new(reader)); + let input = file_io.new_input(&uri)?; + let file_size = if requests.iter().any(|request| request.length.is_none()) { + input + .metadata() + .await + .map_err(|e| crate::Error::UnexpectedError { + message: format!("Failed to read metadata for BlobDescriptor URI '{uri}': {e}"), + source: Some(Box::new(e)), + })? + .size + } else { + 0 + }; + let mut bounded_requests = Vec::with_capacity(requests.len()); + for request in requests { + let length = request + .length + .unwrap_or_else(|| file_size.saturating_sub(request.offset)); + request + .offset + .checked_add(length) + .ok_or_else(|| crate::Error::DataInvalid { + message: format!( + "BlobDescriptor range overflows u64: offset={}, length={length}", + request.offset + ), + source: None, + })?; + value_capacity = value_capacity.saturating_add(length as usize); + if length == 0 { + cells[request.row] = ResolvedBlobCell::Value(Bytes::new()); + continue; + } + bounded_requests.push(BlobReadRequest { + row: request.row, + offset: request.offset, + length, + }); } - let reader = readers.get(&uri).unwrap(); - for merged in merge_blob_read_requests(requests) { - let data = reader.read(merged.start..merged.end).await?; + + if bounded_requests.is_empty() { + continue; + } + + let reader = input.reader().await?; + for merged in merge_blob_read_requests(bounded_requests) { + let data = reader.read(merged.start..merged.end).await.map_err(|e| { + crate::Error::UnexpectedError { + message: format!( + "Failed to read BlobDescriptor URI '{uri}' range {}..{}: {e}", + merged.start, merged.end + ), + source: Some(Box::new(e)), + } + })?; + let expected_len = merged.end - merged.start; + let actual_len = data.len() as u64; + if actual_len != expected_len { + return Err(crate::Error::DataInvalid { + message: format!( + "Failed to read BlobDescriptor URI '{uri}': short read for range {}..{}, expected={expected_len} bytes, actual={actual_len} bytes", + merged.start, merged.end + ), + source: None, + }); + } for request in merged.requests { - let start = (request.offset - merged.start) as usize; - let end = start + request.length as usize; + let start = usize::try_from(request.offset - merged.start).map_err(|e| { + crate::Error::DataInvalid { + message: format!( + "BlobDescriptor slice offset exceeds usize: offset={}, merged_start={}", + request.offset, merged.start + ), + source: Some(Box::new(e)), + } + })?; + let length = + usize::try_from(request.length).map_err(|e| crate::Error::DataInvalid { + message: format!( + "BlobDescriptor slice length exceeds usize: {}", + request.length + ), + source: Some(Box::new(e)), + })?; + let end = start + .checked_add(length) + .filter(|end| *end <= data.len()) + .ok_or_else(|| crate::Error::DataInvalid { + message: format!( + "BlobDescriptor slice exceeds read data: start={start}, length={length}, actual={}", + data.len() + ), + source: None, + })?; cells[request.row] = ResolvedBlobCell::Value(data.slice(start..end)); } } @@ -130,6 +190,13 @@ enum ResolvedBlobCell { Value(Bytes), } +#[derive(Debug)] +struct BlobReadRequestSpec { + row: usize, + offset: u64, + length: Option, +} + #[derive(Debug, Clone, PartialEq, Eq)] struct BlobReadRequest { row: usize, diff --git a/docs/src/sql.md b/docs/src/sql.md index 0207062c..d550c734 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -294,6 +294,10 @@ the supported directives is rejected. | `__BLOB_DESCRIPTOR_FIELD` | `blob-descriptor-field` | Store serialized `BlobDescriptor` bytes inline | | `__BLOB_VIEW_FIELD` | `blob-view-field` | Store serialized `BlobViewStruct` bytes inline | +For serialized `BlobDescriptor` values supplied by another Paimon engine, +`length = -1` means reading from `offset` to the end of the referenced object. +The offset must be non-negative, and lengths below `-1` are invalid. + The same directives are supported by `ALTER TABLE ... ADD COLUMN`. ### Blob View