You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org> on 2026/08/29 05:08:31 UTC
[PR] feat(blob): add standalone BlobDescriptor reader [paimon-rust]
XiaoHongbo-Hope opened a new pull request, #762:
URL: https://github.com/apache/paimon-rust/pull/762
## Summary
- Add a table-independent Rust `BlobReader` for serialized `BlobDescriptor` values.
- Reuse existing validation, FileIO, URI grouping, range merging, and bounded concurrency.
- Preserve input order and sanitize credentials from errors.
This is the Rust core split from draft #761. C and Go bindings are intentionally outside this PR and will follow separately.
## Testing
- `cargo test -p paimon table::blob_resolver::tests --lib`
- `cargo clippy -p paimon --all-targets -- -D warnings`
- `cargo fmt --all -- --check`
- `git diff --check apache/main...HEAD`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(blob): add standalone BlobDescriptor reader [paimon-rust]
Posted by "XiaoHongbo-Hope (via GitHub)" <gi...@apache.org>.
XiaoHongbo-Hope commented on code in PR #762:
URL: https://github.com/apache/paimon-rust/pull/762#discussion_r3885996851
##########
crates/paimon/src/table/blob_resolver.rs:
##########
@@ -32,6 +32,152 @@ pub(crate) const BLOB_DESCRIPTOR_READ_CONCURRENCY: usize = 8;
const BLOB_DESCRIPTOR_READ_BYTE_UNIT: u64 = 1024 * 1024;
const BLOB_DESCRIPTOR_READ_MAX_IN_FLIGHT_BYTES: u64 = 64 * 1024 * 1024;
+/// Reads serialized [`BlobDescriptor`] values without requiring a table.
+#[derive(Clone, Debug, Default)]
+pub struct BlobReader {
+ storage_options: HashMap<String, String>,
+ file_io: Option<FileIO>,
+}
+
+impl BlobReader {
+ pub fn new(storage_options: HashMap<String, String>) -> Self {
+ Self {
+ storage_options,
+ file_io: None,
+ }
+ }
+
+ /// Create a reader that reuses an existing FileIO.
+ pub fn from_file_io(file_io: FileIO) -> Self {
+ Self {
+ storage_options: HashMap::new(),
+ file_io: Some(file_io),
+ }
+ }
+
+ /// Read a descriptor batch in input order.
+ pub async fn read_blobs(&self, descriptors: &[Vec<u8>]) -> Result<Vec<Vec<u8>>> {
+ let mut by_uri = HashMap::<String, Vec<(usize, &[u8])>>::new();
+ for (index, bytes) in descriptors.iter().enumerate() {
+ let descriptor = BlobDescriptor::deserialize(bytes)
+ .map_err(|error| blob_error_with_context(error, &[index], None))?;
+ descriptor.range_spec().map_err(|error| {
+ blob_error_with_context(error, &[index], Some(descriptor.uri()))
+ })?;
+ by_uri
+ .entry(descriptor.uri().to_string())
+ .or_default()
+ .push((index, bytes));
+ }
+
+ let limiter = BlobReadLimiter::new();
+ let groups: Vec<Vec<(usize, Vec<u8>)>> = stream::iter(by_uri)
+ .map(|(uri, entries)| {
+ let limiter = limiter.clone();
+ async move {
+ let indices = entries.iter().map(|(index, _)| *index).collect::<Vec<_>>();
+ let file_io = match &self.file_io {
+ Some(file_io) => file_io.clone(),
+ None => FileIO::from_path(&uri)
+ .and_then(|builder| {
+ builder.with_props(self.storage_options.iter()).build()
+ })
+ .map_err(|error| {
+ blob_error_with_context(error, &indices, Some(&uri))
+ })?,
+ };
+ let mut builder = BinaryBuilder::with_capacity(entries.len(), 0);
+ for (_, descriptor) in &entries {
+ builder.append_value(descriptor);
+ }
+ let resolved = resolve_blob_column(&builder.finish(), &file_io, limiter)
Review Comment:
Fixed in 7ae349f. Parsed descriptors are normalized to the current v2 representation before entering the mixed-column resolver, and a Java v1 range-read regression test now verifies the referenced bytes are returned.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(blob): add standalone BlobDescriptor reader [paimon-rust]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #762:
URL: https://github.com/apache/paimon-rust/pull/762
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(blob): add standalone BlobDescriptor reader [paimon-rust]
Posted by "QuakeWang (via GitHub)" <gi...@apache.org>.
QuakeWang commented on code in PR #762:
URL: https://github.com/apache/paimon-rust/pull/762#discussion_r3885983627
##########
crates/paimon/src/table/blob_resolver.rs:
##########
@@ -32,6 +32,152 @@ pub(crate) const BLOB_DESCRIPTOR_READ_CONCURRENCY: usize = 8;
const BLOB_DESCRIPTOR_READ_BYTE_UNIT: u64 = 1024 * 1024;
const BLOB_DESCRIPTOR_READ_MAX_IN_FLIGHT_BYTES: u64 = 64 * 1024 * 1024;
+/// Reads serialized [`BlobDescriptor`] values without requiring a table.
+#[derive(Clone, Debug, Default)]
+pub struct BlobReader {
+ storage_options: HashMap<String, String>,
+ file_io: Option<FileIO>,
+}
+
+impl BlobReader {
+ pub fn new(storage_options: HashMap<String, String>) -> Self {
+ Self {
+ storage_options,
+ file_io: None,
+ }
+ }
+
+ /// Create a reader that reuses an existing FileIO.
+ pub fn from_file_io(file_io: FileIO) -> Self {
+ Self {
+ storage_options: HashMap::new(),
+ file_io: Some(file_io),
+ }
+ }
+
+ /// Read a descriptor batch in input order.
+ pub async fn read_blobs(&self, descriptors: &[Vec<u8>]) -> Result<Vec<Vec<u8>>> {
+ let mut by_uri = HashMap::<String, Vec<(usize, &[u8])>>::new();
+ for (index, bytes) in descriptors.iter().enumerate() {
+ let descriptor = BlobDescriptor::deserialize(bytes)
+ .map_err(|error| blob_error_with_context(error, &[index], None))?;
+ descriptor.range_spec().map_err(|error| {
+ blob_error_with_context(error, &[index], Some(descriptor.uri()))
+ })?;
+ by_uri
+ .entry(descriptor.uri().to_string())
+ .or_default()
+ .push((index, bytes));
+ }
+
+ let limiter = BlobReadLimiter::new();
+ let groups: Vec<Vec<(usize, Vec<u8>)>> = stream::iter(by_uri)
+ .map(|(uri, entries)| {
+ let limiter = limiter.clone();
+ async move {
+ let indices = entries.iter().map(|(index, _)| *index).collect::<Vec<_>>();
+ let file_io = match &self.file_io {
+ Some(file_io) => file_io.clone(),
+ None => FileIO::from_path(&uri)
+ .and_then(|builder| {
+ builder.with_props(self.storage_options.iter()).build()
+ })
+ .map_err(|error| {
+ blob_error_with_context(error, &indices, Some(&uri))
+ })?,
+ };
+ let mut builder = BinaryBuilder::with_capacity(entries.len(), 0);
+ for (_, descriptor) in &entries {
+ builder.append_value(descriptor);
+ }
+ let resolved = resolve_blob_column(&builder.finish(), &file_io, limiter)
Review Comment:
Valid Java v1 descriptors are silently returned as raw bytes here. `read_blobs` accepts them via `BlobDescriptor::deserialize`, but `resolve_blob_column` only recognizes the v2 magic header, so the referenced range is never resolved. Please use a strict descriptor path (or normalize to v2) and add a v1 regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org