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