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 283f1f93 Drive nested, MAP and VARIANT projections from structured
read types (#1041)
283f1f93 is described below
commit 283f1f938e77633ca33529e45e92f45fc6b9053e
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 5 22:03:41 2026 +0800
Drive nested, MAP and VARIANT projections from structured read types (#1041)
---
.../python/python/pypaimon_rust/datafusion.pyi | 10 +-
bindings/python/src/read.rs | 233 +--------------
bindings/python/tests/test_read.py | 5 +-
crates/paimon/src/arrow/format/mod.rs | 1 +
crates/paimon/src/arrow/format/parquet.rs | 328 ++++++++++++++++++---
.../paimon/src/arrow/format/variant_projection.rs | 297 +++++++++++++++++++
crates/paimon/src/arrow/nested_evolution.rs | 173 ++++++++++-
crates/paimon/src/arrow/shredding/variant.rs | 110 ++++---
crates/paimon/src/spec/map_selected_keys.rs | 120 ++++++++
crates/paimon/src/spec/mod.rs | 6 +
crates/paimon/src/spec/read_type.rs | 238 +++++++++++++++
crates/paimon/src/table/data_evolution_nested.rs | 114 ++++++-
crates/paimon/src/table/data_evolution_reader.rs | 50 +++-
crates/paimon/src/table/data_file_reader.rs | 231 ++++++++++-----
crates/paimon/src/table/kv_file_reader.rs | 50 ++--
crates/paimon/src/table/query_auth.rs | 84 ++++--
crates/paimon/src/table/read_builder.rs | 2 +
crates/paimon/src/variant.rs | 120 +++++++-
18 files changed, 1739 insertions(+), 433 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 31e4120f..2953e1c5 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -89,13 +89,13 @@ class TableRead:
class ReadBuilder:
def with_read_type(self, read_type_json: str) -> "ReadBuilder":
- """Set a complete Paimon ROW read type serialized as schema JSON."""
+ """Set the complete structured reader type as Paimon ROW schema JSON.
+ Preserve source field IDs for nested ROW pruning. Field descriptions
+ carry VARIANT extraction and selected-key MAP metadata. Results retain
+ the requested structure; output aliases and flattening belong to
callers.
+ """
...
def with_projection(self, columns: List[str]) -> "ReadBuilder": ...
- def with_nested_projection(self, paths: List[List[str]]) -> "ReadBuilder":
- """Project top-level fields or nested ROW leaves by exact name paths.
A MAP
- path keeps the complete MAP so the caller can extract literal keys."""
- ...
def with_case_sensitive(self, case_sensitive: bool) -> "ReadBuilder":
"""
Set whether column-name matching (projection and predicate column
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index 53e88384..8cbaf515 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -33,9 +33,6 @@ use tokio::sync::Notify;
use crate::error::to_py_err;
use crate::predicate::dict_to_table_predicate;
-const MAP_SELECTED_KEYS_PREFIX: &str = "__PAIMON_MAP_SELECTED_KEYS:";
-const MAP_SELECTED_KEYS_DELIMITER: char = ';';
-
struct PyReadProjection<'a> {
columns: &'a Option<Vec<String>>,
read_type: &'a Option<Vec<DataField>>,
@@ -99,106 +96,6 @@ fn apply_read_config(
Ok(())
}
-/// Resolve flat output paths to the authoritative nested read type used by the
-/// core reader. ROW children are pruned recursively. MAP values remain MAPs,
-/// but their requested string keys are carried in the temporary field
-/// description so shared-shredding Parquet files can prune physical columns.
-fn project_nested_read_type(
- fields: &[DataField],
- paths: &[Vec<String>],
-) -> PyResult<Vec<DataField>> {
- if paths.iter().any(Vec::is_empty) {
- return Err(PyValueError::new_err(
- "nested projection paths must not be empty",
- ));
- }
-
- let mut result = Vec::new();
- let mut seen = std::collections::HashSet::new();
- for path in paths {
- let name = &path[0];
- if !seen.insert(name.clone()) {
- continue;
- }
- let field = fields
- .iter()
- .find(|field| field.name() == name)
- .ok_or_else(|| {
- PyValueError::new_err(format!(
- "nested projection field '{}' does not exist",
- path.join(".")
- ))
- })?;
- let matching: Vec<&[String]> = paths
- .iter()
- .filter(|candidate| candidate.first() == Some(name))
- .map(|candidate| candidate[1..].as_ref())
- .collect();
- result.push(project_nested_field(field, &matching, name)?);
- }
- Ok(result)
-}
-
-fn project_nested_field(
- field: &DataField,
- tails: &[&[String]],
- full_name: &str,
-) -> PyResult<DataField> {
- if tails.iter().any(|tail| tail.is_empty()) {
- return Ok(field.clone());
- }
- match field.data_type() {
- DataType::Row(row) => {
- let child_paths: Vec<Vec<String>> = tails.iter().map(|tail|
tail.to_vec()).collect();
- let children = project_nested_read_type(row.fields(),
&child_paths).map_err(|_| {
- PyValueError::new_err(format!(
- "nested projection field '{}' does not exist",
- tails
- .first()
- .map(|tail| format!("{full_name}.{}", tail.join(".")))
- .unwrap_or_else(|| full_name.to_string())
- ))
- })?;
- Ok(DataField::new(
- field.id(),
- field.name().to_string(),
- DataType::Row(RowType::with_nullable(
- field.data_type().is_nullable(),
- children,
- )),
- )
- .with_description(field.description().map(str::to_string)))
- }
- DataType::Map(_) if tails.iter().all(|tail| tail.len() == 1) => {
- let mut keys = Vec::new();
- for tail in tails {
- let key = &tail[0];
- if key.contains(MAP_SELECTED_KEYS_DELIMITER)
- || key.starts_with(MAP_SELECTED_KEYS_PREFIX)
- {
- // Keep the complete MAP for keys which cannot be encoded
- // by the cross-language selected-key convention.
- return Ok(field.clone());
- }
- if !keys.contains(key) {
- keys.push(key.clone());
- }
- }
- Ok(field.clone().with_description(Some(format!(
- "{MAP_SELECTED_KEYS_PREFIX}{}",
- keys.join(&MAP_SELECTED_KEYS_DELIMITER.to_string())
- ))))
- }
- _ => Err(PyValueError::new_err(format!(
- "nested projection field '{}' is not a ROW or MAP",
- tails
- .first()
- .map(|tail| format!("{full_name}.{}", tail.join(".")))
- .unwrap_or_else(|| full_name.to_string())
- ))),
- }
-}
-
/// Extract a sequence of Python `Split` objects into core `DataSplit`s.
Accepts
/// any iterable (list/tuple/generator). Runs under the GIL since it touches
/// Python objects. A non-iterable argument or a non-`Split` element raises
@@ -338,39 +235,7 @@ impl PyReadBuilder {
Ok(slf)
}
- /// Project top-level fields or nested ROW leaves by their exact name
paths.
- /// MAP paths keep the complete MAP so the caller can extract literal keys.
- fn with_nested_projection(
- mut slf: PyRefMut<'_, Self>,
- paths: Vec<Vec<String>>,
- ) -> PyResult<PyRefMut<'_, Self>> {
- let schema = slf.table.schema();
- let mut read_type = project_nested_read_type(schema.fields(), &paths)?;
- let options = schema.options();
- let has_default_aggregation =
options.contains_key("fields.default-aggregate-function");
- for projected in &mut read_type {
- let selected_map = matches!(projected.data_type(),
DataType::Map(_))
- && projected
- .description()
- .is_some_and(|description|
description.starts_with(MAP_SELECTED_KEYS_PREFIX));
- let has_field_aggregation =
- options.contains_key(&format!("fields.{}.aggregate-function",
projected.name()));
- if selected_map && (has_default_aggregation ||
has_field_aggregation) {
- // Merge engines may need the complete MAP to calculate the
- // projected key. Match PyPaimon's complete-MAP fallback.
- *projected = schema
- .fields()
- .iter()
- .find(|field| field.name() == projected.name())
- .expect("nested projection came from this schema")
- .clone();
- }
- }
- slf.read_type = Some(read_type);
- slf.projection = None;
- Ok(slf)
- }
-
+ /// Limit the number of visible rows.
fn with_limit(mut slf: PyRefMut<'_, Self>, limit: usize) -> PyRefMut<'_,
Self> {
slf.limit = Some(limit);
slf
@@ -953,102 +818,6 @@ mod tests {
use super::*;
- use paimon::spec::{DataField, DataType, IntType, MapType, RowType,
VarCharType};
-
- #[test]
- fn nested_projection_prunes_rows_and_marks_selected_map_keys() {
- let fields = vec![
- DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
- DataField::new(
- 1,
- "payload".to_string(),
- DataType::Row(RowType::new(vec![
- DataField::new(2, "version".to_string(),
DataType::Int(IntType::new())),
- DataField::new(
- 3,
- "details".to_string(),
- DataType::Row(RowType::new(vec![
- DataField::new(
- 4,
- "name".to_string(),
- DataType::VarChar(VarCharType::string_type()),
- ),
- DataField::new(5, "score".to_string(),
DataType::Int(IntType::new())),
- ])),
- ),
- ])),
- ),
- DataField::new(
- 6,
- "attrs".to_string(),
- DataType::Map(MapType::new(
- DataType::VarChar(VarCharType::string_type()),
- DataType::Int(IntType::new()),
- )),
- ),
- ];
-
- let projected = project_nested_read_type(
- &fields,
- &[
- vec![
- "payload".to_string(),
- "details".to_string(),
- "score".to_string(),
- ],
- vec!["id".to_string()],
- vec!["attrs".to_string(), "selected".to_string()],
- ],
- )
- .unwrap();
-
- assert_eq!(
- projected.iter().map(DataField::name).collect::<Vec<_>>(),
- vec!["payload", "id", "attrs"]
- );
- let DataType::Row(payload) = projected[0].data_type() else {
- panic!("payload must remain a row");
- };
- assert_eq!(payload.fields().len(), 1);
- assert_eq!(payload.fields()[0].name(), "details");
- let DataType::Row(details) = payload.fields()[0].data_type() else {
- panic!("details must remain a row");
- };
- assert_eq!(
- details
- .fields()
- .iter()
- .map(DataField::name)
- .collect::<Vec<_>>(),
- vec!["score"]
- );
- assert_eq!(projected[2].data_type(), fields[2].data_type());
- assert_eq!(
- projected[2].description(),
- Some("__PAIMON_MAP_SELECTED_KEYS:selected")
- );
- }
-
- #[test]
- fn nested_projection_rejects_an_invalid_struct_path() {
- let fields = vec![DataField::new(
- 1,
- "payload".to_string(),
- DataType::Row(RowType::new(vec![DataField::new(
- 2,
- "version".to_string(),
- DataType::Int(IntType::new()),
- )])),
- )];
-
- let error = project_nested_read_type(
- &fields,
- &[vec!["payload".to_string(), "missing".to_string()]],
- )
- .unwrap_err();
- assert!(error.to_string().contains("payload.missing"));
- }
-
#[test]
fn record_batch_reader_close_interrupts_pending_next() {
let (polled_tx, polled_rx) = mpsc::channel();
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index 6d72e482..27642786 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -213,11 +213,12 @@ def
test_with_read_type_rejects_unsupported_variant_target():
table.new_read_builder().with_read_type(json.dumps(read_type))
-def test_with_nested_projection_replaces_read_type():
+def test_with_read_type_replaces_variant_extraction():
with tempfile.TemporaryDirectory() as warehouse:
table = _make_variant_table_with_data(warehouse)
builder = table.new_read_builder().with_read_type(_variant_read_type(
- table, ["payload"], ["$.age"])).with_nested_projection([["id"]])
+ table, ["payload"], ["$.age"])).with_read_type(json.dumps({"type":
"ROW", "fields": [{
+ "id": table.schema().fields()[0].id(), "name": "id", "type":
"INT"}]}))
plan = builder.new_scan().plan()
result = pa.Table.from_batches(builder.new_read().read(plan.splits()))
assert result.schema.names == ["id"]
diff --git a/crates/paimon/src/arrow/format/mod.rs
b/crates/paimon/src/arrow/format/mod.rs
index b85fad7e..a400b067 100644
--- a/crates/paimon/src/arrow/format/mod.rs
+++ b/crates/paimon/src/arrow/format/mod.rs
@@ -26,6 +26,7 @@ pub(crate) mod parquet;
mod row;
mod shredding;
pub(crate) mod text;
+mod variant_projection;
#[cfg(feature = "vortex")]
mod vortex;
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 213f93af..81b9f15b 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -686,28 +686,18 @@ impl FormatFileReader for ParquetFormatReader {
crate::arrow::residual::widen_scan_fields(read_fields, predicates)
};
- let root_schema = parquet_schema.root_schema();
- let root_indices: Vec<usize> = scan_fields
- .iter()
- .filter_map(|f| {
- root_schema
- .get_fields()
- .iter()
- .position(|pf| pf.name() == f.name())
- })
- .collect();
-
// Build the MAP plan before the projection mask: a selected-key MAP
// can then decode only __field_mapping plus the direct/overflow
// children which may contain the requested keys.
let map_read_plan =
MapShreddingReadPlan::create(&scan_fields,
batch_stream_builder.schema())?
.map(Arc::new);
- let mask = if let Some(plan) = map_read_plan.as_deref() {
- map_shredding_projection_mask(&parquet_schema, &root_indices, plan)
- } else {
- ProjectionMask::roots(&parquet_schema, root_indices)
- };
+ let mask = read_type_projection_mask(
+ &parquet_schema,
+ &scan_fields,
+ map_read_plan.as_deref(),
+ preds,
+ );
batch_stream_builder =
batch_stream_builder.with_projection(mask.clone());
let mut decoder_predicates = build_parquet_row_filter(&parquet_schema,
preds, file_fields)?
@@ -1060,29 +1050,90 @@ impl FormatFileReader for ParquetFormatReader {
}
}
-fn map_shredding_projection_mask(
+fn read_type_projection_mask(
parquet_schema: &parquet::schema::types::SchemaDescriptor,
- selected_roots: &[usize],
- plan: &MapShreddingReadPlan,
+ read_fields: &[DataField],
+ map_plan: Option<&MapShreddingReadPlan>,
+ predicates: &[Predicate],
) -> ProjectionMask {
- let selected_roots: std::collections::HashSet<usize> =
selected_roots.iter().copied().collect();
- let root_fields = parquet_schema.root_schema().get_fields();
- let leaves = (0..parquet_schema.num_columns()).filter(|leaf_index| {
- let root_index = parquet_schema.get_column_root_idx(*leaf_index);
- if !selected_roots.contains(&root_index) {
- return false;
+ fn requested_paths(
+ field: &DataField,
+ physical: &parquet::schema::types::Type,
+ path: &mut Vec<String>,
+ output: &mut std::collections::HashSet<Vec<String>>,
+ ) {
+ if crate::spec::is_map_selected_keys_field(field) {
+ // Selected-key ROW children name logical MAP keys, not physical
+ // Parquet children. Shared MAPs use their read plan above; other
+ // MAPs must remain complete for the core's key-to-ROW assembly.
+ super::variant_projection::all_leaves(physical, path, output);
+ return;
}
- let root_name = root_fields[root_index].name();
- let Some(children) = plan.projected_physical_children(root_name) else {
- return true;
+ 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
+ }),
+ );
+ return;
+ }
+ if !physical.is_primitive() {
+ for field in row.fields() {
+ if let Some(child) = physical
+ .get_fields()
+ .iter()
+ .find(|child| child.name() == field.name())
+ {
+ path.push(child.name().to_string());
+ requested_paths(field, child, path, output);
+ path.pop();
+ }
+ }
+ return;
+ }
+ }
+ super::variant_projection::all_leaves(physical, path, output);
+ }
+ let mut predicate_refs = Vec::new();
+ for predicate in predicates {
+ crate::arrow::residual::collect_predicate_leaf_refs(predicate, &mut
predicate_refs);
+ }
+ let roots = parquet_schema.root_schema().get_fields();
+ let mut selected = std::collections::HashSet::new();
+ for field in read_fields {
+ let Some(root) = roots.iter().find(|root| root.name() == field.name())
else {
+ continue;
};
- parquet_schema
- .column(*leaf_index)
- .path()
- .parts()
- .get(1)
- .is_some_and(|child| children.contains(child))
- });
+ let mut path = vec![root.name().to_string()];
+ if predicate_refs
+ .iter()
+ .any(|(column, _)| *column == field.name())
+ {
+ super::variant_projection::all_leaves(root, &mut path, &mut
selected);
+ } else if let Some(children) =
+ map_plan.and_then(|plan|
plan.projected_physical_children(field.name()))
+ {
+ for child in root
+ .get_fields()
+ .iter()
+ .filter(|child| children.contains(child.name()))
+ {
+ path.push(child.name().to_string());
+ super::variant_projection::all_leaves(child, &mut path, &mut
selected);
+ path.pop();
+ }
+ } else {
+ requested_paths(field, root, &mut path, &mut selected);
+ }
+ }
+ let leaves = (0..parquet_schema.num_columns())
+ .filter(|&leaf|
selected.contains(parquet_schema.column(leaf).path().parts()));
ProjectionMask::leaves(parquet_schema, leaves)
}
@@ -2935,6 +2986,92 @@ mod tests {
use std::time::Duration;
use tokio::sync::mpsc;
+ #[test]
+ fn selected_map_rows_keep_all_map_leaves_without_a_shredding_plan() {
+ 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 int32 value;
+ } }
+ optional group profile {
+ optional group attrs (MAP) { repeated group key_value {
+ required binary key (STRING); optional int32 value;
+ } }
+ optional int32 unused;
+ }
+ optional int32 unused;
+ }",
+ )
+ .unwrap(),
+ ));
+ let map_type = DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Int(IntType::new()),
+ ));
+ let top = crate::spec::map_selected_keys_field(
+ &DataField::new(10, "attrs".into(), map_type.clone()),
+ &["wanted".into()],
+ )
+ .unwrap();
+ // A literal key can even collide with a Parquet layout field name.
+ let nested = crate::spec::map_selected_keys_field(
+ &DataField::new(21, "attrs".into(), map_type),
+ &["key_value".into()],
+ )
+ .unwrap();
+ let fields = vec![
+ top,
+ DataField::new(
+ 20,
+ "profile".into(),
+ DataType::Row(RowType::new(vec![nested])),
+ ),
+ ];
+ let mask = super::read_type_projection_mask(&descriptor, &fields,
None, &[]);
+ for index in 0..4 {
+ assert!(mask.leaf_included(index), "Missing MAP leaf {index}");
+ }
+ assert!(!mask.leaf_included(4));
+ assert!(!mask.leaf_included(5));
+ }
+
+ #[test]
+ fn nested_read_type_masks_physical_leaves_and_keeps_predicate_inputs() {
+ use crate::spec::RowType;
+ let descriptor = SchemaDescriptor::new(Arc::new(parse_message_type(
+ "message schema { optional group profile { optional int32 score;
optional binary unused; } optional int32 id; }"
+ ).unwrap()));
+ let fields = vec![
+ DataField::new(
+ 10,
+ "profile".into(),
+ DataType::Row(RowType::new(vec![
+ DataField::new(11, "score".into(),
DataType::Int(IntType::new())),
+ DataField::new(
+ 12,
+ "unused".into(),
+ DataType::VarChar(VarCharType::string_type()),
+ ),
+ ])),
+ ),
+ DataField::new(20, "id".into(), DataType::Int(IntType::new())),
+ ];
+ let selected =
+ crate::spec::project_read_type(&fields, &[vec!["profile".into(),
"score".into()]])
+ .unwrap();
+ let mask = super::read_type_projection_mask(&descriptor, &selected,
None, &[]);
+ 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]);
+ assert!(widened.leaf_included(0));
+ assert!(widened.leaf_included(1));
+ assert!(!widened.leaf_included(2));
+ }
+
fn test_fields() -> Vec<DataField> {
vec![
DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
@@ -6537,6 +6674,127 @@ mod tests {
(path, file_io, file_size, fields)
}
+ #[tokio::test]
+ async fn selected_map_rows_decode_complete_maps_before_core_assembly() {
+ use crate::spec::RowType;
+ let map_type = map_shredding_fields()[1].data_type().clone();
+ let fields = vec![
+ DataField::new(10, "attrs".into(), map_type.clone()),
+ DataField::new(
+ 20,
+ "profile".into(),
+ DataType::Row(RowType::new(vec![
+ DataField::new(21, "attrs".into(), map_type),
+ DataField::new(22, "unused".into(),
DataType::Int(IntType::new())),
+ ])),
+ ),
+ ];
+ let rows = vec![
+ Some(vec![("wanted", Some(10)), ("key_value", Some(11))]),
+ None,
+ Some(vec![]),
+ Some(vec![
+ ("wanted", Some(40)),
+ ("wanted", Some(50)),
+ ("key_value", Some(41)),
+ ]),
+ ];
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let ArrowDataType::Struct(profile_fields) =
schema.field(1).data_type() else {
+ panic!("profile must be ROW");
+ };
+ let profile = StructArray::new(
+ profile_fields.clone(),
+ vec![
+ Arc::new(build_int64_map_array(&rows)),
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4])),
+ ],
+ Some(NullBuffer::from(vec![true, false, true, true])),
+ );
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![Arc::new(build_int64_map_array(&rows)), Arc::new(profile)],
+ )
+ .unwrap();
+ let requested = crate::spec::project_read_type(
+ &fields,
+ &[
+ vec!["attrs".into(), "wanted".into()],
+ vec!["attrs".into(), "missing".into()],
+ vec!["profile".into(), "attrs".into(), "key_value".into()],
+ ],
+ )
+ .unwrap();
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/selected_map_complete_fallback.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 file_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()),
+ file_size,
+ &requested,
+ None,
+ None,
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(batches.len(), 1);
+ assert_int64_map_rows(
+ batches[0]
+ .column(0)
+ .as_any()
+ .downcast_ref::<MapArray>()
+ .unwrap(),
+ &rows,
+ );
+ let top = crate::arrow::nested_evolution::evolve_field(
+ batches[0].column(0),
+ &fields[0],
+ &requested[0],
+ )
+ .unwrap();
+ let top = top.as_any().downcast_ref::<StructArray>().unwrap();
+ assert!(top.is_null(1));
+ assert_eq!(
+ top.column(0).as_any().downcast_ref::<Int64Array>().unwrap(),
+ &Int64Array::from(vec![Some(10), None, None, Some(40)]),
+ );
+ assert_eq!(top.column(1).null_count(), 4);
+ let profile = crate::arrow::nested_evolution::evolve_field(
+ batches[0].column(1),
+ &fields[1],
+ &requested[1],
+ )
+ .unwrap();
+ let profile = profile.as_any().downcast_ref::<StructArray>().unwrap();
+ assert!(profile.is_null(1));
+ assert_eq!(profile.num_columns(), 1);
+ let attrs = profile
+ .column(0)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert_eq!(
+ attrs
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap(),
+ &Int64Array::from(vec![Some(11), None, None, Some(41)]),
+ );
+ }
+
#[tokio::test]
async fn test_parquet_map_shredding_write_read_roundtrip() {
let rows: Vec<Option<Vec<(&str, Option<i64>)>>> = vec![
diff --git a/crates/paimon/src/arrow/format/variant_projection.rs
b/crates/paimon/src/arrow/format/variant_projection.rs
new file mode 100644
index 00000000..1af09d2a
--- /dev/null
+++ b/crates/paimon/src/arrow/format/variant_projection.rs
@@ -0,0 +1,297 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Read-type-driven Parquet clipping, mirroring VariantShreddingTypePruner.
+
+use crate::spec::{parse_variant_metadata, RowType};
+use crate::variant::{parse_path, PathSegment};
+use parquet::basic::{LogicalType, Repetition};
+use parquet::schema::types::Type;
+use std::collections::{HashMap, HashSet};
+
+#[derive(Default)]
+struct PathNode {
+ children: HashMap<String, PathNode>,
+ element: Option<Box<PathNode>>,
+ keep_all: bool,
+}
+
+/// The requested physical leaves, relative to the Variant column. Invalid
+/// paths retain the full input so strict/try extraction handles the error at
+/// the same point as an unshredded read.
+pub(super) fn projected_paths(row: &RowType, physical: &Type) ->
HashSet<Vec<String>> {
+ let mut root = PathNode::default();
+ for field in row.fields() {
+ let segments = field.description().and_then(|description| {
+ parse_variant_metadata(description)
+ .ok()
+ .and_then(|metadata| parse_path(metadata.path()).ok())
+ });
+ let Some(segments) = segments else {
+ root.keep_all = true;
+ break;
+ };
+ let mut node = &mut root;
+ for segment in segments {
+ node = match segment {
+ PathSegment::Key(key) => node.children.entry(key).or_default(),
+ PathSegment::Index(_) =>
node.element.get_or_insert_with(Default::default),
+ };
+ }
+ node.keep_all = true;
+ }
+ let mut result = HashSet::new();
+ clip(physical, &root, &mut Vec::new(), &mut result);
+ result
+}
+
+pub(super) fn all_leaves(
+ physical: &Type,
+ path: &mut Vec<String>,
+ output: &mut HashSet<Vec<String>>,
+) {
+ if physical.is_primitive() {
+ output.insert(path.clone());
+ } else {
+ for child in physical.get_fields() {
+ path.push(child.name().to_string());
+ all_leaves(child, path, output);
+ path.pop();
+ }
+ }
+}
+
+fn keep_child(child: &Type, path: &mut Vec<String>, output: &mut
HashSet<Vec<String>>) {
+ path.push(child.name().to_string());
+ all_leaves(child, path, output);
+ path.pop();
+}
+
+/// Mirror Java's ParquetListLayoutResolver.isCanonicalList. Other LIST
+/// layouts must stay whole instead of assuming these physical levels.
+fn canonical_list(physical: &Type) -> Option<(&Type, &Type)> {
+ if physical.is_primitive()
+ || !matches!(
+ physical.get_basic_info().logical_type_ref(),
+ Some(LogicalType::List)
+ )
+ || physical.get_fields().len() != 1
+ {
+ return None;
+ }
+ let repeated = &physical.get_fields()[0];
+ if repeated.is_primitive()
+ || repeated.name() != "list"
+ || repeated.get_basic_info().repetition() != Repetition::REPEATED
+ || repeated.get_fields().len() != 1
+ {
+ return None;
+ }
+ let element = &repeated.get_fields()[0];
+ (element.name() == "element" && element.get_basic_info().repetition() !=
Repetition::REPEATED)
+ .then_some((repeated.as_ref(), element.as_ref()))
+}
+
+fn clip(
+ physical: &Type,
+ node: &PathNode,
+ path: &mut Vec<String>,
+ output: &mut HashSet<Vec<String>>,
+) {
+ if physical.is_primitive() || node.keep_all {
+ all_leaves(physical, path, output);
+ return;
+ }
+ let fields = physical.get_fields();
+ let Some(typed) = fields.iter().find(|child| child.name() ==
"typed_value") else {
+ all_leaves(physical, path, output);
+ return;
+ };
+ let logical = typed.get_basic_info().logical_type_ref();
+ let object = !typed.is_primitive() && logical.is_none();
+ let list_layout = canonical_list(typed);
+ let list = list_layout.is_some() && node.element.is_some();
+ if (!object || node.element.is_some()) && !list {
+ all_leaves(physical, path, output);
+ return;
+ }
+ if list && node.element.as_ref().unwrap().keep_all {
+ all_leaves(physical, path, output);
+ return;
+ }
+ if let Some(metadata) = fields.iter().find(|child| child.name() ==
"metadata") {
+ keep_child(metadata, path, output);
+ }
+ if object && node.element.is_none() {
+ let missing = node
+ .children
+ .keys()
+ .any(|key| !typed.get_fields().iter().any(|field| field.name() ==
key));
+ if missing {
+ if let Some(value) = fields.iter().find(|child| child.name() ==
"value") {
+ keep_child(value, path, output);
+ }
+ }
+ path.push("typed_value".into());
+ for child in typed.get_fields() {
+ if let Some(selected) = node.children.get(child.name()) {
+ path.push(child.name().to_string());
+ clip(child, selected, path, output);
+ path.pop();
+ }
+ }
+ path.pop();
+ } else {
+ // Canonical Parquet LIST: typed_value / list / element. Array
+ // indices cannot prune rows, only fields within every element.
+ let (repeated, element) = list_layout.expect("canonical layout was
checked above");
+ if !node.children.is_empty() {
+ if let Some(value) = fields.iter().find(|child| child.name() ==
"value") {
+ keep_child(value, path, output);
+ }
+ }
+ path.extend([
+ typed.name().into(),
+ repeated.name().into(),
+ element.name().into(),
+ ]);
+ clip(element, node.element.as_ref().unwrap(), path, output);
+ path.truncate(path.len() - 3);
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{variant_extraction_row, DataType, FloatType};
+ use parquet::schema::parser::parse_message_type;
+
+ fn paths(requests: &[&str]) -> HashSet<Vec<String>> {
+ let physical = parse_message_type(r#"message schema { optional group v
{
+ optional binary metadata; optional binary value;
+ optional group typed_value {
+ optional group obj { optional binary value; optional group
typed_value {
+ optional group wanted { optional binary value; optional
float typed_value; }
+ optional group unused { optional binary value; optional
float typed_value; }
+ } }
+ optional group arr { optional binary value; optional group
typed_value (LIST) {
+ repeated group list { optional group element {
+ optional binary value; optional group typed_value {
+ optional group wanted { optional binary value;
optional float typed_value; }
+ optional group unused { optional binary value;
optional float typed_value; }
+ }
+ } }
+ } }
+ }
+ } }"#).unwrap();
+ let row = variant_extraction_row(
+ true,
+ requests.iter().map(|path| {
+ (
+ DataType::Float(FloatType::new()),
+ (*path).into(),
+ false,
+ "UTC".into(),
+ )
+ }),
+ )
+ .unwrap();
+ projected_paths(&row, &physical.get_fields()[0])
+ }
+
+ fn includes(paths: &HashSet<Vec<String>>, path: &str) -> bool {
+
paths.contains(&path.split('.').map(str::to_string).collect::<Vec<_>>())
+ }
+
+ #[test]
+ fn object_array_and_missing_paths_prune_unrequested_leaves() {
+ let selected = paths(&["$.obj.wanted", "$.arr[1].wanted",
"$.missing"]);
+ assert!(includes(&selected, "metadata"));
+ assert!(includes(&selected, "value"));
+ assert!(includes(
+ &selected,
+ "typed_value.obj.typed_value.wanted.typed_value"
+ ));
+ assert!(includes(
+ &selected,
+
"typed_value.arr.typed_value.list.element.typed_value.wanted.typed_value"
+ ));
+ assert!(!selected
+ .iter()
+ .any(|parts| parts.iter().any(|part| part == "unused")));
+ let known = paths(&["$.obj.wanted"]);
+ assert!(!includes(&known, "value"));
+ let wrong_case = paths(&["$.Obj.wanted"]);
+ assert!(includes(&wrong_case, "value"));
+ assert_eq!(wrong_case.len(), 2);
+ }
+
+ #[test]
+ fn noncanonical_variant_lists_keep_the_whole_shredded_value() {
+ let row = variant_extraction_row(
+ true,
+ [(
+ DataType::Float(FloatType::new()),
+ "$[0].wanted".into(),
+ false,
+ "UTC".into(),
+ )],
+ )
+ .unwrap();
+ for (middle, middle_repetition, element, element_repetition) in [
+ ("array", "repeated", "element", "optional"),
+ ("typed_value_tuple", "repeated", "element", "optional"),
+ ("bag", "repeated", "element", "optional"),
+ ("list", "repeated", "item", "optional"),
+ ("list", "optional", "element", "optional"),
+ ("list", "repeated", "element", "repeated"),
+ ] {
+ let physical = parse_message_type(&format!(r#"message schema {{
optional group v {{
+ optional binary metadata; optional binary value;
+ optional group typed_value (LIST) {{
+ {middle_repetition} group {middle} {{
+ {element_repetition} group {element} {{
+ optional binary value; optional group typed_value
{{
+ optional group wanted {{ optional binary
value; optional float typed_value; }}
+ optional group unused {{ optional binary
value; optional float typed_value; }}
+ }}
+ }}
+ }}
+ }}
+ }} }}"#)).unwrap();
+ let variant = &physical.get_fields()[0];
+ let mut complete = HashSet::new();
+ all_leaves(variant, &mut Vec::new(), &mut complete);
+ assert_eq!(
+ projected_paths(&row, variant),
+ complete,
+ "Must keep {middle_repetition} {middle} / {element_repetition}
{element} whole",
+ );
+ }
+ }
+
+ #[test]
+ fn whole_values_and_invalid_paths_retain_the_required_shape() {
+ let whole = paths(&["$"]);
+ assert_eq!(paths(&["invalid"]), whole);
+ assert!(paths(&["$.arr[0]"])
+ .iter()
+ .any(|parts| parts.iter().any(|part| part == "unused")));
+ let mixed = paths(&["$.arr[0].wanted", "$.arr.key"]);
+ assert!(includes(&mixed, "typed_value.arr.value"));
+ }
+}
diff --git a/crates/paimon/src/arrow/nested_evolution.rs
b/crates/paimon/src/arrow/nested_evolution.rs
index df71d925..ffcc7bae 100644
--- a/crates/paimon/src/arrow/nested_evolution.rs
+++ b/crates/paimon/src/arrow/nested_evolution.rs
@@ -35,12 +35,108 @@ use std::sync::Arc;
use arrow_array::{
new_null_array, Array, ArrayRef, Int64Array, ListArray, MapArray,
StringArray, StructArray,
+ UInt32Array,
};
use arrow_cast::cast;
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Fields};
use crate::arrow::paimon_type_to_arrow;
-use crate::spec::{is_variant_extraction_row_type, DataType, MapType, RowType};
+use crate::spec::{
+ is_variant_extraction_row_type, map_selected_keys, DataField, DataType,
MapType, RowType,
+};
+
+/// Field-aware projection for synthetic read types and nested schema
evolution.
+pub(crate) fn evolve_field(
+ source: &ArrayRef,
+ source_field: &DataField,
+ target: &DataField,
+) -> crate::Result<ArrayRef> {
+ if let (DataType::Map(map_type), DataType::Row(row_type), Some(keys)) = (
+ source_field.data_type(),
+ target.data_type(),
+ map_selected_keys(target)?,
+ ) {
+ return selected_map_to_row(source, map_type, row_type, &keys);
+ }
+ if matches!(source_field.data_type(), DataType::Variant(_)) {
+ if let DataType::Row(row) = target.data_type() {
+ if is_variant_extraction_row_type(target.data_type()) {
+ // A format read plan may already have assembled the synthetic
+ // ROW while its field context still describes the VARIANT.
+ if source.data_type() ==
&paimon_type_to_arrow(target.data_type())? {
+ return Ok(source.clone());
+ }
+ return
crate::arrow::shredding::variant::assemble_variant_extraction_array(
+ source.as_ref(),
+ row,
+ );
+ }
+ }
+ }
+ evolve_column(source, source_field.data_type(), target.data_type())
+}
+
+pub(crate) fn selected_map_to_row(
+ source: &ArrayRef,
+ map_type: &MapType,
+ row_type: &RowType,
+ keys: &[String],
+) -> crate::Result<ArrayRef> {
+ let map =
+ source
+ .as_any()
+ .downcast_ref::<MapArray>()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "Selected MAP input must be a MapArray".into(),
+ source: None,
+ })?;
+ let names = map
+ .keys()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "Selected MAP keys must be STRING".into(),
+ source: None,
+ })?;
+ let mut children = Vec::new();
+ for (key, field) in keys.iter().zip(row_type.fields()) {
+ let indices: UInt32Array = (0..map.len())
+ .map(|row| {
+ if map.is_null(row) {
+ return None;
+ }
+ let offsets = map.value_offsets();
+ (offsets[row] as usize..offsets[row + 1] as usize)
+ .find(|&index| names.is_valid(index) && names.value(index)
== key)
+ .map(|index| index as u32)
+ })
+ .collect();
+ let values =
+ arrow_select::take::take(map.values().as_ref(), &indices,
None).map_err(|e| {
+ crate::Error::UnexpectedError {
+ message: "Failed to gather selected MAP values".into(),
+ source: Some(Box::new(e)),
+ }
+ })?;
+ children.push(evolve_column(
+ &values,
+ map_type.value_type(),
+ field.data_type(),
+ )?);
+ }
+ let ArrowDataType::Struct(fields) =
paimon_type_to_arrow(&DataType::Row(row_type.clone()))?
+ else {
+ unreachable!()
+ };
+ Ok(Arc::new(
+ StructArray::try_new(fields, children,
map.nulls().cloned()).map_err(|e| {
+ crate::Error::DataInvalid {
+ message: format!("Failed to build selected-key ROW: {e}"),
+ source: Some(Box::new(e)),
+ }
+ })?,
+ ))
+}
/// Reconcile `source` (as described by `source_type`, the type the data file
/// actually holds) with `target_type` (the type the read schema wants).
@@ -351,11 +447,7 @@ fn evolve_struct(
),
source: None,
})?;
- arrays.push(evolve_column(
- column,
- source_field.data_type(),
- target_field.data_type(),
- )?)
+ arrays.push(evolve_field(column, source_field, target_field)?)
}
}
}
@@ -550,6 +642,75 @@ mod tests {
use arrow_buffer::NullBuffer;
use arrow_schema::{DataType as ArrowDataType, Fields};
+ #[test]
+ fn selected_map_read_type_uses_key_metadata_not_child_names() {
+ use arrow_array::builder::{Int32Builder, MapBuilder, StringBuilder};
+ let mut builder = MapBuilder::new(None, StringBuilder::new(),
Int32Builder::new());
+ for (key, value) in [("key", 7), ("key", 9), ("", 3)] {
+ builder.keys().append_value(key);
+ builder.values().append_value(value);
+ }
+ builder.append(true).unwrap();
+ builder.append(false).unwrap();
+ builder.append(true).unwrap();
+ let source: ArrayRef = Arc::new(builder.finish());
+ let source_field = field(
+ 10,
+ "attrs",
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Int(IntType::new()),
+ )),
+ );
+ let requested = crate::spec::map_selected_keys_field(
+ &source_field,
+ &["key".into(), "".into(), "missing".into()],
+ )
+ .unwrap();
+ let DataType::Row(row) = requested.data_type() else {
+ panic!()
+ };
+ let target = DataField::new(
+ requested.id(),
+ requested.name().into(),
+ DataType::Row(RowType::new(
+ row.fields()
+ .iter()
+ .enumerate()
+ .map(|(index, child)| {
+ DataField::new(
+ child.id(),
+ format!("output_{index}"),
+ DataType::BigInt(BigIntType::new()),
+ )
+ })
+ .collect(),
+ )),
+ )
+ .with_description(requested.description().map(str::to_string));
+ let output = evolve_field(&source, &source_field, &target).unwrap();
+ let output = output.as_any().downcast_ref::<StructArray>().unwrap();
+ assert!(output.is_null(1));
+ assert!(output.is_valid(2));
+ let first = output
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ assert_eq!(first.iter().collect::<Vec<_>>(), vec![Some(7), None,
None]);
+ assert_eq!(
+ output
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>(),
+ vec![Some(3), None, None]
+ );
+ assert_eq!(output.column(2).null_count(), 3);
+ }
+
fn field(id: i32, name: &str, dt: DataType) -> DataField {
DataField::new(id, name.to_string(), dt)
}
diff --git a/crates/paimon/src/arrow/shredding/variant.rs
b/crates/paimon/src/arrow/shredding/variant.rs
index 9f75d406..83e90ca4 100644
--- a/crates/paimon/src/arrow/shredding/variant.rs
+++ b/crates/paimon/src/arrow/shredding/variant.rs
@@ -817,7 +817,6 @@ pub(crate) fn assemble_shredded_variant_batch(
let mut changed = false;
let mut columns = Vec::with_capacity(batch.num_columns());
let mut output_fields = Vec::with_capacity(batch.num_columns());
- let logical_schema = build_target_arrow_schema(read_fields)?;
for (idx, arrow_field) in schema.fields().iter().enumerate() {
let column = batch.column(idx);
@@ -829,8 +828,13 @@ pub(crate) fn assemble_shredded_variant_batch(
if let Some(assembled) =
assemble_array_to_logical(column.as_ref(),
read_fields[field_idx].data_type())?
{
+ output_fields.push(
+ arrow_field
+ .as_ref()
+ .clone()
+ .with_data_type(assembled.data_type().clone()),
+ );
columns.push(assembled);
- output_fields.push(logical_schema.field(field_idx).clone());
changed = true;
continue;
}
@@ -901,6 +905,15 @@ pub(crate) fn assemble_variant_extraction_array(
return Ok(projected);
}
+ let shredded_type = if is_shredded_variant_array(input) {
+ let DataType::Row(row) = arrow_to_paimon_type(input.data_type(),
true)? else {
+ unreachable!()
+ };
+ let schema = build_variant_schema(&row)?;
+ Some((row, schema))
+ } else {
+ None
+ };
let mut values_by_field = vec![Vec::with_capacity(input.len());
fields.len()];
let mut validities = Vec::with_capacity(input.len());
for row in 0..input.len() {
@@ -913,24 +926,40 @@ pub(crate) fn assemble_variant_extraction_array(
}
validities.push(true);
- let variant = variant_from_storage_row(input, row)?;
+ let shredded = shredded_type
+ .as_ref()
+ .map(|(row_type, _)| struct_row_at(input, row, row_type))
+ .transpose()?
+ .flatten();
+ let variant = if shredded_type.is_none() {
+ variant_from_storage_row(input, row)?
+ } else {
+ None
+ };
for (field_idx, field) in fields.iter().enumerate() {
let field_metadata = &metadata[field_idx];
- let value = match &variant {
- Some(variant) => match variant.get_path(field_metadata.path())
{
- Ok(Some(extracted)) => cast_variant_to_shredded_value(
- extracted,
- field.data_type(),
- field_metadata.fail_on_error(),
- )?,
- Ok(None) => None,
- Err(e) if !field_metadata.fail_on_error() => {
- let _ = e;
- None
- }
- Err(e) => return Err(e),
+ let extracted = match (&shredded, &shredded_type) {
+ (Some(shredded), Some((_, schema))) => {
+ crate::variant::extract_shredded_path(shredded, schema,
field_metadata.path())
+ }
+ _ => match &variant {
+ Some(variant) => variant
+ .get_path(field_metadata.path())
+ .and_then(|selected| {
+ selected.map(|value|
value.to_owned_variant()).transpose()
+ }),
+ None => Ok(None),
},
- None => 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);
}
@@ -1078,39 +1107,38 @@ fn assemble_row_array_to_logical(
message: "Variant assembled ROW column must be
StructArray".to_string(),
source: None,
})?;
- if input.num_columns() != row_type.fields().len() {
- return Ok(None);
- }
-
let mut changed = false;
let mut columns = Vec::with_capacity(input.num_columns());
- for (idx, field) in row_type.fields().iter().enumerate() {
- if let Some(assembled) =
- assemble_array_to_logical(input.column(idx).as_ref(),
field.data_type())?
- {
+ let mut fields = Vec::with_capacity(input.num_columns());
+ // Parquet leaf clipping preserves file order. Resolve each logical child
+ // by its physical name and retain that order for field-ID reconciliation.
+ for (physical, column) in input.fields().iter().zip(input.columns()) {
+ let logical = row_type
+ .fields()
+ .iter()
+ .find(|field| field.name() == physical.name());
+ let assembled = logical
+ .map(|field| assemble_array_to_logical(column.as_ref(),
field.data_type()))
+ .transpose()?
+ .flatten();
+ if let Some(assembled) = assembled {
+ fields.push(Arc::new(
+ physical
+ .as_ref()
+ .clone()
+ .with_data_type(assembled.data_type().clone()),
+ ));
columns.push(assembled);
changed = true;
} else {
- columns.push(input.column(idx).clone());
+ fields.push(Arc::clone(physical));
+ columns.push(Arc::clone(column));
}
}
-
if !changed {
return Ok(None);
}
-
- let fields: Fields = row_type
- .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 fields: Fields = fields.into();
if fields.is_empty() {
Ok(Some(Arc::new(StructArray::new_empty_fields(
input.len(),
@@ -1137,7 +1165,7 @@ fn is_variant_storage_array(array: &dyn Array) -> bool {
.iter()
.any(|field| field.name() == name && field.data_type() ==
&ArrowDataType::Binary)
};
- has_binary(VARIANT_VALUE_FIELD_NAME)
+ (has_binary(VARIANT_VALUE_FIELD_NAME) || is_shredded_variant_array(array))
&& has_binary(VARIANT_METADATA_FIELD_NAME)
&& (!is_variant_arrow_fields(fields) ||
is_shredded_variant_array(array))
}
diff --git a/crates/paimon/src/spec/map_selected_keys.rs
b/crates/paimon/src/spec/map_selected_keys.rs
new file mode 100644
index 00000000..3abf7dac
--- /dev/null
+++ b/crates/paimon/src/spec/map_selected_keys.rs
@@ -0,0 +1,120 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Java's temporary selected-key MAP ROW read type.
+
+use crate::spec::{DataField, DataType, RowType};
+use crate::{Error, Result};
+use std::collections::HashSet;
+
+/// Java field-description prefix for a temporary selected-key MAP ROW.
+pub const MAP_SELECTED_KEYS_METADATA_KEY: &str = "__PAIMON_MAP_SELECTED_KEYS:";
+
+/// Whether the field is a synthetic ROW carrying selected MAP key metadata.
+pub fn is_map_selected_keys_field(field: &DataField) -> bool {
+ matches!(field.data_type(), DataType::Row(_))
+ && field
+ .description()
+ .is_some_and(|value|
value.starts_with(MAP_SELECTED_KEYS_METADATA_KEY))
+}
+
+/// Decode keys in child ordinal order, validating arity and uniqueness.
+/// Ordinary fields return `None`.
+pub fn map_selected_keys(field: &DataField) -> Result<Option<Vec<String>>> {
+ if !is_map_selected_keys_field(field) {
+ return Ok(None);
+ }
+ let DataType::Row(row) = field.data_type() else {
+ unreachable!()
+ };
+ let encoded = field
+ .description()
+ .unwrap()
+ .strip_prefix(MAP_SELECTED_KEYS_METADATA_KEY)
+ .unwrap();
+ let keys: Vec<String> = encoded.split(';').map(str::to_string).collect();
+ if row.fields().is_empty()
+ || keys
+ .iter()
+ .any(|key| key.starts_with(MAP_SELECTED_KEYS_METADATA_KEY))
+ || keys.len() != row.fields().len()
+ || keys.iter().collect::<HashSet<_>>().len() != keys.len()
+ {
+ return Err(Error::DataInvalid {
+ message: format!(
+ "Selected-key metadata for '{}' must match the ROW fields and
contain unique keys",
+ field.name()
+ ),
+ source: None,
+ });
+ }
+ Ok(Some(keys))
+}
+
+/// Build Java's selected-key ROW for a STRING-keyed MAP, preserving the parent
+/// field ID and the complete value type. Each selected value is nullable.
+pub fn map_selected_keys_field(field: &DataField, keys: &[String]) ->
Result<DataField> {
+ let DataType::Map(map) = field.data_type() else {
+ return Err(Error::DataInvalid {
+ message: "Selected-key projection requires a MAP field".into(),
+ source: None,
+ });
+ };
+ if !matches!(map.key_type(), DataType::VarChar(_)) {
+ return Err(Error::Unsupported {
+ message: "Selected-key projection requires STRING map keys".into(),
+ });
+ }
+ let mut seen = HashSet::new();
+ if keys.is_empty()
+ || keys.iter().any(|key| {
+ key.contains(';')
+ || key.starts_with(MAP_SELECTED_KEYS_METADATA_KEY)
+ || !seen.insert(key)
+ })
+ {
+ return Err(Error::DataInvalid {
+ message: "Selected MAP keys must be nonempty, unique, and
encodable with Java metadata"
+ .into(),
+ source: None,
+ });
+ }
+ let children = keys
+ .iter()
+ .enumerate()
+ .map(|(index, key)| {
+ Ok(DataField::new(
+ index as i32,
+ key.clone(),
+ map.value_type().copy_with_nullable(true)?,
+ ))
+ })
+ .collect::<Result<Vec<_>>>()?;
+ Ok(DataField::new(
+ field.id(),
+ field.name().into(),
+ DataType::Row(RowType::with_nullable(
+ field.data_type().is_nullable(),
+ children,
+ )),
+ )
+ .with_description(Some(format!(
+ "{MAP_SELECTED_KEYS_METADATA_KEY}{}",
+ keys.join(";")
+ )))
+ .with_default_value(field.default_value().map(str::to_string)))
+}
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index e7b2a9af..9c64f6fd 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -57,6 +57,12 @@ pub use schema::*;
mod schema_change;
pub use schema_change::*;
+mod read_type;
+pub use read_type::*;
+
+mod map_selected_keys;
+pub use map_selected_keys::*;
+
mod variant_metadata;
pub use variant_metadata::*;
diff --git a/crates/paimon/src/spec/read_type.rs
b/crates/paimon/src/spec/read_type.rs
new file mode 100644
index 00000000..17d3c69e
--- /dev/null
+++ b/crates/paimon/src/spec/read_type.rs
@@ -0,0 +1,238 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Resolve nested reader requests while preserving ROW structure and field
IDs.
+
+use crate::spec::{
+ map_selected_keys_field, DataField, DataType, RowType,
MAP_SELECTED_KEYS_METADATA_KEY,
+};
+use crate::{Error, Result};
+const MAP_KEYS_DELIMITER: char = ';';
+fn invalid(message: impl Into<String>) -> Error {
+ Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ }
+}
+
+/// Resolve flat output paths to the authoritative nested read type used by the
+/// core reader. ROW children are pruned recursively. Selected MAP keys use
+/// Java's temporary ROW field metadata so shared-shredding Parquet files can
+/// prune physical columns.
+pub fn project_read_type(fields: &[DataField], paths: &[Vec<String>]) ->
Result<Vec<DataField>> {
+ if paths.iter().any(Vec::is_empty) {
+ return Err(invalid("nested projection paths must not be empty"));
+ }
+
+ let mut result = Vec::new();
+ let mut seen = std::collections::HashSet::new();
+ for path in paths {
+ let name = &path[0];
+ if !seen.insert(name.clone()) {
+ continue;
+ }
+ let field = fields
+ .iter()
+ .find(|field| field.name() == name)
+ .ok_or_else(|| {
+ invalid(format!(
+ "nested projection field '{}' does not exist",
+ path.join(".")
+ ))
+ })?;
+ let matching: Vec<&[String]> = paths
+ .iter()
+ .filter(|candidate| candidate.first() == Some(name))
+ .map(|candidate| candidate[1..].as_ref())
+ .collect();
+ result.push(project_nested_field(field, &matching, name)?);
+ }
+ Ok(result)
+}
+
+fn project_nested_field(
+ field: &DataField,
+ tails: &[&[String]],
+ full_name: &str,
+) -> Result<DataField> {
+ if tails.iter().any(|tail| tail.is_empty()) {
+ return Ok(field.clone());
+ }
+ match field.data_type() {
+ DataType::Row(row) => {
+ let child_paths: Vec<Vec<String>> = tails.iter().map(|tail|
tail.to_vec()).collect();
+ let children = project_read_type(row.fields(),
&child_paths).map_err(|_| {
+ invalid(format!(
+ "nested projection field '{}' does not exist",
+ tails
+ .first()
+ .map(|tail| format!("{full_name}.{}", tail.join(".")))
+ .unwrap_or_else(|| full_name.to_string())
+ ))
+ })?;
+ Ok(DataField::new(
+ field.id(),
+ field.name().to_string(),
+ DataType::Row(RowType::with_nullable(
+ field.data_type().is_nullable(),
+ children,
+ )),
+ )
+ .with_description(field.description().map(str::to_string))
+ .with_default_value(field.default_value().map(str::to_string)))
+ }
+ DataType::Map(_) if tails.iter().all(|tail| tail.len() == 1) => {
+ let mut keys = Vec::new();
+ for tail in tails {
+ let key = &tail[0];
+ if key.contains(MAP_KEYS_DELIMITER)
+ || key.starts_with(MAP_SELECTED_KEYS_METADATA_KEY)
+ {
+ // Keep the complete MAP for keys which cannot be encoded
+ // by the cross-language selected-key convention.
+ return Ok(field.clone());
+ }
+ if !keys.contains(key) {
+ keys.push(key.clone());
+ }
+ }
+ map_selected_keys_field(field, &keys)
+ }
+ _ => Err(invalid(format!(
+ "nested projection field '{}' is not a ROW or MAP",
+ tails
+ .first()
+ .map(|tail| format!("{full_name}.{}", tail.join(".")))
+ .unwrap_or_else(|| full_name.to_string())
+ ))),
+ }
+}
+
+/// Validate synthetic MAP readers against the current logical schema, before
+/// historical physical value types are reconciled. Java selected children
+/// retain the MAP value type (apart from nullability), including all ROW IDs.
+pub(crate) fn validate_selected_map_fields(
+ source: &[DataField],
+ requested: &[DataField],
+) -> Result<()> {
+ for target in requested {
+ let Some(field) = source.iter().find(|field| field.id() ==
target.id()) else {
+ continue;
+ };
+ if crate::spec::is_map_selected_keys_field(target) {
+ let (DataType::Map(map), DataType::Row(row)) = (field.data_type(),
target.data_type())
+ else {
+ return Err(invalid("Selected-key ROW requires a MAP source
field"));
+ };
+ crate::spec::map_selected_keys(target)?;
+ if !matches!(map.key_type(), DataType::VarChar(_))
+ || row
+ .fields()
+ .iter()
+ .any(|child|
!map.value_type().equals_ignore_nullable(child.data_type()))
+ {
+ return Err(invalid(format!(
+ "Selected-key ROW '{}' must retain the STRING MAP's
complete value type",
+ target.name()
+ )));
+ }
+ } else if let (DataType::Row(source_row), DataType::Row(target_row)) =
+ (field.data_type(), target.data_type())
+ {
+ validate_selected_map_fields(source_row.fields(),
target_row.fields())?;
+ }
+ }
+ Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::spec::{IntType, MapType, VarCharType};
+
+ fn fields() -> Vec<DataField> {
+ vec![
+ DataField::new(
+ 10,
+ "profile".into(),
+ DataType::Row(RowType::new(vec![
+ DataField::new(11, "score".into(),
DataType::Int(IntType::new()))
+ .with_default_value(Some("7".into())),
+ DataField::new(
+ 12,
+ "attrs".into(),
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Int(IntType::new()),
+ )),
+ ),
+ ])),
+ )
+ .with_description(Some("profile description".into())),
+ DataField::new(20, "id".into(), DataType::Int(IntType::new())),
+ ]
+ }
+
+ fn path(parts: &[&str]) -> Vec<String> {
+ parts.iter().map(|part| (*part).into()).collect()
+ }
+
+ #[test]
+ fn structured_projection_preserves_ids_metadata_and_key_order() {
+ let read = project_read_type(
+ &fields(),
+ &[
+ path(&["profile", "attrs", ""]),
+ path(&["profile", "score"]),
+ path(&["profile", "attrs", "missing"]),
+ path(&["profile", "attrs", ""]),
+ path(&["id"]),
+ ],
+ )
+ .unwrap();
+ assert_eq!(read[0].id(), 10);
+ assert_eq!(read[0].description(), Some("profile description"));
+ let DataType::Row(row) = read[0].data_type() else {
+ panic!()
+ };
+ assert_eq!(row.fields()[0].id(), 12);
+ assert_eq!(
+ crate::spec::map_selected_keys(&row.fields()[0]).unwrap(),
+ Some(vec!["".into(), "missing".into()])
+ );
+ assert_eq!(row.fields()[1].id(), 11);
+ assert_eq!(row.fields()[1].default_value(), Some("7"));
+ assert_eq!(read[1].id(), 20);
+ }
+
+ #[test]
+ fn whole_fields_win_and_unencodable_keys_read_complete_map() {
+ let fields = fields();
+ assert_eq!(
+ project_read_type(&fields, &[path(&["profile", "score"]),
path(&["profile"])]).unwrap(),
+ fields[..1]
+ );
+ let read = project_read_type(&fields, &[path(&["profile", "attrs",
"a;b"])]).unwrap();
+ let DataType::Row(row) = read[0].data_type() else {
+ panic!()
+ };
+ assert!(matches!(row.fields()[0].data_type(), DataType::Map(_)));
+ assert!(project_read_type(&fields, &[Vec::new()]).is_err());
+ assert!(project_read_type(&fields, &[path(&["profile",
"unknown"])]).is_err());
+ assert!(project_read_type(&fields, &[]).unwrap().is_empty());
+ }
+}
diff --git a/crates/paimon/src/table/data_evolution_nested.rs
b/crates/paimon/src/table/data_evolution_nested.rs
index 6d65be42..c75203f6 100644
--- a/crates/paimon/src/table/data_evolution_nested.rs
+++ b/crates/paimon/src/table/data_evolution_nested.rs
@@ -154,7 +154,12 @@ pub(super) fn select_nested_row(
fn collect_leaf_ids(fields: &[DataField], output: &mut Vec<i32>) {
for field in fields {
match field.data_type() {
- DataType::Row(row) => collect_leaf_ids(row.fields(), output),
+ DataType::Row(row)
+ if
!crate::spec::is_variant_extraction_row_type(field.data_type())
+ && !crate::spec::is_map_selected_keys_field(field) =>
+ {
+ collect_leaf_ids(row.fields(), output)
+ }
_ => output.push(field.id()),
}
}
@@ -223,7 +228,14 @@ pub(super) fn source_read_field(
.fields()
.iter()
.filter(|child| selected_ids.contains(&child.id()))
- .cloned()
+ .map(|child| {
+ requested_row
+ .fields()
+ .iter()
+ .find(|requested| requested.id() == child.id())
+ .map(|requested| source_child_field(child, requested))
+ .unwrap_or_else(|| child.clone())
+ })
.collect::<Vec<_>>();
if children.is_empty() {
let physical = source_fields
@@ -267,6 +279,38 @@ pub(super) fn source_read_field(
))
}
+fn source_child_field(current: &DataField, requested: &DataField) -> DataField
{
+ if crate::spec::is_variant_extraction_row_type(requested.data_type())
+ || crate::spec::is_map_selected_keys_field(requested)
+ {
+ return requested.clone();
+ }
+ if let (DataType::Row(current_row), DataType::Row(requested_row)) =
+ (current.data_type(), requested.data_type())
+ {
+ let children = current_row
+ .fields()
+ .iter()
+ .map(|child| {
+ requested_row
+ .fields()
+ .iter()
+ .find(|requested| requested.id() == child.id())
+ .map(|requested| source_child_field(child, requested))
+ .unwrap_or_else(|| child.clone())
+ })
+ .collect();
+ return super::data_evolution_fields::field_with_type(
+ current,
+ DataType::Row(RowType::with_nullable(
+ current.data_type().is_nullable(),
+ children,
+ )),
+ );
+ }
+ current.clone()
+}
+
pub(super) fn assemble_nested_row(
plan: &NestedFieldPlan,
target_type: &ArrowDataType,
@@ -435,6 +479,72 @@ mod tests {
field(1, "profile", DataType::Row(RowType::new(children)))
}
+ #[test]
+ fn synthetic_children_bind_to_the_original_leaf_id_and_keep_metadata() {
+ use crate::spec::{variant_extraction_row, FloatType, MapType,
VariantType};
+ let variant = field(30, "payload",
DataType::Variant(VariantType::new()));
+ let attrs = field(
+ 40,
+ "attrs",
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Int(IntType::new()),
+ )),
+ );
+ let extraction = field(
+ 30,
+ "payload",
+ DataType::Row(
+ variant_extraction_row(
+ true,
+ [(
+ DataType::Float(FloatType::new()),
+ "$.ratio".into(),
+ true,
+ "UTC".into(),
+ )],
+ )
+ .unwrap(),
+ ),
+ );
+ let selected = crate::spec::map_selected_keys_field(&attrs,
&["wanted".into()]).unwrap();
+ let current = row(vec![variant.clone(), attrs.clone(), age()]);
+ let requested = row(vec![extraction.clone(), selected.clone()]);
+ let whole = vec![current.clone()];
+ let sources = [PhysicalRowSource {
+ source_index: 0,
+ fields: &whole,
+ }];
+ assert_eq!(
+ select_nested_row(&requested, &sources)
+ .unwrap()
+ .unwrap()
+ .whole_source,
+ Some(0)
+ );
+
+ let newer = vec![row(vec![age()])];
+ let older = vec![row(vec![variant, attrs])];
+ let sources = [
+ PhysicalRowSource {
+ source_index: 0,
+ fields: &newer,
+ },
+ PhysicalRowSource {
+ source_index: 1,
+ fields: &older,
+ },
+ ];
+ let selection = select_nested_row(&requested,
&sources).unwrap().unwrap();
+ assert_eq!(selection.children, vec![Some(1), Some(1)]);
+ assert_eq!(selection.whole_source, None);
+ let old_request = source_read_field(&requested, ¤t, 1, &older,
&selection).unwrap();
+ let DataType::Row(old_row) = old_request.data_type() else {
+ panic!()
+ };
+ assert_eq!(old_row.fields(), &[extraction, selected]);
+ }
+
#[test]
fn latest_leaf_provider_wins_across_partial_files() {
let newer = vec![row(vec![age()])];
diff --git a/crates/paimon/src/table/data_evolution_reader.rs
b/crates/paimon/src/table/data_evolution_reader.rs
index 3e648ca4..88b0bd46 100644
--- a/crates/paimon/src/table/data_evolution_reader.rs
+++ b/crates/paimon/src/table/data_evolution_reader.rs
@@ -1173,6 +1173,15 @@ impl DataEvolutionReader {
.unwrap_or_else(|| {
arrow_array::new_null_array(target_field.data_type(), rows_to_emit)
});
+ let array = if let Some((source_idx, field_offset)) =
provider {
+ crate::arrow::nested_evolution::evolve_field(
+ &array,
+
&source_plan.sources[*source_idx].read_fields()[*field_offset],
+ &read_type[idx],
+ )?
+ } else {
+ array
+ };
columns.push(array);
}
@@ -2499,7 +2508,13 @@ fn build_source_plan_with_row_id_pushdown(
}
}
}
- let source_idx = if field.data_type().is_blob_file_field()
+ let source_field = table_fields
+ .iter()
+ .find(|source| source.id() == field.id())
+ .filter(|_| crate::spec::is_map_selected_keys_field(field))
+ .filter(|source| source.data_type().is_blob_file_field())
+ .unwrap_or(field);
+ let source_idx = if source_field.data_type().is_blob_file_field()
&& !blob_descriptor_fields.contains(field.name())
{
blob_source_indices.get(&field.id()).copied()
@@ -2516,7 +2531,7 @@ fn build_source_plan_with_row_id_pushdown(
};
if let Some(source_idx) = source_idx {
- let field_offset =
sources[source_idx].add_read_field(field.clone());
+ let field_offset =
sources[source_idx].add_read_field(source_field.clone());
column_plan.push(Some((source_idx, field_offset)));
nested_plan.push(None);
} else if !field.data_type().is_nullable() {
@@ -4140,6 +4155,37 @@ mod tests {
);
}
+ #[test]
+ fn selected_map_blob_uses_the_blob_provider_and_original_map_type() {
+ let map = DataField::new(
+ 2,
+ "payload".into(),
+ DataType::Map(crate::spec::MapType::new(
+ DataType::VarChar(crate::spec::VarCharType::string_type()),
+ DataType::Blob(BlobType::new()),
+ )),
+ );
+ let selected = crate::spec::map_selected_keys_field(&map,
&["key".into()]).unwrap();
+ let files = vec![
+ data_file("others.parquet", 0, 3, 1, Some(vec!["id"])),
+ data_file("payload.blob", 0, 3, 1, Some(vec!["payload"])),
+ ];
+ let group = PreparedMergeGroup::new(&files).unwrap();
+ let plan = build_source_plan_with_row_id_pushdown(
+ &group,
+ &[resolved_info(vec![1]), resolved_info(vec![2])],
+ &[selected],
+ std::slice::from_ref(&map),
+ &HashSet::new(),
+ false,
+ false,
+ )
+ .unwrap();
+ assert_eq!(plan.column_plan, vec![Some((1, 0))]);
+ assert!(matches!(plan.sources[1], FieldSource::BlobBunch { .. }));
+ assert_eq!(plan.sources[1].read_fields(), &[map]);
+ }
+
#[test]
fn test_build_source_plan_groups_multiple_blob_columns() {
let files = vec![
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 8d9c5eb0..82092a58 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -548,6 +548,8 @@ impl DataFileReader {
let read_timing = self.read_timing.clone();
let target_schema = build_target_arrow_schema(&read_type)?;
+ let row_tracking_enabled =
+
crate::spec::CoreOptions::new(&table_options).row_tracking_enabled();
let file_fields = data_fields.clone().unwrap_or_else(||
table_fields.clone());
let data_schema_fields = data_schema_fields_for_file(
&file_fields,
@@ -555,16 +557,18 @@ impl DataFileReader {
data_schema_fields.as_deref(),
)?;
// What the reader is asked for.
- let projected_read_fields: Vec<DataField> =
- if data_fields.is_some() || file_meta.write_cols.is_some() {
- read_data_fields(&data_schema_fields, &read_type,
self.nested_field_enabled)?
- } else {
- read_type
- .iter()
- .filter(|field| field.name() != ROW_ID_FIELD_NAME)
- .cloned()
- .collect()
- };
+ let projected_read_fields: Vec<DataField> = if data_fields.is_some()
+ || 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)?
+ } else {
+ read_type
+ .iter()
+ .filter(|field| field.name() != ROW_ID_FIELD_NAME)
+ .cloned()
+ .collect()
+ };
let path_to_read = split.data_file_path(&file_meta);
let configured_reader = create_format_reader_with_budget(
&path_to_read,
@@ -583,15 +587,17 @@ impl DataFileReader {
// The decoded batch is described by `format_read_fields`, so map
// `read_type` onto *that* list: its entries carry the types the
columns
// actually come back as, which is what reconciling them needs.
- let (index_mapping, source_fields) =
- if data_fields.is_some() || file_meta.write_cols.is_some() {
- (
- create_index_mapping(&read_type, &format_read_fields),
- Some(format_read_fields.clone()),
- )
- } else {
- (None, None)
- };
+ let (index_mapping, source_fields) = if data_fields.is_some()
+ || file_meta.write_cols.is_some()
+ || read_type.iter().any(contains_selected_map)
+ {
+ (
+ create_index_mapping(&read_type, &format_read_fields),
+ Some(format_read_fields.clone()),
+ )
+ } else {
+ (None, None)
+ };
// Remap predicates from table-level to file-level indices.
let file_predicates = if row_id_residual {
@@ -719,6 +725,18 @@ impl DataFileReader {
.map(|col_idx| batch.column(col_idx))
});
+ if row_tracking_enabled
+ && read_type[i].id() ==
crate::spec::SEQUENCE_NUMBER_FIELD_ID
+ && target_field.name() ==
crate::spec::SEQUENCE_NUMBER_FIELD_NAME
+ {
+ columns.push(row_tracking_sequence_column(
+ source_col,
+ file_meta.max_sequence_number,
+ num_rows,
+ )?);
+ continue;
+ }
+
match (source_col, source_field) {
(Some(col), Some(source_field)) => {
columns.push(reconcile_column(col, source_field,
&read_type[i])?);
@@ -850,16 +868,18 @@ impl DataFileReader {
let data_schema_fields =
data_schema_fields_for_file(&file_fields,
file_meta.write_cols.as_deref(), None)?;
// What the reader is asked for.
- let projected_read_fields: Vec<DataField> =
- if data_fields.is_some() || file_meta.write_cols.is_some() {
- read_data_fields(&data_schema_fields, &read_type,
self.nested_field_enabled)?
- } else {
- read_type
- .iter()
- .filter(|field| field.name() != ROW_ID_FIELD_NAME)
- .cloned()
- .collect()
- };
+ let projected_read_fields: Vec<DataField> = if data_fields.is_some()
+ || 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)?
+ } else {
+ read_type
+ .iter()
+ .filter(|field| field.name() != ROW_ID_FIELD_NAME)
+ .cloned()
+ .collect()
+ };
let path_to_read = split.data_file_path(&file_meta);
let configured_reader = create_format_reader_with_budget(
&path_to_read,
@@ -878,15 +898,17 @@ impl DataFileReader {
// The decoded batch is described by `format_read_fields`, so map
// `read_type` onto *that* list: its entries carry the types the
columns
// actually come back as, which is what reconciling them needs.
- let (index_mapping, source_fields) =
- if data_fields.is_some() || file_meta.write_cols.is_some() {
- (
- create_index_mapping(&read_type, &format_read_fields),
- Some(format_read_fields.clone()),
- )
- } else {
- (None, None)
- };
+ let (index_mapping, source_fields) = if data_fields.is_some()
+ || file_meta.write_cols.is_some()
+ || read_type.iter().any(contains_selected_map)
+ {
+ (
+ create_index_mapping(&read_type, &format_read_fields),
+ Some(format_read_fields.clone()),
+ )
+ } else {
+ (None, None)
+ };
// Remap predicates from table-level to file-level indices.
let file_predicates = {
@@ -955,19 +977,16 @@ fn reconcile_column(
source_field: &DataField,
target_field: &DataField,
) -> crate::Result<Arc<dyn Array>> {
- crate::arrow::nested_evolution::evolve_column(
- col,
- source_field.data_type(),
- target_field.data_type(),
- )
- .map_err(|e| Error::UnexpectedError {
- message: format!(
- "Failed to reconcile column '{}' read as {:?} with read type {:?}",
- target_field.name(),
- col.data_type(),
- target_field.data_type()
- ),
- source: Some(Box::new(e)),
+ crate::arrow::nested_evolution::evolve_field(col, source_field,
target_field).map_err(|e| {
+ Error::UnexpectedError {
+ message: format!(
+ "Failed to reconcile column '{}' read as {:?} with read type
{:?}",
+ target_field.name(),
+ col.data_type(),
+ target_field.data_type()
+ ),
+ source: Some(Box::new(e)),
+ }
})
}
@@ -1041,6 +1060,36 @@ fn project_file_batch(
})
}
+fn contains_selected_map(field: &DataField) -> bool {
+ crate::spec::is_map_selected_keys_field(field)
+ || matches!(field.data_type(), DataType::Row(row) if
row.fields().iter().any(contains_selected_map))
+}
+
+fn row_tracking_sequence_column(
+ physical: Option<&Arc<dyn Array>>,
+ max_sequence_number: i64,
+ num_rows: usize,
+) -> crate::Result<Arc<dyn Array>> {
+ let physical = physical
+ .map(|array| {
+ array
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Row-tracking sequence column must be
BIGINT".into(),
+ source: None,
+ })
+ })
+ .transpose()?;
+ Ok(Arc::new(Int64Array::from_iter_values((0..num_rows).map(
+ |index| {
+ physical
+ .filter(|array| !array.is_null(index))
+ .map_or(max_sequence_number, |array| array.value(index))
+ },
+ ))))
+}
+
fn read_data_fields(
all_data_fields: &[DataField],
expected_fields: &[DataField],
@@ -1057,10 +1106,24 @@ fn read_data_fields(
data_field.data_type(),
nested_field_enabled,
)? {
- read_fields.push(data_field_with_type(data_field,
pruned_type));
+ let mut field = data_field_with_type(data_field, pruned_type);
+ if crate::spec::is_map_selected_keys_field(expected) {
+ field =
field.with_description(expected.description().map(str::to_string));
+ }
+ read_fields.push(field);
}
}
}
+ for expected in expected_fields {
+ if expected.id() == crate::spec::SEQUENCE_NUMBER_FIELD_ID
+ && expected.name() == crate::spec::SEQUENCE_NUMBER_FIELD_NAME
+ && !read_fields.iter().any(|field| field.id() == expected.id())
+ {
+ // Row tracking stores this physical system column outside the
+ // logical table schema, including alongside selected MAP keys.
+ read_fields.push(expected.clone());
+ }
+ }
Ok(read_fields)
}
@@ -1077,22 +1140,8 @@ fn prune_data_type(
let DataType::Row(data_row) = data_type else {
return Ok(Some(data_type.clone()));
};
- let mut fields = Vec::new();
- for read_field in read_row.fields() {
- if let Some(data_field) = data_row
- .fields()
- .iter()
- .find(|field| field.id() == read_field.id())
- {
- if let Some(pruned_type) = prune_data_type(
- read_field.data_type(),
- data_field.data_type(),
- nested_field_enabled,
- )? {
- fields.push(data_field_with_type(data_field,
pruned_type));
- }
- }
- }
+ let mut fields =
+ read_data_fields(data_row.fields(), read_row.fields(),
nested_field_enabled)?;
if fields.is_empty() {
if !nested_field_enabled {
return Ok(None);
@@ -1515,6 +1564,52 @@ mod row_tests {
}
}
+ #[test]
+ fn row_tracking_sequence_fills_only_missing_values_from_file_metadata() {
+ let physical: Arc<dyn Array> = Arc::new(Int64Array::from(vec![Some(3),
None, Some(5)]));
+ let filled = row_tracking_sequence_column(Some(&physical), 9,
3).unwrap();
+ assert_eq!(
+ filled.as_any().downcast_ref::<Int64Array>().unwrap(),
+ &Int64Array::from(vec![3, 9, 5]),
+ );
+ let absent = row_tracking_sequence_column(None, 9, 3).unwrap();
+ assert_eq!(
+ absent.as_any().downcast_ref::<Int64Array>().unwrap(),
+ &Int64Array::from(vec![9, 9, 9]),
+ );
+ }
+
+ #[test]
+ fn selected_map_read_preserves_physical_sequence_number() {
+ let map = field(
+ 1,
+ "attrs",
+ DataType::Map(crate::spec::MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Int(IntType::new()),
+ )),
+ );
+ let selected = crate::spec::map_selected_keys_field(&map,
&["key".into()]).unwrap();
+ let sequence = field(
+ crate::spec::SEQUENCE_NUMBER_FIELD_ID,
+ crate::spec::SEQUENCE_NUMBER_FIELD_NAME,
+ DataType::BigInt(BigIntType::with_nullable(false)),
+ );
+ let read = read_data_fields(
+ std::slice::from_ref(&map),
+ &[selected, sequence.clone()],
+ false,
+ )
+ .unwrap();
+ assert_eq!(read.len(), 2);
+ assert_eq!(read[0].data_type(), map.data_type());
+ assert_eq!(
+ read[0].description(),
+ Some("__PAIMON_MAP_SELECTED_KEYS:key")
+ );
+ assert_eq!(read[1], sequence);
+ }
+
#[test]
fn read_data_fields_preserves_variant_extraction_row_type() {
let configured = DataType::Row(RowType::new(vec![field(
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index e36ed774..0e67eae3 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -38,9 +38,9 @@ use crate::file_index::evaluator::evaluate_file_index;
use crate::file_index::file_index_result::FileIndexResult;
use crate::io::FileIO;
use crate::spec::{
- is_variant_extraction_row, is_variant_extraction_row_type, BigIntType,
CoreOptions, DataField,
- DataFileMeta, DataType as PaimonDataType, MergeEngine,
PartialUpdateConfig, Predicate,
- TinyIntType, SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME,
VALUE_KIND_FIELD_ID,
+ is_variant_extraction_row, BigIntType, CoreOptions, DataField,
DataFileMeta,
+ DataType as PaimonDataType, MergeEngine, PartialUpdateConfig, Predicate,
TinyIntType,
+ SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID,
VALUE_KIND_FIELD_NAME,
};
use crate::table::schema_manager::SchemaManager;
@@ -126,6 +126,11 @@ pub(super) fn retain_primary_key_conjuncts(
.collect()
}
+fn contains_variant_extraction(field: &DataField) -> bool {
+ crate::spec::is_variant_extraction_row_type(field.data_type())
+ || matches!(field.data_type(), PaimonDataType::Row(row) if
row.fields().iter().any(contains_variant_extraction))
+}
+
fn widen_partial_update_sequence_group_fields(
merge_engine: MergeEngine,
table_options: &HashMap<String, String>,
@@ -397,27 +402,33 @@ impl KeyValueFileReader {
// File readers run before PK deduplication. Keep Variant values in
their
// logical storage type until the visible row has been selected: a
strict
// extraction must not fail on an older, overwritten version.
+ let aggregate_fields = if self.config.merge_engine ==
MergeEngine::PartialUpdate {
+
PartialUpdateConfig::new(&self.config.table_options).validated_aggregate_functions(
+ &self.config.table_fields,
+ &self.config.table_primary_keys,
+ )?
+ } else {
+ HashMap::new()
+ };
let merge_read_type: Vec<DataField> = self
.config
.read_type
.iter()
.map(|field| {
- if !is_variant_extraction_row_type(field.data_type()) {
- return Ok(field.clone());
- }
- self.config
+ let source = self
+ .config
.table_fields
.iter()
- .find(|source| source.id() == field.id())
- .filter(|source| matches!(source.data_type(),
PaimonDataType::Variant(_)))
- .cloned()
- .ok_or_else(|| Error::DataInvalid {
- message: format!(
- "Variant extraction field '{}' has no matching
VARIANT table field",
- field.name()
- ),
- source: None,
- })
+ .find(|source| source.id() == field.id());
+ if let Some(source) = source {
+ let aggregate = self.config.merge_engine ==
MergeEngine::Aggregation
+ || aggregate_fields.contains_key(source.name());
+ let has_variant = contains_variant_extraction(field);
+ if field.data_type() != source.data_type() && (has_variant
|| aggregate) {
+ return Ok(source.clone());
+ }
+ }
+ Ok(field.clone())
})
.collect::<crate::Result<_>>()?;
@@ -818,7 +829,10 @@ impl KeyValueFileReader {
{
assemble_variant_extraction_array(column.as_ref(), row)
}
- _ => Ok(column.clone()),
+ _ => {
+ let source =
merge_read_type.iter().find(|field| field.id() ==
config.read_type[out_idx].id()).unwrap();
+
crate::arrow::nested_evolution::evolve_field(column, source,
&config.read_type[out_idx])
+ },
}
})
.collect::<crate::Result<Vec<_>>>()?;
diff --git a/crates/paimon/src/table/query_auth.rs
b/crates/paimon/src/table/query_auth.rs
index fb24d3ba..15d03d6d 100644
--- a/crates/paimon/src/table/query_auth.rs
+++ b/crates/paimon/src/table/query_auth.rs
@@ -77,11 +77,10 @@ pub(crate) async fn reject_unauthorized_stats(
}
let older = schemas.schema(file.schema_id).await?;
if let Some(gone) = older.fields().iter().find(|f| {
- !current.fields().iter().any(|c| {
- c.id() == f.id()
- && c.name() == f.name()
- && contains(c.data_type(), f.data_type())
- })
+ !current
+ .fields()
+ .iter()
+ .any(|c| c.id() == f.id() && c.name() == f.name() &&
contains_field(c, f))
}) {
return refuse(gone.name());
}
@@ -90,17 +89,29 @@ pub(crate) async fn reject_unauthorized_stats(
Ok(())
}
+fn contains_field(wide: &crate::spec::DataField, narrow:
&crate::spec::DataField) -> bool {
+ if let (crate::spec::DataType::Map(map), crate::spec::DataType::Row(row)) =
+ (wide.data_type(), narrow.data_type())
+ {
+ return matches!(map.key_type(), crate::spec::DataType::VarChar(_))
+ && crate::spec::map_selected_keys(narrow).is_ok_and(|keys|
keys.is_some())
+ && row
+ .fields()
+ .iter()
+ .all(|child|
map.value_type().equals_ignore_nullable(child.data_type()));
+ }
+ contains(wide.data_type(), narrow.data_type())
+}
+
/// Whether `narrow` reads nothing `wide` lacks: nested children match by id
/// and name, so a `ROW` projection passes and a re-added child does not.
fn contains(wide: &crate::spec::DataType, narrow: &crate::spec::DataType) ->
bool {
use crate::spec::DataType;
match (wide, narrow) {
(DataType::Row(w), DataType::Row(n)) => n.fields().iter().all(|nf| {
- w.fields().iter().any(|wf| {
- wf.id() == nf.id()
- && wf.name() == nf.name()
- && contains(wf.data_type(), nf.data_type())
- })
+ w.fields()
+ .iter()
+ .any(|wf| wf.id() == nf.id() && wf.name() == nf.name() &&
contains_field(wf, nf))
}),
(DataType::Array(w), DataType::Array(n)) => contains(w.element_type(),
n.element_type()),
(DataType::Multiset(w), DataType::Multiset(n)) => {
@@ -110,7 +121,9 @@ fn contains(wide: &crate::spec::DataType, narrow:
&crate::spec::DataType) -> boo
contains(w.key_type(), n.key_type()) && contains(w.value_type(),
n.value_type())
}
// A `variant_get` pushdown reads a `VARIANT` column as a `ROW` of
paths.
- (DataType::Variant(_), DataType::Row(_)) => true,
+ (DataType::Variant(_), DataType::Row(_)) => {
+ crate::spec::is_variant_extraction_row_type(narrow)
+ }
(w, n) => w == n,
}
}
@@ -151,11 +164,9 @@ pub(crate) fn reject_noncanonical_fields(
}
// The whole shape: an older field can keep `(id, name)` and carry an
// extra nested child.
- let canonical = schema_fields.iter().any(|f| {
- f.id() == field.id()
- && f.name() == field.name()
- && contains(f.data_type(), field.data_type())
- });
+ let canonical = schema_fields
+ .iter()
+ .any(|f| f.id() == field.id() && f.name() == field.name() &&
contains_field(f, field));
if !canonical {
return Err(unsupported(&format!(
"'{}' (field id {}) is not a column of the current schema,
which is what the \
@@ -173,6 +184,40 @@ mod tests {
use super::reject_system_columns;
use crate::table::{query_auth_table, rest_query_auth_table};
+ #[test]
+ fn authorized_map_allows_only_valid_selected_value_types() {
+ use crate::spec::*;
+ let source = DataField::new(
+ 1,
+ "attrs".into(),
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Int(IntType::new()),
+ )),
+ );
+ let selected = map_selected_keys_field(&source,
&["key".into()]).unwrap();
+ super::reject_noncanonical_fields(
+ std::slice::from_ref(&selected),
+ std::slice::from_ref(&source),
+ )
+ .unwrap();
+ let forged = selected.clone().with_description(None);
+ assert!(
+ super::reject_noncanonical_fields(&[forged],
std::slice::from_ref(&source)).is_err()
+ );
+ let changed = DataField::new(
+ selected.id(),
+ selected.name().into(),
+ DataType::Row(RowType::new(vec![DataField::new(
+ 0,
+ "key".into(),
+ DataType::BigInt(BigIntType::new()),
+ )])),
+ )
+ .with_description(selected.description().map(str::to_string));
+ assert!(super::reject_noncanonical_fields(&[changed],
&[source]).is_err());
+ }
+
#[tokio::test]
async fn test_a_grant_is_pinned_to_the_handle_that_obtained_it() {
let a = crate::table::rest_query_auth_table().await;
@@ -470,7 +515,12 @@ mod tests {
fn test_a_variant_extraction_is_the_one_shape_change_allowed() {
use crate::spec::{DataField, DataType, IntType, RowType, VariantType};
let int = || DataType::Int(IntType::new());
- let row = || DataType::Row(RowType::new(vec![DataField::new(0,
"p".into(), int())]));
+ let row = || {
+ DataType::Row(RowType::new(vec![DataField::new(0, "p".into(),
int())
+ .with_description(Some(
+ crate::spec::build_variant_metadata("$.p", true,
"UTC").unwrap(),
+ ))]))
+ };
let schema = vec![
DataField::new(1, "v".into(),
DataType::Variant(VariantType::new())),
DataField::new(2, "n".into(), int()),
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index 5b8295e0..b3fd6730 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/crates/paimon/src/table/read_builder.rs
@@ -572,6 +572,8 @@ impl<'a> PaimonReadBuilder<'a> {
Some(fields) => fields,
};
+ crate::spec::validate_selected_map_fields(self.table.schema.fields(),
&read_type)?;
+
// Pass the FULL data predicate through (including `And`/`Or`/`Not`).
// Pushdown/stats skip compound nodes; the residual pass enforces the
full
// predicate exactly. Pruning here would drop compound predicates.
diff --git a/crates/paimon/src/variant.rs b/crates/paimon/src/variant.rs
index 829e9e95..d9616315 100644
--- a/crates/paimon/src/variant.rs
+++ b/crates/paimon/src/variant.rs
@@ -169,17 +169,21 @@ impl<'a> VariantRef<'a> {
}
pub fn get_path(&self, path: &str) -> Result<Option<VariantRef<'a>>> {
+ self.get_path_segments(&parse_path(path)?)
+ }
+
+ fn get_path_segments(&self, segments: &[PathSegment]) ->
Result<Option<VariantRef<'a>>> {
let mut current = *self;
- for segment in parse_path(path)? {
+ for segment in segments {
match (segment, current.kind()?) {
(PathSegment::Key(key), VariantKind::Object) => {
- let Some(next) = current.get_field_by_key(&key)? else {
+ let Some(next) = current.get_field_by_key(key)? else {
return Ok(None);
};
current = next;
}
(PathSegment::Index(index), VariantKind::Array) => {
- let Some(next) = current.get_element_at_index(index)? else
{
+ let Some(next) = current.get_element_at_index(*index)?
else {
return Ok(None);
};
current = next;
@@ -1230,12 +1234,12 @@ fn write_json(value: &[u8], metadata: &[u8], pos:
usize, out: &mut String) -> Re
}
#[derive(Clone, Debug, PartialEq, Eq)]
-enum PathSegment {
+pub(crate) enum PathSegment {
Key(String),
Index(usize),
}
-fn parse_path(path: &str) -> Result<Vec<PathSegment>> {
+pub(crate) fn parse_path(path: &str) -> Result<Vec<PathSegment>> {
let bytes = path.as_bytes();
if !bytes.starts_with(b"$") {
return data_invalid(format!("Invalid Variant path: {path}"));
@@ -2682,6 +2686,81 @@ pub(crate) fn rebuild_shredded(
builder.result()
}
+/// 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.
+pub(crate) fn extract_shredded_path(
+ row: &ShreddedRow,
+ schema: &VariantSchema,
+ path: &str,
+) -> Result<Option<GenericVariant>> {
+ let metadata = binary_field(row, schema.top_level_metadata_idx)?;
+ if !shredded_row_has_value(row, schema) && schema.variant_idx.is_some() {
+ return data_invalid("Malformed shredded Variant: missing value");
+ }
+ extract_shredded_segments(row, schema, metadata, &parse_path(path)?)
+}
+
+fn extract_shredded_segments(
+ row: &ShreddedRow,
+ schema: &VariantSchema,
+ metadata: &[u8],
+ segments: &[PathSegment],
+) -> Result<Option<GenericVariant>> {
+ let Some((segment, remaining)) = segments.split_first() else {
+ if !shredded_row_has_value(row, schema) {
+ return Ok(None);
+ }
+ let mut builder = VariantBuilder::new();
+ rebuild_shredded_into(row, metadata, schema, &mut builder)?;
+ return builder.result().map(Some);
+ };
+ 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(
+ child,
+ &schema.object_schema.as_ref().unwrap()[index].schema,
+ metadata,
+ remaining,
+ );
+ }
+ }
+ (PathSegment::Index(index), ShreddedValue::List(elements))
+ if schema.array_schema.is_some() =>
+ {
+ return match elements.get(*index) {
+ Some(element) => extract_shredded_segments(
+ element,
+ schema.array_schema.as_ref().unwrap(),
+ metadata,
+ remaining,
+ ),
+ None => Ok(None),
+ };
+ }
+ _ => {}
+ }
+ }
+ match optional_binary_field(row, schema.variant_idx)? {
+ Some(value) => {
+ let variant = GenericVariant::from_parts(value.to_vec(),
metadata.to_vec())?;
+ variant
+ .as_ref()?
+ .get_path_segments(segments)?
+ .map(|selected| selected.to_owned_variant())
+ .transpose()
+ }
+ None => Ok(None),
+ }
+}
+
pub(crate) fn cast_variant_to_shredded_value(
variant: VariantRef<'_>,
data_type: &DataType,
@@ -3307,6 +3386,37 @@ mod tests {
build_variant_schema(&physical).unwrap()
}
+ #[test]
+ fn clipped_shredded_paths_do_not_rebuild_unrequested_parent_values() {
+ let full = simple_object_shredding_schema();
+ let mut clipped = full.clone();
+ clipped.variant_idx = None;
+ for (json, expected) in [
+ (r#"{"age":7,"city":"x","extra":true}"#, Some("7")),
+ ("42", None),
+ ("null", None),
+ ("[]", None),
+ ("{}", None),
+ ] {
+ let variant = GenericVariant::parse_json(json).unwrap();
+ let shredded = cast_shredded(&variant, &full).unwrap();
+ let actual = extract_shredded_path(&shredded, &clipped,
"$.age").unwrap();
+ assert_eq!(
+ actual.map(|value| value.to_json().unwrap()).as_deref(),
+ expected,
+ "{json}"
+ );
+ }
+ let malformed = ShreddedRow::new(full.num_fields);
+ assert!(extract_shredded_path(&malformed, &full, "$.age").is_err());
+ let scalar = GenericVariant::parse_json("42").unwrap();
+ let row = cast_shredded(&scalar, &full).unwrap();
+ assert!(
+ rebuild_shredded(&row, &clipped).is_err(),
+ "whole values still require the pruned fallback"
+ );
+ }
+
#[test]
fn variant_shredding_schema_matches_java_shape() {
let configured = DataType::Row(RowType::new(vec