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 0c589fbf feat(parquet): project nested map value fields (#1246)
0c589fbf is described below
commit 0c589fbf2e3f34273234476c367f3368b29bf82c
Author: XiaoHongbo <[email protected]>
AuthorDate: Wed Oct 28 23:42:56 2026 +0800
feat(parquet): project nested map value fields (#1246)
---
crates/paimon/src/arrow/format/parquet.rs | 171 ++++++++++++++++++++++++++++
crates/paimon/src/table/data_file_reader.rs | 95 +++++++++++++++-
2 files changed, 261 insertions(+), 5 deletions(-)
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 69fdeafd..c678b529 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -1393,6 +1393,34 @@ fn read_type_projection_mask(
return;
}
let data_type = field.data_type();
+ if let DataType::Map(map) = data_type {
+ if !physical.is_primitive() && physical.get_fields().len() == 1 {
+ let entries = &physical.get_fields()[0];
+ if !entries.is_primitive()
+ && entries.get_fields().len() == 2
+ && entries.get_fields()[0].name() == "key"
+ && entries.get_fields()[1].name() == "value"
+ {
+ path.push(entries.name().to_string());
+ let key = &entries.get_fields()[0];
+ path.push(key.name().to_string());
+ super::variant_projection::all_leaves(key, path, output);
+ path.pop();
+ let value = &entries.get_fields()[1];
+ path.push(value.name().to_string());
+ requested_paths(
+ &DataField::new(field.id(), "value".into(),
map.value_type().clone()),
+ value,
+ path,
+ output,
+ null_residuals,
+ );
+ path.pop();
+ path.pop();
+ return;
+ }
+ }
+ }
if let DataType::Row(row) = data_type {
if crate::spec::is_variant_extraction_row_type(data_type) {
let projected =
super::variant_projection::projected_paths(row, physical);
@@ -3338,6 +3366,49 @@ mod tests {
use std::time::Duration;
use tokio::sync::mpsc;
+ #[test]
+ fn map_value_read_type_prunes_children_and_keeps_predicate_inputs() {
+ use crate::spec::RowType;
+ let descriptor = SchemaDescriptor::new(Arc::new(
+ parse_message_type(
+ "message schema { optional group attrs (MAP) {
+ repeated group key_value { required binary key (STRING);
+ optional group value { optional int32 wanted; optional
int64 unused; }
+ }
+ } }",
+ )
+ .unwrap(),
+ ));
+ let fields = vec![DataField::new(
+ 10,
+ "attrs".into(),
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Row(RowType::new(vec![DataField::new(
+ 11,
+ "wanted".into(),
+ DataType::Int(IntType::new()),
+ )])),
+ )),
+ )];
+ let mask =
+ super::read_type_projection_mask(&descriptor, &fields, 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("attrs").unwrap();
+ let widened = super::read_type_projection_mask(
+ &descriptor,
+ &fields,
+ None,
+ &[predicate],
+ &Default::default(),
+ );
+ for index in 0..3 {
+ assert!(widened.leaf_included(index));
+ }
+ }
+
#[test]
fn selected_map_rows_keep_all_map_leaves_without_a_shredding_plan() {
use crate::spec::RowType;
@@ -7610,6 +7681,106 @@ mod tests {
(path, file_io, file_size, fields)
}
+ #[tokio::test]
+ async fn
map_value_projection_preserves_nulls_empty_maps_and_duplicate_keys() {
+ use crate::spec::RowType;
+ let wanted = DataField::new(11, "wanted".into(),
DataType::Int(IntType::new()));
+ let unused = DataField::new(12, "unused".into(),
DataType::BigInt(BigIntType::new()));
+ let map_field = |children| {
+ DataField::new(
+ 10,
+ "attrs".into(),
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Row(RowType::new(children)),
+ )),
+ )
+ };
+ let fields = vec![map_field(vec![wanted.clone(), unused])];
+ let requested = vec![map_field(vec![wanted])];
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let ArrowDataType::Map(entries_field, ordered) =
schema.field(0).data_type() else {
+ panic!("expected map");
+ };
+ let ArrowDataType::Struct(entry_fields) = entries_field.data_type()
else {
+ panic!("expected entries");
+ };
+ let ArrowDataType::Struct(value_fields) = entry_fields[1].data_type()
else {
+ panic!("expected struct values");
+ };
+ let values = StructArray::new(
+ value_fields.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![Some(7), None, Some(9),
Some(10)])),
+ Arc::new(Int64Array::from(vec![70, 80, 90, 100])),
+ ],
+ Some(NullBuffer::from(vec![true, true, false, true])),
+ );
+ let entries = StructArray::new(
+ entry_fields.clone(),
+ vec![
+ Arc::new(StringArray::from(vec!["a", "a", "b", "c"])),
+ Arc::new(values),
+ ],
+ None,
+ );
+ let input = MapArray::new(
+ entries_field.clone(),
+ OffsetBuffer::new(ScalarBuffer::from(vec![0, 0, 0, 2, 3, 4])),
+ entries,
+ Some(NullBuffer::from(vec![false, true, true, true, true])),
+ *ordered,
+ );
+ let original: Arc<dyn Array> = Arc::new(input);
+ let expected =
+ crate::arrow::nested_evolution::evolve_field(&original,
&fields[0], &requested[0])
+ .unwrap();
+ let batch = RecordBatch::try_new(schema.clone(),
vec![original]).unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/map_value_projection.parquet";
+ let output = file_io.new_output(path).unwrap();
+ let mut writer =
+ create_format_writer(&output, schema, "zstd", 1, None,
Some(&fields), None)
+ .await
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ let size = writer.close().await.unwrap().file_size;
+ let batches = create_format_reader(path, false, &requested)
+ .unwrap()
+ .read_batch_stream(
+
Box::new(file_io.new_input(path).unwrap().reader().await.unwrap()),
+ size,
+ &requested,
+ None,
+ None,
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ let decoded = batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<MapArray>()
+ .unwrap();
+ let values = decoded
+ .values()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_eq!(values.column_names(), vec!["wanted"]);
+ let actual = crate::arrow::nested_evolution::evolve_field(
+ batches[0].column(0),
+ &fields[0],
+ &requested[0],
+ )
+ .unwrap();
+ assert_eq!(actual.to_data(), expected.to_data());
+ }
+
#[tokio::test]
async fn selected_map_rows_decode_complete_maps_before_core_assembly() {
use crate::spec::RowType;
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 68be79d4..9d548321 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -702,7 +702,12 @@ impl DataFileReader {
|| file_meta.write_cols.is_some()
|| read_type.iter().any(contains_selected_map)
{
- read_data_fields(&data_schema_fields, &read_type,
self.nested_field_enabled)?
+ read_data_fields_for_format(
+ &data_schema_fields,
+ &read_type,
+ self.nested_field_enabled,
+ file_meta.file_name.ends_with(".parquet"),
+ )?
} else {
read_type
.iter()
@@ -1030,7 +1035,12 @@ impl DataFileReader {
|| file_meta.write_cols.is_some()
|| read_type.iter().any(contains_selected_map)
{
- read_data_fields(&data_schema_fields, &read_type,
self.nested_field_enabled)?
+ read_data_fields_for_format(
+ &data_schema_fields,
+ &read_type,
+ self.nested_field_enabled,
+ file_meta.file_name.ends_with(".parquet"),
+ )?
} else {
read_type
.iter()
@@ -1317,10 +1327,25 @@ fn row_tracking_sequence_column(
))))
}
+#[cfg(test)]
fn read_data_fields(
all_data_fields: &[DataField],
expected_fields: &[DataField],
nested_field_enabled: bool,
+) -> crate::Result<Vec<DataField>> {
+ read_data_fields_for_format(
+ all_data_fields,
+ expected_fields,
+ nested_field_enabled,
+ false,
+ )
+}
+
+fn read_data_fields_for_format(
+ all_data_fields: &[DataField],
+ expected_fields: &[DataField],
+ nested_field_enabled: bool,
+ prune_map_values: bool,
) -> crate::Result<Vec<DataField>> {
let mut read_fields = Vec::new();
for data_field in all_data_fields {
@@ -1328,10 +1353,11 @@ fn read_data_fields(
.iter()
.find(|field| field.id() == data_field.id())
{
- if let Some(pruned_type) = prune_data_type(
+ if let Some(pruned_type) = prune_data_type_for_format(
expected.data_type(),
data_field.data_type(),
nested_field_enabled,
+ prune_map_values,
)? {
let mut field = data_field_with_type(data_field, pruned_type);
if crate::spec::is_map_selected_keys_field(expected) {
@@ -1367,12 +1393,40 @@ fn is_physical_system_field(field: &DataField) -> bool {
)
}
+#[cfg(test)]
fn prune_data_type(
read_type: &DataType,
data_type: &DataType,
nested_field_enabled: bool,
+) -> crate::Result<Option<DataType>> {
+ prune_data_type_for_format(read_type, data_type, nested_field_enabled,
false)
+}
+
+fn prune_data_type_for_format(
+ read_type: &DataType,
+ data_type: &DataType,
+ nested_field_enabled: bool,
+ prune_map_values: bool,
) -> crate::Result<Option<DataType>> {
match read_type {
+ DataType::Map(read_map) if prune_map_values => {
+ let DataType::Map(data_map) = data_type else {
+ return Ok(Some(data_type.clone()));
+ };
+ // Keep file field IDs and ordering; reconciliation handles
renames.
+ let value = prune_data_type_for_format(
+ read_map.value_type(),
+ data_map.value_type(),
+ true,
+ true,
+ )?
+ .unwrap_or_else(|| data_map.value_type().clone());
+ Ok(Some(DataType::Map(crate::spec::MapType::with_nullable(
+ data_type.is_nullable(),
+ data_map.key_type().clone(),
+ value,
+ ))))
+ }
DataType::Row(read_row) if is_variant_extraction_row_type(read_type)
=> {
Ok(Some(DataType::Row(read_row.clone())))
}
@@ -1380,8 +1434,12 @@ fn prune_data_type(
let DataType::Row(data_row) = data_type else {
return Ok(Some(data_type.clone()));
};
- let mut fields =
- read_data_fields(data_row.fields(), read_row.fields(),
nested_field_enabled)?;
+ let mut fields = read_data_fields_for_format(
+ data_row.fields(),
+ read_row.fields(),
+ nested_field_enabled,
+ prune_map_values,
+ )?;
if fields.is_empty() {
if !nested_field_enabled {
return Ok(None);
@@ -4610,4 +4668,31 @@ mod prune_container_tests {
assert_eq!(prune_data_type(&read, &data, false).unwrap().unwrap(),
data);
}
+
+ #[test]
+ fn parquet_map_value_pruning_keeps_file_names_ids_and_order() {
+ let data = DataType::Map(MapType::new(
+ str_t(),
+ row(vec![
+ f(5, "old_name", str_t()),
+ f(6, "unused", str_t()),
+ f(7, "last", str_t()),
+ ]),
+ ));
+ let read = DataType::Map(MapType::new(
+ str_t(),
+ row(vec![f(7, "last", str_t()), f(5, "renamed", str_t())]),
+ ));
+ let expected = DataType::Map(MapType::new(
+ str_t(),
+ row(vec![f(5, "old_name", str_t()), f(7, "last", str_t())]),
+ ));
+ assert_eq!(
+ prune_data_type_for_format(&read, &data, false, true)
+ .unwrap()
+ .unwrap(),
+ expected
+ );
+ assert_eq!(prune_data_type(&read, &data, false).unwrap().unwrap(),
data);
+ }
}
