You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/10/10 03:48:39 UTC
(paimon-rust) branch main updated: [core] Skip shredded Variant residuals that statistics prove null (#1080)
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 f32f5fda [core] Skip shredded Variant residuals that statistics prove null (#1080)
f32f5fda is described below
commit f32f5fdabfd323ff3d9693f2531a4fdbace99352
Author: Dapeng Sun(孙大鹏) <da...@alibaba-inc.com>
AuthorDate: Sat Oct 10 11:48:33 2026 +0800
[core] Skip shredded Variant residuals that statistics prove null (#1080)
---
crates/paimon/src/arrow/format/parquet.rs | 439 +++++++++++++++++++++++++++++-
1 file changed, 425 insertions(+), 14 deletions(-)
diff --git a/crates/paimon/src/arrow/format/parquet.rs b/crates/paimon/src/arrow/format/parquet.rs
index cbc434a4..33ddbfae 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -969,6 +969,7 @@ impl FormatFileReader for ParquetFormatReader {
&scan_fields,
map_read_plan.as_deref(),
preds,
+ &null_shredded_residuals(batch_stream_builder.metadata()),
);
batch_stream_builder = batch_stream_builder.with_projection(mask.clone());
@@ -1322,17 +1323,67 @@ impl FormatFileReader for ParquetFormatReader {
}
}
+/// Shredded object-field residuals (`typed_value.<key>.value`) that every row group's
+/// statistics prove null; candidates for the extraction-time skip below.
+fn null_shredded_residuals(metadata: &ParquetMetaData) -> std::collections::HashSet<Vec<String>> {
+ let schema = metadata.file_metadata().schema_descr();
+ (0..schema.num_columns())
+ .filter(|&leaf| {
+ let column = schema.column(leaf);
+ let parts = column.path().parts();
+ parts.len() >= 3
+ && parts[parts.len() - 1] == "value"
+ && parts[parts.len() - 3] == "typed_value"
+ && metadata.num_row_groups() > 0
+ && metadata.row_groups().iter().all(|row_group| {
+ let chunk = row_group.column(leaf);
+ chunk
+ .statistics()
+ .and_then(|statistics| statistics.null_count_opt())
+ .is_some_and(|nulls| i64::try_from(nulls).ok() == Some(chunk.num_values()))
+ })
+ })
+ .map(|leaf| schema.column(leaf).path().parts().to_vec())
+ .collect()
+}
+
+/// Of the projected all-null residuals, the ones Variant extraction may leave
+/// unread while keeping each field's structural definition levels intact.
+fn skippable_residuals(
+ projected: &std::collections::HashSet<Vec<String>>,
+ candidates: &[Vec<String>],
+) -> std::collections::HashSet<Vec<String>> {
+ let mut candidates = candidates.to_vec();
+ candidates.sort_by_key(|path| std::cmp::Reverse(path.len()));
+ let mut skippable = std::collections::HashSet::new();
+ for candidate in &candidates {
+ let group = &candidate[..candidate.len() - 1];
+ let keeps_field_presence = projected.iter().any(|leaf| {
+ leaf.len() > group.len()
+ && leaf[..group.len()] == group[..]
+ && leaf[group.len()] == "typed_value"
+ && !skippable.contains(leaf)
+ });
+ if keeps_field_presence {
+ skippable.insert(candidate.clone());
+ }
+ }
+ skippable
+}
+
fn read_type_projection_mask(
parquet_schema: &parquet::schema::types::SchemaDescriptor,
read_fields: &[DataField],
map_plan: Option<&MapShreddingReadPlan>,
predicates: &[Predicate],
+ null_residuals: &std::collections::HashSet<Vec<String>>,
) -> ProjectionMask {
fn requested_paths(
field: &DataField,
physical: &parquet::schema::types::Type,
path: &mut Vec<String>,
output: &mut std::collections::HashSet<Vec<String>>,
+ null_residuals: &std::collections::HashSet<Vec<String>>,
) {
if crate::spec::is_map_selected_keys_field(field) {
// Selected-key ROW children name logical MAP keys, not physical
@@ -1344,15 +1395,24 @@ fn read_type_projection_mask(
let data_type = field.data_type();
if let DataType::Row(row) = data_type {
if crate::spec::is_variant_extraction_row_type(data_type) {
- output.extend(
- super::variant_projection::projected_paths(row, physical)
- .into_iter()
- .map(|parts| {
- let mut full = path.clone();
- full.extend(parts);
- full
- }),
- );
+ let projected = super::variant_projection::projected_paths(row, physical);
+ let candidates = null_residuals
+ .iter()
+ .filter(|residual| {
+ residual.starts_with(path.as_slice())
+ && projected.contains(&residual[path.len()..])
+ })
+ .map(|residual| residual[path.len()..].to_vec())
+ .collect::<Vec<_>>();
+ let skippable = skippable_residuals(&projected, &candidates);
+ for parts in projected {
+ if skippable.contains(&parts) {
+ continue;
+ }
+ let mut full = path.clone();
+ full.extend(parts);
+ output.insert(full);
+ }
return;
}
if !physical.is_primitive() {
@@ -1363,7 +1423,7 @@ fn read_type_projection_mask(
.find(|child| child.name() == field.name())
{
path.push(child.name().to_string());
- requested_paths(field, child, path, output);
+ requested_paths(field, child, path, output, null_residuals);
path.pop();
}
}
@@ -1401,7 +1461,7 @@ fn read_type_projection_mask(
path.pop();
}
} else {
- requested_paths(field, root, &mut path, &mut selected);
+ requested_paths(field, root, &mut path, &mut selected, null_residuals);
}
}
let leaves = (0..parquet_schema.num_columns())
@@ -3301,7 +3361,8 @@ mod tests {
DataType::Row(RowType::new(vec![nested])),
),
];
- let mask = super::read_type_projection_mask(&descriptor, &fields, None, &[]);
+ let mask =
+ super::read_type_projection_mask(&descriptor, &fields, None, &[], &Default::default());
for index in 0..4 {
assert!(mask.leaf_included(index), "Missing MAP leaf {index}");
}
@@ -3333,17 +3394,65 @@ mod tests {
let selected =
crate::spec::project_read_type(&fields, &[vec!["profile".into(), "score".into()]])
.unwrap();
- let mask = super::read_type_projection_mask(&descriptor, &selected, None, &[]);
+ let mask = super::read_type_projection_mask(
+ &descriptor,
+ &selected,
+ None,
+ &[],
+ &Default::default(),
+ );
assert!(mask.leaf_included(0));
assert!(!mask.leaf_included(1));
assert!(!mask.leaf_included(2));
let predicate = PredicateBuilder::new(&fields).is_null("profile").unwrap();
- let widened = super::read_type_projection_mask(&descriptor, &selected, None, &[predicate]);
+ let widened = super::read_type_projection_mask(
+ &descriptor,
+ &selected,
+ None,
+ &[predicate],
+ &Default::default(),
+ );
assert!(widened.leaf_included(0));
assert!(widened.leaf_included(1));
assert!(!widened.leaf_included(2));
}
+ #[test]
+ fn null_residual_skip_keeps_field_structure() {
+ fn leaf(parts: &[&str]) -> Vec<String> {
+ parts.iter().map(|part| part.to_string()).collect()
+ }
+ let projected = [
+ leaf(&["metadata"]),
+ leaf(&["typed_value", "obj", "value"]),
+ leaf(&["typed_value", "obj", "typed_value", "x", "value"]),
+ leaf(&["typed_value", "y", "value"]),
+ leaf(&["typed_value", "y", "typed_value"]),
+ ]
+ .into_iter()
+ .collect::<std::collections::HashSet<_>>();
+ let candidates = vec![
+ leaf(&["typed_value", "obj", "value"]),
+ leaf(&["typed_value", "obj", "typed_value", "x", "value"]),
+ leaf(&["typed_value", "y", "value"]),
+ ];
+ let skippable = super::skippable_residuals(&projected, &candidates);
+ // y keeps its typed leaf, so its all-null residual is skippable.
+ assert!(skippable.contains(&leaf(&["typed_value", "y", "value"])));
+ // x is residual-only: its value is the group's sole leaf and stays.
+ assert!(!skippable.contains(&leaf(&["typed_value", "obj", "typed_value", "x", "value"])));
+ // obj's residual is skippable because x.value survives below it.
+ assert!(skippable.contains(&leaf(&["typed_value", "obj", "value"])));
+ let no_typed_leaf = projected
+ .iter()
+ .filter(|path| path.as_slice() != leaf(&["typed_value", "y", "typed_value"]).as_slice())
+ .cloned()
+ .collect::<std::collections::HashSet<_>>();
+ let without = super::skippable_residuals(&no_typed_leaf, &candidates);
+ assert!(!without.contains(&leaf(&["typed_value", "y", "value"])));
+ assert!(without.contains(&leaf(&["typed_value", "obj", "value"])));
+ }
+
fn test_fields() -> Vec<DataField> {
vec![
DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
@@ -7876,6 +7985,308 @@ mod tests {
assert_eq!(actual.as_ref(), expected.as_ref());
}
+ #[tokio::test]
+ async fn shredded_variant_extraction_skips_null_residuals() {
+ let fields = vec![DataField::new(
+ 0,
+ "v".to_string(),
+ DataType::Variant(VariantType::new()),
+ )];
+ let options = HashMap::from([(
+ "variant.shreddingSchema".to_string(),
+ r#"{"type":"ROW","fields":[{"name":"v","type":{"type":"ROW","fields":[{"name":"x","type":"DOUBLE"},{"name":"y","type":"BIGINT"}]}}]}"#.to_string(),
+ )]);
+ let variants = [r#"{"x":1.5e0,"y":2,"z":"rest"}"#, r#"{"x":4e0,"y":"text"}"#]
+ .iter()
+ .map(|json| GenericVariant::parse_json(json).unwrap())
+ .collect::<Vec<_>>();
+ let ArrowDataType::Struct(variant_fields) = variant_arrow_type() else {
+ unreachable!()
+ };
+ let column = StructArray::new(
+ variant_fields,
+ vec![
+ Arc::new(BinaryArray::from_iter_values(
+ variants.iter().map(|v| v.value()),
+ )),
+ Arc::new(BinaryArray::from_iter_values(
+ variants.iter().map(|v| v.metadata()),
+ )),
+ ],
+ None,
+ );
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(column.clone())]).unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = format!("memory:/null_residuals_{}.parquet", uuid::Uuid::new_v4());
+ let mut writer = create_format_writer(
+ &file_io.new_output(&path).unwrap(),
+ schema,
+ "zstd",
+ 1,
+ None,
+ Some(&fields),
+ Some(&options),
+ )
+ .await
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ writer.close().await.unwrap();
+ let data = file_io.new_input(&path).unwrap().read().await.unwrap();
+
+ let double = || DataType::Double(crate::spec::DoubleType::new());
+ let row_type = crate::spec::variant_extraction_row(
+ true,
+ ["$.x", "$.y"]
+ .iter()
+ .map(|path| (double(), path.to_string(), false, "UTC".to_string()))
+ .collect::<Vec<_>>(),
+ )
+ .unwrap();
+ let read_fields = vec![DataField::new(
+ 0,
+ "v".to_string(),
+ DataType::Row(row_type.clone()),
+ )];
+ let batches = ParquetFormatReader::default()
+ .read_batch_stream(
+ Box::new(TrackingFileRead::new(data.clone())),
+ data.len() as u64,
+ &read_fields,
+ None,
+ None,
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let read = batches[0]
+ .column_by_name("v")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let typed = read
+ .column_by_name("typed_value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let group = |key: &str| {
+ typed
+ .column_by_name(key)
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap()
+ .clone()
+ };
+ assert!(group("x").column_by_name("value").is_none());
+ assert!(group("y").column_by_name("value").is_some());
+ let expected =
+ crate::arrow::shredding::variant::assemble_variant_extraction_array(&column, &row_type)
+ .unwrap();
+ let actual =
+ crate::arrow::shredding::variant::assemble_variant_extraction_array(read, &row_type)
+ .unwrap();
+ assert_eq!(actual.as_ref(), expected.as_ref());
+ }
+
+ #[tokio::test]
+ async fn shredded_variant_extraction_preserves_null_groups() {
+ let fields = vec![DataField::new(
+ 0,
+ "v".to_string(),
+ DataType::Variant(VariantType::new()),
+ )];
+ let options = HashMap::from([(
+ "variant.shreddingSchema".to_string(),
+ r#"{"type":"ROW","fields":[{"name":"v","type":{"type":"ROW","fields":[{"name":"obj","type":{"type":"ROW","fields":[{"name":"x","type":"VARIANT"}]}},{"name":"y","type":"BIGINT"}]}}]}"#.to_string(),
+ )]);
+ let variants = [r#"{"obj":{},"y":2}"#, r#"{"y":3}"#]
+ .iter()
+ .map(|json| GenericVariant::parse_json(json).unwrap())
+ .collect::<Vec<_>>();
+ let ArrowDataType::Struct(variant_fields) = variant_arrow_type() else {
+ unreachable!()
+ };
+ let column = StructArray::new(
+ variant_fields,
+ vec![
+ Arc::new(BinaryArray::from_iter_values(
+ variants.iter().map(|v| v.value()),
+ )),
+ Arc::new(BinaryArray::from_iter_values(
+ variants.iter().map(|v| v.metadata()),
+ )),
+ ],
+ None,
+ );
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(column.clone())]).unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = format!("memory:/null_groups_{}.parquet", uuid::Uuid::new_v4());
+ let mut writer = create_format_writer(
+ &file_io.new_output(&path).unwrap(),
+ schema,
+ "zstd",
+ 1,
+ None,
+ Some(&fields),
+ Some(&options),
+ )
+ .await
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ writer.close().await.unwrap();
+ let data = file_io.new_input(&path).unwrap().read().await.unwrap();
+ let row_type = crate::spec::variant_extraction_row(
+ true,
+ vec![
+ (
+ DataType::VarChar(crate::spec::VarCharType::string_type()),
+ "$.obj".to_string(),
+ false,
+ "UTC".to_string(),
+ ),
+ (
+ DataType::Double(crate::spec::DoubleType::new()),
+ "$.y".to_string(),
+ false,
+ "UTC".to_string(),
+ ),
+ ],
+ )
+ .unwrap();
+ let read_fields = vec![DataField::new(
+ 0,
+ "v".to_string(),
+ DataType::Row(row_type.clone()),
+ )];
+ let batches = ParquetFormatReader::default()
+ .read_batch_stream(
+ Box::new(TrackingFileRead::new(data.clone())),
+ data.len() as u64,
+ &read_fields,
+ None,
+ None,
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let read = batches[0]
+ .column_by_name("v")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let expected =
+ crate::arrow::shredding::variant::assemble_variant_extraction_array(&column, &row_type)
+ .unwrap();
+ let actual =
+ crate::arrow::shredding::variant::assemble_variant_extraction_array(read, &row_type)
+ .unwrap();
+ assert_eq!(actual.as_ref(), expected.as_ref());
+ }
+
+ #[tokio::test]
+ async fn shredded_variant_extraction_preserves_absent_variant_child() {
+ let fields = vec![DataField::new(
+ 0,
+ "v".to_string(),
+ DataType::Variant(VariantType::new()),
+ )];
+ let options = HashMap::from([(
+ "variant.shreddingSchema".to_string(),
+ r#"{"type":"ROW","fields":[{"name":"v","type":{"type":"ROW","fields":[{"name":"w","type":"VARIANT"},{"name":"y","type":"BIGINT"}]}}]}"#.to_string(),
+ )]);
+ let variants = [r#"{"y":2}"#, r#"{"y":3}"#]
+ .iter()
+ .map(|json| GenericVariant::parse_json(json).unwrap())
+ .collect::<Vec<_>>();
+ let ArrowDataType::Struct(variant_fields) = variant_arrow_type() else {
+ unreachable!()
+ };
+ let column = StructArray::new(
+ variant_fields,
+ vec![
+ Arc::new(BinaryArray::from_iter_values(
+ variants.iter().map(|v| v.value()),
+ )),
+ Arc::new(BinaryArray::from_iter_values(
+ variants.iter().map(|v| v.metadata()),
+ )),
+ ],
+ None,
+ );
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(column.clone())]).unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = format!("memory:/absent_child_{}.parquet", uuid::Uuid::new_v4());
+ let mut writer = create_format_writer(
+ &file_io.new_output(&path).unwrap(),
+ schema,
+ "zstd",
+ 1,
+ None,
+ Some(&fields),
+ Some(&options),
+ )
+ .await
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ writer.close().await.unwrap();
+ let data = file_io.new_input(&path).unwrap().read().await.unwrap();
+
+ let row_type = crate::spec::variant_extraction_row(
+ true,
+ vec![(
+ DataType::Variant(VariantType::new()),
+ "$.w".to_string(),
+ false,
+ "UTC".to_string(),
+ )],
+ )
+ .unwrap();
+ let read_fields = vec![DataField::new(
+ 0,
+ "v".to_string(),
+ DataType::Row(row_type.clone()),
+ )];
+ let batches = ParquetFormatReader::default()
+ .read_batch_stream(
+ Box::new(TrackingFileRead::new(data.clone())),
+ data.len() as u64,
+ &read_fields,
+ None,
+ None,
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ let read = batches[0]
+ .column_by_name("v")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let expected =
+ crate::arrow::shredding::variant::assemble_variant_extraction_array(&column, &row_type)
+ .unwrap();
+ let actual =
+ crate::arrow::shredding::variant::assemble_variant_extraction_array(read, &row_type)
+ .unwrap();
+ assert_eq!(actual.as_ref(), expected.as_ref());
+ }
+
#[tokio::test]
async fn shredded_variant_extraction_reads_metadata_as_dictionary() {
let fields = vec![DataField::new(