This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 3da16e36 perf(variant): Fix slow shredded residual reads (#1166)
3da16e36 is described below
commit 3da16e361d11955c075d2b70344928dce5eea013
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Oct 19 20:15:33 2026 +0800
perf(variant): Fix slow shredded residual reads (#1166)
---
crates/paimon/src/arrow/shredding/variant.rs | 336 ++++++++++++++--
crates/paimon/src/variant.rs | 561 ++++++++++++++++++++++++++-
2 files changed, 867 insertions(+), 30 deletions(-)
diff --git a/crates/paimon/src/arrow/shredding/variant.rs
b/crates/paimon/src/arrow/shredding/variant.rs
index c0e5201d..5f462805 100644
--- a/crates/paimon/src/arrow/shredding/variant.rs
+++ b/crates/paimon/src/arrow/shredding/variant.rs
@@ -35,14 +35,15 @@ use crate::variant::{
};
use crate::{Error, Result};
use arrow_array::{
- new_null_array, Array, ArrayRef, BinaryArray, BooleanArray, Date32Array,
Decimal128Array,
- Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, Int8Array,
LargeBinaryArray,
- ListArray, RecordBatch, StringArray, StructArray,
TimestampMicrosecondArray,
- TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray,
+ new_null_array, types::Int32Type, Array, ArrayRef, BinaryArray,
BooleanArray, Date32Array,
+ Decimal128Array, DictionaryArray, Float32Array, Float64Array, Int16Array,
Int32Array,
+ Int64Array, Int8Array, LargeBinaryArray, ListArray, RecordBatch,
StringArray, StructArray,
+ TimestampMicrosecondArray, TimestampMillisecondArray,
TimestampNanosecondArray,
+ TimestampSecondArray,
};
use arrow_buffer::{BooleanBuffer, NullBuffer, OffsetBuffer, ScalarBuffer};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Fields,
TimeUnit};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex};
const VARIANT_SHREDDING_SCHEMA_OPTION: &str = "variant.shreddingSchema";
@@ -909,18 +910,100 @@ pub(crate) fn assemble_variant_extraction_array(
if let Some(projected) = assemble_typed_shredded_projection(input, fields,
&metadata)? {
return Ok(projected);
}
+ if let Some(projected) = assemble_mixed_shredded_projection(input, fields,
&metadata)? {
+ return Ok(projected);
+ }
assemble_row_wise_variant_extraction(input, fields, &metadata)
}
+/// Keep fully typed fields column-wise even when another requested field
needs the
+/// residual path. A single residual must not force unrelated fields through
the
+/// row-wise Variant decoder.
+fn assemble_mixed_shredded_projection(
+ input: &StructArray,
+ fields: &[DataField],
+ metadata: &[crate::spec::VariantFieldMetadata],
+) -> Result<Option<ArrayRef>> {
+ if !is_shredded_variant_array(input) {
+ return Ok(None);
+ }
+
+ let mut columns = Vec::with_capacity(fields.len());
+ let mut fallback_indices = Vec::new();
+ for (index, (field, field_metadata)) in
fields.iter().zip(metadata).enumerate() {
+ let typed = typed_shredded_leaf(input, field_metadata.path())
+ .and_then(|leaf| cast_typed_leaf(leaf.as_ref(), field.data_type(),
input.nulls()));
+ if typed.is_none() {
+ fallback_indices.push(index);
+ }
+ columns.push(typed);
+ }
+ if fallback_indices.is_empty() || fallback_indices.len() == fields.len() {
+ return Ok(None);
+ }
+ // The row-wise path must validate metadata for strict typed fields even
+ // when every residual field uses try casts.
+ if metadata.iter().any(|field| field.fail_on_error())
+ && fallback_indices
+ .iter()
+ .all(|&index| !metadata[index].fail_on_error())
+ {
+ return Ok(None);
+ }
+
+ let fallback_fields = fallback_indices
+ .iter()
+ .map(|&index| fields[index].clone())
+ .collect::<Vec<_>>();
+ let fallback_metadata = fallback_indices
+ .iter()
+ .map(|&index| metadata[index].clone())
+ .collect::<Vec<_>>();
+ let fallback =
+ assemble_row_wise_variant_extraction(input, &fallback_fields,
&fallback_metadata)?;
+ let fallback = fallback.as_any().downcast_ref::<StructArray>().unwrap();
+ for (fallback_index, &field_index) in fallback_indices.iter().enumerate() {
+ columns[field_index] = Some(fallback.column(fallback_index).clone());
+ }
+
+ let arrow_fields: Fields = fields
+ .iter()
+ .map(|field| {
+ Ok(ArrowField::new(
+ field.name(),
+ paimon_type_to_arrow(field.data_type())?,
+ field.data_type().is_nullable(),
+ ))
+ })
+ .collect::<Result<Vec<_>>>()?
+ .into();
+ let validities = (0..input.len()).map(|row| input.is_valid(row)).collect();
+ let array = StructArray::try_new(
+ arrow_fields,
+ columns.into_iter().map(Option::unwrap).collect(),
+ Some(null_buffer(validities)),
+ )
+ .map_err(|e| Error::UnexpectedError {
+ message: format!("Failed to build mixed Variant extraction
StructArray: {e}"),
+ source: Some(Box::new(e)),
+ })?;
+ Ok(Some(Arc::new(array)))
+}
+
fn assemble_row_wise_variant_extraction(
input: &StructArray,
fields: &[DataField],
metadata: &[crate::spec::VariantFieldMetadata],
) -> Result<ArrayRef> {
- let materialized = materialize_variant_metadata(input)?;
- let input = &materialized;
+ let shredded = is_shredded_variant_array(input);
+ let materialized = if shredded {
+ None
+ } else {
+ Some(materialize_variant_metadata(input)?)
+ };
+ let input = materialized.as_ref().unwrap_or(input);
- let shredded_type = if is_shredded_variant_array(input) {
+ let shredded_type = if shredded {
let DataType::Row(row) = arrow_to_paimon_type(input.data_type(),
true)? else {
unreachable!()
};
@@ -931,6 +1014,10 @@ fn assemble_row_wise_variant_extraction(
};
let mut values_by_field = vec![Vec::with_capacity(input.len());
fields.len()];
let mut validities = Vec::with_capacity(input.len());
+ // Arrow dictionary keys identify immutable metadata values within this
batch.
+ // Once a residual value validates one key, other rows using that key do
not
+ // need to compare and revalidate the same large dictionary bytes.
+ let mut validated_dictionary_keys = HashSet::new();
for row in 0..input.len() {
if input.is_null(row) {
validities.push(false);
@@ -943,7 +1030,9 @@ fn assemble_row_wise_variant_extraction(
validities.push(true);
let shredded = shredded_type
.as_ref()
- .map(|(row_type, _)| struct_row_at(input, row, row_type))
+ .map(|(row_type, schema)| {
+ struct_row_at_skipping(input, row, row_type,
schema.top_level_metadata_idx)
+ })
.transpose()?
.flatten();
let variant = if shredded_type.is_none() {
@@ -951,33 +1040,79 @@ fn assemble_row_wise_variant_extraction(
} else {
None
};
+ let (shredded_metadata, dictionary_key) = if shredded.is_some() {
+ match variant_metadata_at(input, row) {
+ Ok((bytes, key)) => (Some(bytes), key),
+ Err(error) => {
+ if metadata.iter().any(|field| field.fail_on_error()) {
+ return Err(error);
+ }
+ (None, None)
+ }
+ }
+ } else {
+ (None, None)
+ };
+ let mut metadata_validation = shredded_metadata.map(|bytes| {
+ crate::variant::VariantMetadataValidation::new(
+ bytes,
+ dictionary_key.is_some_and(|key|
validated_dictionary_keys.contains(&key)),
+ )
+ });
for (field_idx, field) in fields.iter().enumerate() {
let field_metadata = &metadata[field_idx];
- let extracted = match (&shredded, &shredded_type) {
+ let value = match (&shredded, &shredded_type) {
(Some(shredded), Some((_, schema))) => {
- crate::variant::extract_shredded_path(shredded, schema,
field_metadata.path())
+ let extracted = match &mut metadata_validation {
+ Some(validation) =>
crate::variant::extract_shredded_path_cast(
+ shredded,
+ schema,
+ validation,
+ field_metadata.path(),
+ field.data_type(),
+ field_metadata.fail_on_error(),
+ ),
+ None => Ok(None),
+ };
+ match extracted {
+ Ok(value) => value,
+ Err(_) if !field_metadata.fail_on_error() => None,
+ Err(error) => return Err(error),
+ }
+ }
+ _ => {
+ let extracted = match &variant {
+ Some(variant) => {
+ variant
+ .get_path(field_metadata.path())
+ .and_then(|selected| {
+ selected.map(|value|
value.to_owned_variant()).transpose()
+ })
+ }
+ None => Ok(None),
+ };
+ match extracted {
+ Ok(Some(extracted)) => cast_variant_to_shredded_value(
+ extracted.as_ref()?,
+ field.data_type(),
+ field_metadata.fail_on_error(),
+ )?,
+ Ok(None) => None,
+ Err(_) if !field_metadata.fail_on_error() => None,
+ Err(error) => return Err(error),
+ }
}
- _ => match &variant {
- Some(variant) => variant
- .get_path(field_metadata.path())
- .and_then(|selected| {
- selected.map(|value|
value.to_owned_variant()).transpose()
- }),
- None => Ok(None),
- },
- };
- let value = match extracted {
- Ok(Some(extracted)) => cast_variant_to_shredded_value(
- extracted.as_ref()?,
- field.data_type(),
- field_metadata.fail_on_error(),
- )?,
- Ok(None) => None,
- Err(_) if !field_metadata.fail_on_error() => None,
- Err(error) => return Err(error),
};
values_by_field[field_idx].push(value);
}
+ if metadata_validation
+ .as_ref()
+ .is_some_and(|validation| validation.is_validated())
+ {
+ if let Some(key) = dictionary_key {
+ validated_dictionary_keys.insert(key);
+ }
+ }
}
let arrow_fields: Fields = fields
@@ -1677,6 +1812,65 @@ fn struct_row_at(
Ok(Some(shredded))
}
+fn struct_row_at_skipping(
+ array: &StructArray,
+ row: usize,
+ row_type: &RowType,
+ skip_field: Option<usize>,
+) -> Result<Option<ShreddedRow>> {
+ if array.is_null(row) {
+ return Ok(None);
+ }
+ let mut shredded = ShreddedRow::new(row_type.fields().len());
+ for (field_idx, field) in row_type.fields().iter().enumerate() {
+ if Some(field_idx) == skip_field {
+ continue;
+ }
+ if let Some(value) = value_at(array.column(field_idx).as_ref(), row,
field.data_type())? {
+ shredded.set(field_idx, value)?;
+ }
+ }
+ Ok(Some(shredded))
+}
+
+fn variant_metadata_at(input: &StructArray, row: usize) -> Result<(&[u8],
Option<i32>)> {
+ let column = input
+ .column_by_name(VARIANT_METADATA_FIELD_NAME)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Shredded Variant metadata is missing".to_string(),
+ source: None,
+ })?;
+ if column.is_null(row) {
+ return Err(Error::DataInvalid {
+ message: "Shredded Variant metadata is null".to_string(),
+ source: None,
+ });
+ }
+ if let Some(binary) = column.as_any().downcast_ref::<BinaryArray>() {
+ return Ok((binary.value(row), None));
+ }
+ if let Some(dictionary) =
column.as_any().downcast_ref::<DictionaryArray<Int32Type>>() {
+ let values =
downcast_array::<BinaryArray>(dictionary.values().as_ref(), "Binary")?;
+ let key = dictionary.keys().value(row);
+ let index = usize::try_from(key).map_err(|_| Error::DataInvalid {
+ message: "Shredded Variant metadata dictionary key is
negative".to_string(),
+ source: None,
+ })?;
+ if index >= values.len() || values.is_null(index) {
+ return Err(Error::DataInvalid {
+ message: "Shredded Variant metadata dictionary key is
invalid".to_string(),
+ source: None,
+ });
+ }
+ return Ok((values.value(index), Some(key)));
+ }
+ Err(Error::DataInvalid {
+ message: "Shredded Variant metadata must be Binary or
Dictionary(Int32, Binary)"
+ .to_string(),
+ source: None,
+ })
+}
+
fn value_at(array: &dyn Array, row: usize, data_type: &DataType) ->
Result<Option<ShreddedValue>> {
if array.is_null(row) {
return Ok(None);
@@ -2632,6 +2826,90 @@ mod tests {
}
}
+ #[test]
+ fn mixed_shredded_projection_keeps_typed_fields_column_wise() {
+ let input = shredded_for_test(
+ &[
+ Some(r#"{"a":0,"b":1}"#),
+ Some(r#"{"a":2.5e0,"b":2}"#),
+ Some(r#"{"b":3}"#),
+ None,
+ ],
+ r#"[{"name":"a","type":"DOUBLE"},{"name":"b","type":"BIGINT"}]"#,
+ );
+ let float = || DataType::Float(FloatType::new());
+ let (row_type, metadata) =
+ extraction_for_test(&[(float(), "$.b"), (float(), "$.a"),
(float(), "$.b")]);
+ assert!(
+ assemble_typed_shredded_projection(&input, row_type.fields(),
&metadata)
+ .unwrap()
+ .is_none()
+ );
+ assert!(typed_shredded_leaf(&input, "$.b").is_some());
+ assert!(typed_shredded_leaf(&input, "$.a").is_none());
+ let expected =
+ assemble_row_wise_variant_extraction(&input, row_type.fields(),
&metadata).unwrap();
+ let actual =
+
assemble_variant_extraction_array(&with_dictionary_metadata(&input), &row_type)
+ .unwrap();
+ assert_eq!(actual.as_ref(), expected.as_ref());
+ }
+
+ #[test]
+ fn mixed_shredded_projection_keeps_strict_metadata_error() {
+ let input = shredded_for_test(
+ &[Some(r#"{"a":0,"b":2}"#)],
+ r#"[{"name":"a","type":"DOUBLE"},{"name":"b","type":"BIGINT"}]"#,
+ );
+ let metadata_index = input
+ .fields()
+ .iter()
+ .position(|field| field.name() == VARIANT_METADATA_FIELD_NAME)
+ .unwrap();
+ let mut columns = input.columns().to_vec();
+ columns[metadata_index] =
Arc::new(BinaryArray::from(vec![None::<&[u8]>]));
+ let mut fields = input
+ .fields()
+ .iter()
+ .map(|field| field.as_ref().clone())
+ .collect::<Vec<_>>();
+ fields[metadata_index] =
fields[metadata_index].clone().with_nullable(true);
+ let missing_metadata =
+ StructArray::try_new(fields.into(), columns,
input.nulls().cloned()).unwrap();
+ let row_type = variant_extraction_row(
+ true,
+ vec![
+ (
+ DataType::Float(FloatType::new()),
+ "$.b".to_string(),
+ true,
+ "UTC".to_string(),
+ ),
+ (
+ DataType::Float(FloatType::new()),
+ "$.a".to_string(),
+ false,
+ "UTC".to_string(),
+ ),
+ ],
+ )
+ .unwrap();
+ let metadata = row_type
+ .fields()
+ .iter()
+ .map(|field|
parse_variant_metadata(field.description().unwrap()).unwrap())
+ .collect::<Vec<_>>();
+ assert!(typed_shredded_leaf(&missing_metadata, "$.b").is_some());
+ assert!(assemble_mixed_shredded_projection(
+ &missing_metadata,
+ row_type.fields(),
+ &metadata
+ )
+ .unwrap()
+ .is_none());
+ assert!(assemble_variant_extraction_array(&missing_metadata,
&row_type).is_err());
+ }
+
#[test]
fn inferred_schema_transforms_and_reassembles_variant_batch() {
let logical_fields = vec![
diff --git a/crates/paimon/src/variant.rs b/crates/paimon/src/variant.rs
index 20aa42b2..02c06781 100644
--- a/crates/paimon/src/variant.rs
+++ b/crates/paimon/src/variant.rs
@@ -17,7 +17,8 @@
//! Java-compatible helpers for Paimon Variant values.
-use std::collections::{HashMap, HashSet};
+use std::cell::RefCell;
+use std::collections::{HashMap, HashSet, VecDeque};
use crate::spec::{
ArrayType, BigIntType, BooleanType, DataField, DataType, DateType,
DecimalType, DoubleType,
@@ -36,6 +37,15 @@ const BASIC_TYPE_BITS: u8 = 2;
const BASIC_TYPE_MASK: u8 = 0x3;
const TYPE_INFO_MASK: u8 = 0x3f;
const MAX_SHORT_STR_SIZE: usize = 0x3f;
+const MAX_CACHED_VARIANT_METADATA_BYTES: usize = 512 * 1024;
+const MAX_CACHED_VARIANT_METADATA_TOTAL_BYTES: usize = 1024 * 1024;
+const MAX_CACHED_VARIANT_METADATA_ENTRIES: usize = 4;
+
+thread_local! {
+ /// Exact-byte cache: metadata dictionaries are often repeated across rows
and
+ /// independent projected expressions, but must never be trusted by hash
alone.
+ static VALIDATED_METADATA: RefCell<VecDeque<Vec<u8>>> = const {
RefCell::new(VecDeque::new()) };
+}
const PRIMITIVE: u8 = 0;
const SHORT_STR: u8 = 1;
@@ -128,12 +138,53 @@ pub struct VariantRef<'a> {
pos: usize,
}
+/// Validation state is tied to these exact metadata bytes, not just to a row
or a
+/// dictionary key that may have a different meaning in another batch.
+pub(crate) struct VariantMetadataValidation<'a> {
+ bytes: &'a [u8],
+ validated: bool,
+}
+
+impl<'a> VariantMetadataValidation<'a> {
+ pub(crate) fn new(bytes: &'a [u8], validated: bool) -> Self {
+ Self { bytes, validated }
+ }
+
+ pub(crate) fn is_validated(&self) -> bool {
+ self.validated
+ }
+}
+
impl<'a> VariantRef<'a> {
pub fn new(value: &'a [u8], metadata: &'a [u8], pos: usize) ->
Result<Self> {
validate_payload(value, metadata)?;
Self::new_at(value, metadata, pos)
}
+ /// Validate the shared metadata once per shredded row, while still
validating every
+ /// residual value before exposing a borrowed view of it.
+ fn new_with_cached_metadata(
+ value: &'a [u8],
+ pos: usize,
+ metadata: &mut VariantMetadataValidation<'a>,
+ ) -> Result<Self> {
+ if metadata.bytes.is_empty() || (metadata.bytes[0] & VERSION_MASK) !=
VERSION {
+ return data_invalid("Malformed Variant metadata version");
+ }
+ if value.len() > SIZE_LIMIT || metadata.bytes.len() > SIZE_LIMIT {
+ return data_invalid("Variant value or metadata exceeds constructor
size limit");
+ }
+ if !metadata.validated {
+ validate_metadata(metadata.bytes)?;
+ metadata.validated = true;
+ }
+ let root_size = validate_value(value, metadata.bytes, 0, 0)?;
+ if root_size != value.len() {
+ return data_invalid("Malformed Variant root size");
+ }
+ Self::new_at(value, metadata.bytes, pos)
+ }
+
fn new_at(value: &'a [u8], metadata: &'a [u8], pos: usize) -> Result<Self>
{
check_index(value, pos)?;
Ok(Self {
@@ -442,6 +493,11 @@ pub fn validate_payload(value: &[u8], metadata: &[u8]) ->
Result<()> {
}
fn validate_metadata(metadata: &[u8]) -> Result<()> {
+ if metadata.len() <= MAX_CACHED_VARIANT_METADATA_BYTES
+ && VALIDATED_METADATA.with(|cached| cached.borrow().iter().any(|entry|
entry == metadata))
+ {
+ return Ok(());
+ }
let offset_size = metadata_offset_size(metadata)?;
let dict_size = read_unsigned(metadata, 1, offset_size)?;
let offset_start = 1 + offset_size;
@@ -475,6 +531,20 @@ fn validate_metadata(metadata: &[u8]) -> Result<()> {
return data_invalid("Malformed Variant metadata duplicate key");
}
}
+ if metadata.len() <= MAX_CACHED_VARIANT_METADATA_BYTES {
+ VALIDATED_METADATA.with(|cached| {
+ let mut cached = cached.borrow_mut();
+ let mut total_bytes = cached.iter().map(Vec::len).sum::<usize>();
+ while cached.len() >= MAX_CACHED_VARIANT_METADATA_ENTRIES
+ || total_bytes + metadata.len() >
MAX_CACHED_VARIANT_METADATA_TOTAL_BYTES
+ {
+ if let Some(evicted) = cached.pop_front() {
+ total_bytes -= evicted.len();
+ }
+ }
+ cached.push_back(metadata.to_vec());
+ });
+ }
Ok(())
}
@@ -2689,6 +2759,7 @@ pub(crate) fn rebuild_shredded(
/// Resolve a path directly against a clipped shredded schema. Unrequested
/// physical siblings and a pruned parent `value` are not needed to rebuild
/// the selected subtree. Whole-value rebuilding remains strict.
+#[cfg(test)]
pub(crate) fn extract_shredded_path(
row: &ShreddedRow,
schema: &VariantSchema,
@@ -2701,6 +2772,110 @@ pub(crate) fn extract_shredded_path(
extract_shredded_segments(row, schema, metadata, &parse_path(path)?)
}
+/// Cast a selected shredded path without copying the shared metadata for each
+/// residual field. The validation state is bound to one row's metadata bytes.
+pub(crate) fn extract_shredded_path_cast<'a>(
+ row: &'a ShreddedRow,
+ schema: &VariantSchema,
+ metadata: &mut VariantMetadataValidation<'a>,
+ path: &str,
+ data_type: &DataType,
+ fail_on_error: bool,
+) -> Result<Option<ShreddedValue>> {
+ if !shredded_row_has_value(row, schema) && schema.variant_idx.is_some() {
+ return data_invalid("Malformed shredded Variant: missing value");
+ }
+ extract_shredded_segments_cast(
+ row,
+ schema,
+ metadata,
+ &parse_path(path)?,
+ data_type,
+ fail_on_error,
+ )
+}
+
+fn extract_shredded_segments_cast<'a>(
+ row: &'a ShreddedRow,
+ schema: &VariantSchema,
+ metadata: &mut VariantMetadataValidation<'a>,
+ segments: &[PathSegment],
+ data_type: &DataType,
+ fail_on_error: bool,
+) -> Result<Option<ShreddedValue>> {
+ let Some((segment, remaining)) = segments.split_first() else {
+ // A selected scalar leaf can be read without rebuilding the whole
+ // Variant (and copying its shared metadata) for every field.
+ if schema.scalar_schema.is_some() {
+ let typed = schema.typed_idx.and_then(|index| row.field(index));
+ let residual = optional_binary_field(row, schema.variant_idx)?;
+ match (typed, residual) {
+ (None, Some(value)) => {
+ let variant = VariantRef::new_with_cached_metadata(value,
0, metadata)?;
+ return cast_variant_to_shredded_value(variant, data_type,
fail_on_error);
+ }
+ (None, None) => return Ok(None),
+ _ => {}
+ }
+ }
+ let extracted = extract_shredded_segments(row, schema, metadata.bytes,
segments)?;
+ return extracted
+ .map(|variant| {
+ cast_variant_to_shredded_value(variant.as_ref()?, data_type,
fail_on_error)
+ })
+ .transpose()
+ .map(Option::flatten);
+ };
+ if let Some(typed) = schema.typed_idx.and_then(|index| row.field(index)) {
+ match (segment, typed) {
+ (PathSegment::Key(key), ShreddedValue::Row(object))
+ if schema.object_schema.is_some() =>
+ {
+ if let Some(index) = schema.object_field_index(key) {
+ let Some(ShreddedValue::Row(child)) = object.field(index)
else {
+ return data_invalid("Malformed shredded Variant object
field");
+ };
+ return extract_shredded_segments_cast(
+ child,
+ &schema.object_schema.as_ref().unwrap()[index].schema,
+ metadata,
+ remaining,
+ data_type,
+ fail_on_error,
+ );
+ }
+ }
+ (PathSegment::Index(index), ShreddedValue::List(elements))
+ if schema.array_schema.is_some() =>
+ {
+ return match elements.get(*index) {
+ Some(element) => extract_shredded_segments_cast(
+ element,
+ schema.array_schema.as_ref().unwrap(),
+ metadata,
+ remaining,
+ data_type,
+ fail_on_error,
+ ),
+ None => Ok(None),
+ };
+ }
+ _ => {}
+ }
+ }
+ match optional_binary_field(row, schema.variant_idx)? {
+ Some(value) => {
+ let variant = VariantRef::new_with_cached_metadata(value, 0,
metadata)?;
+ variant
+ .get_path_segments(segments)?
+ .map(|selected| cast_variant_to_shredded_value(selected,
data_type, fail_on_error))
+ .transpose()
+ .map(Option::flatten)
+ }
+ None => Ok(None),
+ }
+}
+
fn extract_shredded_segments(
row: &ShreddedRow,
schema: &VariantSchema,
@@ -3382,6 +3557,114 @@ mod tests {
vec![VERSION, 0, 0]
}
+ #[test]
+ fn metadata_cache_requires_exact_valid_bytes() {
+ let make_large = |suffix: char| {
+ let fields = (0..5342)
+ .map(|i|
format!("\"field_{i:04}_{suffix}_{i:032}_abcdefghijklmnop\":0"))
+ .collect::<Vec<_>>()
+ .join(",");
+ GenericVariant::parse_json(&format!("{{{fields}}}"))
+ .unwrap()
+ .metadata()
+ .to_vec()
+ };
+ let first = make_large('a');
+ let second = make_large('b');
+ let third = GenericVariant::parse_json("42").unwrap();
+ assert_ne!(first, second);
+ assert!(first.len() > 256 * 1024);
+ VALIDATED_METADATA.with(|cached| cached.borrow_mut().clear());
+ validate_metadata(&first).unwrap();
+ validate_metadata(&second).unwrap();
+ validate_metadata(third.metadata()).unwrap();
+ let cached_pointers = VALIDATED_METADATA.with(|cached| {
+ cached
+ .borrow()
+ .iter()
+ .map(|metadata| metadata.as_ptr())
+ .collect::<Vec<_>>()
+ });
+ for _ in 0..100 {
+ validate_metadata(&first).unwrap();
+ validate_metadata(third.metadata()).unwrap();
+ validate_metadata(&second).unwrap();
+ }
+ VALIDATED_METADATA.with(|cached| {
+ assert_eq!(
+ cached
+ .borrow()
+ .iter()
+ .map(|metadata| metadata.as_ptr())
+ .collect::<Vec<_>>(),
+ cached_pointers,
+ "repeated dictionaries must remain cached instead of being
rebuilt"
+ );
+ });
+
+ let mut malformed = first.clone();
+ *malformed.last_mut().unwrap() = 0xff;
+ assert_eq!(malformed.len(), first.len());
+ assert!(validate_metadata(&malformed).is_err());
+ validate_metadata(&first).unwrap();
+ }
+
+ #[test]
+ fn metadata_cache_respects_entry_and_byte_limits() {
+ VALIDATED_METADATA.with(|cached| cached.borrow_mut().clear());
+ for i in 0..5 {
+ let variant =
GenericVariant::parse_json(&format!(r#"{{"field_{i}":0}}"#)).unwrap();
+ validate_metadata(variant.metadata()).unwrap();
+ }
+ VALIDATED_METADATA.with(|cached| {
+ let cached = cached.borrow();
+ assert_eq!(cached.len(), MAX_CACHED_VARIANT_METADATA_ENTRIES);
+ assert!(
+ cached.iter().map(Vec::len).sum::<usize>()
+ <= MAX_CACHED_VARIANT_METADATA_TOTAL_BYTES
+ );
+ });
+
+ let wide = |suffix: usize, count: usize| {
+ let fields = (0..count)
+ .map(|i|
format!(r#""field_{i:04}_{suffix}_{i:032}_abcdefghijklmnop":0"#))
+ .collect::<Vec<_>>()
+ .join(",");
+ GenericVariant::parse_json(&format!("{{{fields}}}"))
+ .unwrap()
+ .metadata()
+ .to_vec()
+ };
+ let large = wide(0, 5_342);
+ let oversize = wide(1, 9_000);
+ assert!(large.len() > MAX_CACHED_VARIANT_METADATA_TOTAL_BYTES / 4);
+ assert!(large.len() <= MAX_CACHED_VARIANT_METADATA_BYTES);
+ assert!(oversize.len() > MAX_CACHED_VARIANT_METADATA_BYTES);
+ VALIDATED_METADATA.with(|cached| cached.borrow_mut().clear());
+ for suffix in 0..4 {
+ let metadata = wide(suffix, 5_342);
+ validate_metadata(&metadata).unwrap();
+ VALIDATED_METADATA.with(|cached| {
+ let cached = cached.borrow();
+ assert!(cached.len() <= MAX_CACHED_VARIANT_METADATA_ENTRIES);
+ assert!(
+ cached.iter().map(Vec::len).sum::<usize>()
+ <= MAX_CACHED_VARIANT_METADATA_TOTAL_BYTES
+ );
+ });
+ }
+ VALIDATED_METADATA.with(|cached| {
+ assert!(!cached.borrow().iter().any(|entry| entry == &large));
+ });
+ let before = VALIDATED_METADATA.with(|cached| cached.borrow().len());
+ validate_metadata(&oversize).unwrap();
+ VALIDATED_METADATA.with(|cached| {
+ let cached = cached.borrow();
+ assert_eq!(cached.len(), before);
+ assert!(!cached.iter().any(|entry| entry == &oversize));
+ });
+ }
+
fn simple_object_shredding_schema() -> VariantSchema {
let configured = DataType::Row(RowType::new(vec![
DataField::new(7, "age".to_string(),
DataType::BigInt(BigIntType::new())),
@@ -3433,6 +3716,282 @@ mod tests {
);
}
+ #[test]
+ fn cached_shredded_residual_cast_matches_existing_extraction() {
+ let configured = DataType::Row(RowType::new(vec![
+ DataField::new(7, "a".to_string(),
DataType::Double(DoubleType::new())),
+ DataField::new(8, "b".to_string(),
DataType::Double(DoubleType::new())),
+ ]));
+ let DataType::Row(physical) =
variant_shredding_type(&configured).unwrap() else {
+ panic!("expected row shredding type");
+ };
+ let schema = build_variant_schema(&physical).unwrap();
+ let target = DataType::Float(FloatType::new());
+ for json in [
+ r#"{"a":0,"b":1}"#,
+ r#"{"a":2.5,"b":0}"#,
+ r#"{"a":"bad","b":1}"#,
+ r#"{"a":null,"b":1}"#,
+ r#"{"b":1}"#,
+ r#"{"c":0}"#,
+ r#"{"d":0}"#,
+ "42",
+ "null",
+ ] {
+ let variant = GenericVariant::parse_json(json).unwrap();
+ let shredded = cast_shredded(&variant, &schema).unwrap();
+ for fail_on_error in [false, true] {
+ let mut metadata_validation =
+ VariantMetadataValidation::new(variant.metadata(), false);
+ for path in ["$.a", "$.b"] {
+ let expected =
+ extract_shredded_path(&shredded, &schema,
path).and_then(|selected| {
+ selected
+ .map(|value| {
+ cast_variant_to_shredded_value(
+ value.as_ref()?,
+ &target,
+ fail_on_error,
+ )
+ })
+ .transpose()
+ .map(Option::flatten)
+ });
+ let actual = extract_shredded_path_cast(
+ &shredded,
+ &schema,
+ &mut metadata_validation,
+ path,
+ &target,
+ fail_on_error,
+ );
+ assert_eq!(
+ actual.is_err(),
+ expected.is_err(),
+ "{json} {path} {fail_on_error}"
+ );
+ if let (Ok(actual), Ok(expected)) = (actual, expected) {
+ assert_eq!(actual, expected, "{json} {path}
{fail_on_error}");
+ }
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn
cached_shredded_nested_object_and_array_paths_match_existing_extraction() {
+ let configured = DataType::Row(RowType::new(vec![
+ DataField::new(
+ 7,
+ "object".to_string(),
+ DataType::Row(RowType::new(vec![DataField::new(
+ 8,
+ "a".to_string(),
+ DataType::Double(DoubleType::new()),
+ )])),
+ ),
+ DataField::new(
+ 9,
+ "items".to_string(),
+
DataType::Array(ArrayType::new(DataType::Double(DoubleType::new()))),
+ ),
+ ]));
+ let DataType::Row(physical) =
variant_shredding_type(&configured).unwrap() else {
+ panic!("expected row shredding type");
+ };
+ let schema = build_variant_schema(&physical).unwrap();
+ let target = DataType::Float(FloatType::new());
+ for json in [
+ r#"{"object":{"a":0,"extra":1},"items":[0,2.5]}"#,
+ r#"{"object":{"a":2.5},"items":[3.5,0]}"#,
+ r#"{"object":null,"items":[]}"#,
+ ] {
+ let variant = GenericVariant::parse_json(json).unwrap();
+ let shredded = cast_shredded(&variant, &schema).unwrap();
+ let mut metadata =
VariantMetadataValidation::new(variant.metadata(), false);
+ for path in ["$.object.a", "$.object.extra", "$.items[0]",
"$.items[1]"] {
+ let expected = extract_shredded_path(&shredded, &schema, path)
+ .unwrap()
+ .and_then(|selected| {
+
cast_variant_to_shredded_value(selected.as_ref().unwrap(), &target, false)
+ .unwrap()
+ });
+ let actual = extract_shredded_path_cast(
+ &shredded,
+ &schema,
+ &mut metadata,
+ path,
+ &target,
+ false,
+ )
+ .unwrap();
+ assert_eq!(actual, expected, "{json} {path}");
+ }
+ }
+ }
+
+ #[test]
+ fn cached_shredded_residual_preserves_malformed_and_conflicting_values() {
+ let DataType::Row(physical) =
+
variant_shredding_type(&DataType::Double(DoubleType::new())).unwrap()
+ else {
+ panic!("expected row shredding type");
+ };
+ let schema = build_variant_schema(&physical).unwrap();
+ let variant = GenericVariant::parse_json("0").unwrap();
+ let residual = cast_shredded(&variant, &schema).unwrap();
+ let target = DataType::Float(FloatType::new());
+ let compare = |row: &ShreddedRow, metadata_bytes: &[u8]| {
+ let expected = extract_shredded_path(row, &schema,
"$").and_then(|selected| {
+ selected
+ .map(|value|
cast_variant_to_shredded_value(value.as_ref()?, &target, false))
+ .transpose()
+ .map(Option::flatten)
+ });
+ let mut metadata = VariantMetadataValidation::new(metadata_bytes,
false);
+ let actual =
+ extract_shredded_path_cast(row, &schema, &mut metadata, "$",
&target, false);
+ assert_eq!(actual.is_err(), expected.is_err());
+ if let (Ok(actual), Ok(expected)) = (actual, expected) {
+ assert_eq!(actual, expected);
+ }
+ };
+ compare(&residual, variant.metadata());
+
+ let mut bad_value = residual.clone();
+ bad_value
+ .set(
+ schema.variant_idx.unwrap(),
+ ShreddedValue::Binary(vec![0xff]),
+ )
+ .unwrap();
+ compare(&bad_value, variant.metadata());
+
+ let mut bad_metadata = residual.clone();
+ let mut bytes = variant.metadata().to_vec();
+ bytes.pop();
+ bad_metadata
+ .set(
+ schema.top_level_metadata_idx.unwrap(),
+ ShreddedValue::Binary(bytes.clone()),
+ )
+ .unwrap();
+ compare(&bad_metadata, &bytes);
+
+ let mut both = residual.clone();
+ both.set(schema.typed_idx.unwrap(), ShreddedValue::Float64(1.5))
+ .unwrap();
+ compare(&both, variant.metadata());
+ }
+
+ /// Decode-only comparison, deliberately excluding Parquet I/O and
DataLoader work.
+ /// Run with `cargo test --release -p paimon --lib
benchmark_wide_shredded_residual_cast -- --ignored --nocapture`.
+ #[test]
+ #[ignore = "local synthetic performance comparison"]
+ fn benchmark_wide_shredded_residual_cast() {
+ use std::hint::black_box;
+ use std::time::{Duration, Instant};
+
+ let configured = DataType::Row(RowType::new(
+ (0..200)
+ .map(|i| {
+ DataField::new(i + 1, format!("f{i}"),
DataType::Double(DoubleType::new()))
+ })
+ .collect(),
+ ));
+ let DataType::Row(physical) =
variant_shredding_type(&configured).unwrap() else {
+ panic!("expected row shredding type");
+ };
+ let schema = build_variant_schema(&physical).unwrap();
+ let fields = (0..5_342)
+ .map(|i| {
+ if !(27..200).contains(&i) {
+ format!(r#""f{i}":0"#)
+ } else {
+ format!(r#""f{i}":{i}.5"#)
+ }
+ })
+ .collect::<Vec<_>>()
+ .join(",");
+ let variant =
GenericVariant::parse_json(&format!("{{{fields}}}")).unwrap();
+ let shredded = cast_shredded(&variant, &schema).unwrap();
+ let Some(ShreddedValue::Row(object)) =
+ schema.typed_idx.and_then(|index| shredded.field(index))
+ else {
+ panic!("expected shredded object");
+ };
+ for field in schema.object_schema.as_ref().unwrap().iter().take(27) {
+ let index = schema.object_field_index(&field.field_name).unwrap();
+ let Some(ShreddedValue::Row(child)) = object.field(index) else {
+ panic!("expected shredded child");
+ };
+ assert!(child.field(field.schema.variant_idx.unwrap()).is_some());
+ }
+ let paths = (0..27).map(|i| format!("$.f{i}")).collect::<Vec<_>>();
+ let target = DataType::Float(FloatType::new());
+ for path in &paths {
+ let old = extract_shredded_path(&shredded, &schema, path)
+ .unwrap()
+ .and_then(|value| {
+ cast_variant_to_shredded_value(value.as_ref().unwrap(),
&target, false).unwrap()
+ });
+ let mut metadata =
VariantMetadataValidation::new(variant.metadata(), false);
+ let new =
+ extract_shredded_path_cast(&shredded, &schema, &mut metadata,
path, &target, false)
+ .unwrap();
+ assert_eq!(new, old);
+ }
+
+ let mut old_times = Vec::new();
+ let mut new_times = Vec::new();
+ for round in 0..5 {
+ let measure = |optimized: bool| {
+ let start = Instant::now();
+ for _ in 0..512 {
+ let mut metadata =
VariantMetadataValidation::new(variant.metadata(), false);
+ for path in &paths {
+ let value = if optimized {
+ extract_shredded_path_cast(
+ &shredded,
+ &schema,
+ &mut metadata,
+ path,
+ &target,
+ false,
+ )
+ .unwrap()
+ } else {
+ extract_shredded_path(&shredded, &schema, path)
+ .unwrap()
+ .and_then(|value| {
+ cast_variant_to_shredded_value(
+ value.as_ref().unwrap(),
+ &target,
+ false,
+ )
+ .unwrap()
+ })
+ };
+ black_box(value);
+ }
+ }
+ start.elapsed()
+ };
+ if round % 2 == 0 {
+ old_times.push(measure(false));
+ new_times.push(measure(true));
+ } else {
+ new_times.push(measure(true));
+ old_times.push(measure(false));
+ }
+ }
+ old_times.sort();
+ new_times.sort();
+ let old: Duration = old_times[old_times.len() / 2];
+ let new: Duration = new_times[new_times.len() / 2];
+ println!("512 synthetic rows, 5342 metadata keys, 27 residual fields
of 200 projected: legacy={old:?}, direct={new:?}");
+ }
+
#[test]
fn variant_shredding_schema_matches_java_shape() {
let configured = DataType::Row(RowType::new(vec