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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
