QuakeWang commented on code in PR #784:
URL: https://github.com/apache/paimon-rust/pull/784#discussion_r3914749652
##########
crates/paimon/src/arrow/format/blob.rs:
##########
@@ -764,6 +995,277 @@ fn build_blob_array_descriptors(
Ok(BlobReadValue::Array(elements))
}
+async fn fetch_blob_map_values(
+ reader: &dyn FileRead,
+ planned_reads: Vec<PlannedBlobArrayRead>,
+ file_path: &str,
+ descriptor_mode: bool,
+) -> crate::Result<Vec<BlobReadValue>> {
+ futures::stream::iter(planned_reads.into_iter().map(|planned_read| async
move {
+ match planned_read {
+ PlannedBlobArrayRead::Null => Ok(BlobReadValue::Null),
+ PlannedBlobArrayRead::Placeholder =>
Ok(BlobReadValue::Placeholder),
+ PlannedBlobArrayRead::Read(payload_range) => {
+ read_blob_map_entry(reader, payload_range, file_path,
descriptor_mode).await
+ }
+ }
+ }))
+ .buffered(BLOB_READ_CONCURRENCY)
+ .try_collect()
+ .await
+}
+
+async fn read_blob_map_entry(
+ reader: &dyn FileRead,
+ payload_range: Range<u64>,
+ file_path: &str,
+ descriptor_mode: bool,
+) -> crate::Result<BlobReadValue> {
+ let payload_length = payload_range
+ .end
+ .checked_sub(payload_range.start)
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> payload range:
{payload_range:?}"),
+ source: None,
+ })?;
+ if payload_length < BLOB_MAP_MIN_PAYLOAD_SIZE {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "MAP<X, BLOB> payload is too small: expected at least
{BLOB_MAP_MIN_PAYLOAD_SIZE} bytes, got {payload_length}"
+ ),
+ source: None,
+ });
+ }
+
+ let header = read_blob_map_range(
+ reader,
+ payload_range.start..payload_range.start + BLOB_MAP_HEADER_SIZE,
+ "header",
+ )
+ .await?;
+ let magic = i32::from_le_bytes(header[..4].try_into().unwrap());
+ if magic != BLOB_MAP_MAGIC_NUMBER {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Invalid MAP<X, BLOB> payload magic number: expected
{BLOB_MAP_MAGIC_NUMBER}, got {magic}"
+ ),
+ source: None,
+ });
+ }
+ if header[4] != BLOB_MAP_VERSION {
+ return Err(Error::Unsupported {
+ message: format!(
+ "Unsupported MAP<X, BLOB> payload version: expected
{BLOB_MAP_VERSION}, got {}",
+ header[4]
+ ),
+ });
+ }
+ let entry_count = i32::from_le_bytes(header[5..9].try_into().unwrap());
+ if entry_count < 0 {
+ return Err(Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> entry count:
{entry_count}"),
+ source: None,
+ });
+ }
+ let entry_count = entry_count as usize;
+
+ let index_lengths_start = payload_range.end - BLOB_MAP_INDEX_LENGTHS_SIZE;
+ let index_lengths = read_blob_map_range(
+ reader,
+ index_lengths_start..payload_range.end,
+ "index lengths",
+ )
+ .await?;
+ let key_index_length =
i32::from_le_bytes(index_lengths[..4].try_into().unwrap());
+ let value_index_length =
i32::from_le_bytes(index_lengths[4..8].try_into().unwrap());
+ let max_indexes = payload_length - BLOB_MAP_MIN_PAYLOAD_SIZE;
+ if key_index_length < 0 || key_index_length as u64 > max_indexes {
+ return Err(Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> key index length:
{key_index_length}"),
+ source: None,
+ });
+ }
+ if value_index_length < 0 || value_index_length as u64 > max_indexes {
+ return Err(Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> value index length:
{value_index_length}"),
+ source: None,
+ });
+ }
+ let key_index_length = key_index_length as u64;
+ let value_index_length = value_index_length as u64;
+ if key_index_length + value_index_length > max_indexes
+ || entry_count as u64 > key_index_length
+ || entry_count as u64 > value_index_length
+ {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> indexes do not match the
payload".to_string(),
+ source: None,
+ });
+ }
+
+ let value_index_start = index_lengths_start - value_index_length;
+ let key_index_start = value_index_start - key_index_length;
+ let key_index =
+ read_blob_map_range(reader, key_index_start..value_index_start, "key
index").await?;
+ let value_index = read_blob_map_range(
+ reader,
+ value_index_start..index_lengths_start,
+ "value index",
+ )
+ .await?;
+ let key_lengths = decode_delta_varints(&key_index).map_err(|e|
Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> key index: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ let value_lengths = decode_delta_varints(&value_index).map_err(|e|
Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> value index: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ if key_lengths.len() != entry_count || value_lengths.len() != entry_count {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> entry count does not match index
lengths".to_string(),
+ source: None,
+ });
+ }
+
+ let data_start = payload_range.start + BLOB_MAP_HEADER_SIZE;
+ let data_length = key_index_start - data_start;
+ let mut key_data_length = 0u64;
+ for &length in &key_lengths {
+ if length == BLOB_MAP_NULL_LENGTH {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> null keys cannot be represented by
Arrow".to_string(),
+ source: None,
+ });
+ }
+ let length = u64::try_from(length).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> key length: {length}"),
+ source: Some(Box::new(e)),
+ })?;
+ key_data_length = key_data_length
+ .checked_add(length)
+ .filter(|total| *total <= data_length)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "MAP<X, BLOB> key lengths exceed the payload data
length".to_string(),
+ source: None,
+ })?;
+ }
+ let value_data_length = data_length - key_data_length;
+ let mut total_value_length = 0u64;
+ for &length in &value_lengths {
+ if length == BLOB_MAP_NULL_LENGTH {
+ continue;
+ }
+ let length = u64::try_from(length).map_err(|e| Error::DataInvalid {
+ message: format!("Invalid MAP<X, BLOB> value length: {length}"),
+ source: Some(Box::new(e)),
+ })?;
+ total_value_length = total_value_length
+ .checked_add(length)
+ .filter(|total| *total <= value_data_length)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "MAP<X, BLOB> value lengths exceed the payload data
length".to_string(),
+ source: None,
+ })?;
+ }
+ if total_value_length != value_data_length {
+ return Err(Error::DataInvalid {
+ message: "MAP<X, BLOB> key/value lengths do not match the payload
data length"
+ .to_string(),
+ source: None,
+ });
+ }
+
+ let key_data =
+ read_blob_map_range(reader, data_start..data_start + key_data_length,
"key data").await?;
+ let mut keys = Vec::with_capacity(entry_count);
+ let mut cursor = 0usize;
+ let mut unique = std::collections::HashSet::with_capacity(entry_count);
+ for length in key_lengths {
+ let length = length as usize;
+ let end = cursor + length;
+ let key = key_data.slice(cursor..end);
+ if !unique.insert(key.clone()) {
+ return Err(Error::DataInvalid {
+ message: "Invalid MAP<X, BLOB> payload: duplicate
key".to_string(),
+ source: None,
+ });
+ }
+ keys.push(key);
+ cursor = end;
+ }
+
+ let mut value_offset = data_start + key_data_length;
+ let mut reads = Vec::with_capacity(entry_count);
+ for length in value_lengths {
+ if length == BLOB_MAP_NULL_LENGTH {
+ reads.push(None);
+ } else {
+ let length = length as u64;
+ reads.push(Some(value_offset..value_offset + length));
+ value_offset += length;
+ }
+ }
+ let values = if descriptor_mode {
+ reads
+ .into_iter()
+ .map(|range| {
+ range
+ .map(|range| {
+ let offset =
+ i64::try_from(range.start).map_err(|e|
Error::DataInvalid {
+ message: "MAP<X, BLOB> descriptor offset
exceeds i64".to_string(),
+ source: Some(Box::new(e)),
+ })?;
+ let length = i64::try_from(range.end -
range.start).map_err(|e| {
+ Error::DataInvalid {
+ message: "MAP<X, BLOB> descriptor length
exceeds i64".to_string(),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ Ok(Bytes::from(
+ BlobDescriptor::new(file_path.to_string(), offset,
length).serialize(),
+ ))
+ })
+ .transpose()
+ })
+ .collect::<crate::Result<Vec<_>>>()?
+ } else {
+ let payload = read_blob_entry(reader,
blob_entry_range(&payload_range)).await?;
Review Comment:
Please reject oversized inline payloads before this read. total_value_length
is a u64, but these values are later collected into an Arrow BinaryArray, whose
32-bit offsets panic once the total byte length exceeds i32::MAX. ARRAY<BLOB>
already preflights this case; MAP should do the same and also validate the
flattened batch total.
##########
bindings/go/tests/blob_reader_test.go:
##########
@@ -101,6 +104,61 @@ func TestBlobReaderReadBlobAndBatch(t *testing.T) {
}
}
+func TestStringBlobMapDescriptors(t *testing.T) {
Review Comment:
Could this test cover the actual read path instead of starting from a
hand-built Arrow map? As written, it only tests descriptor extraction and
BlobReader. A writer-generated fixture read through NewReadBuilderWithOptions
would also cover the Rust decoder, C Data conversion, and Go API.
--
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]
