Skip to content

Commit 0bc0405

Browse files
committed
some fixes
Signed-off-by: Adam Gutglick <adam@spiraldb.com>
1 parent 488edb5 commit 0bc0405

3 files changed

Lines changed: 296 additions & 7 deletions

File tree

encodings/parquet-variant/src/array.rs

Lines changed: 175 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,11 @@ use vortex_array::arrays::List;
1919
use vortex_array::arrays::ListArray;
2020
use vortex_array::arrays::Struct;
2121
use vortex_array::arrays::StructArray;
22+
use vortex_array::arrays::Variant;
2223
use vortex_array::arrays::VariantArray;
2324
use vortex_array::arrays::list::ListArrayExt;
2425
use vortex_array::arrays::struct_::StructArrayExt;
26+
use vortex_array::arrays::variant::VariantArrayExt;
2527
#[expect(
2628
deprecated,
2729
reason = "TODO(aduffy): figure out what to do with Parquet Variant"
@@ -42,6 +44,7 @@ use vortex_array::vtable::validity_to_child;
4244
use vortex_buffer::BitBuffer;
4345
use vortex_error::VortexExpect;
4446
use vortex_error::VortexResult;
47+
use vortex_error::vortex_err;
4548

4649
use crate::ParquetVariant;
4750
use crate::ParquetVariantArray;
@@ -307,6 +310,111 @@ fn inferred_shredded_field_validity(
307310
Ok(Validity::from_mask(validity, Nullability::Nullable))
308311
}
309312

313+
/// Reconstructs a Parquet `typed_value` tree from the storage-agnostic canonical shredded tree.
314+
///
315+
/// This is the inverse of [`logical_shredded_from_parquet_typed_value`]: canonicalization strips
316+
/// the Parquet `value`/`typed_value` wrapper shells out of the shredded tree (representing
317+
/// partially shredded fields as nested [`VariantArray`]s), and this re-adds them. The result is a
318+
/// valid Parquet `typed_value` that can be reattached to a [`ParquetVariant`] and serialized to
319+
/// Arrow, where [`parquet_variant_compute::unshred_variant`] merges it back with the residual
320+
/// `value`.
321+
///
322+
/// `forward` then `inverse` is structure-preserving for the shapes canonicalization produces;
323+
/// fields the forward transform omitted (Parquet wrapper shells with no `typed_value`) are served
324+
/// from the residual `value` and are intentionally not reconstructed here.
325+
pub(crate) fn parquet_typed_value_from_logical_shredded(
326+
shredded: ArrayRef,
327+
ctx: &mut ExecutionCtx,
328+
) -> VortexResult<ArrayRef> {
329+
if let Some(list_array) = shredded.as_opt::<List>() {
330+
let elements = parquet_shredded_field_from_logical(list_array.elements().clone(), ctx)?;
331+
return Ok(ListArray::try_new(
332+
elements,
333+
list_array.offsets().clone(),
334+
list_array.list_validity(),
335+
)?
336+
.into_array());
337+
}
338+
339+
let Some(struct_array) = shredded.as_opt::<Struct>() else {
340+
// A bare typed leaf (a fully shredded scalar) is already a valid Parquet `typed_value`.
341+
return Ok(shredded);
342+
};
343+
344+
let mut names = Vec::with_capacity(struct_array.names().len());
345+
let mut fields = Vec::with_capacity(struct_array.names().len());
346+
for (name, field) in struct_array
347+
.names()
348+
.iter()
349+
.zip(struct_array.iter_unmasked_fields())
350+
{
351+
names.push(FieldName::from(name.as_ref()));
352+
fields.push(parquet_shredded_field_from_logical(field.clone(), ctx)?);
353+
}
354+
355+
Ok(StructArray::try_new(
356+
FieldNames::from_iter(names),
357+
fields,
358+
struct_array.len(),
359+
struct_array.validity()?,
360+
)?
361+
.into_array())
362+
}
363+
364+
/// Reconstructs one Parquet shredded field shell (`{value?, typed_value}`) from its canonical
365+
/// representation, the inverse of [`logical_shredded_from_parquet_field`].
366+
fn parquet_shredded_field_from_logical(
367+
logical_field: ArrayRef,
368+
ctx: &mut ExecutionCtx,
369+
) -> VortexResult<ArrayRef> {
370+
let len = logical_field.len();
371+
372+
// Partially shredded fields canonicalize to a nested Variant whose core storage holds the
373+
// residual `value` and whose own shredded tree holds the typed children.
374+
if let Some(variant) = logical_field.as_opt::<Variant>() {
375+
let core = variant
376+
.core_storage()
377+
.as_opt::<ParquetVariant>()
378+
.ok_or_else(|| {
379+
vortex_err!(
380+
"cannot rebuild Parquet shredded field: nested Variant lacks Parquet Variant core storage"
381+
)
382+
})?;
383+
let value = core.value_array().cloned().ok_or_else(|| {
384+
vortex_err!("cannot rebuild Parquet shredded field: partially shredded Variant has no residual value")
385+
})?;
386+
let typed_value = variant
387+
.shredded()
388+
.cloned()
389+
.map(|shredded| parquet_typed_value_from_logical_shredded(shredded, ctx))
390+
.transpose()?;
391+
392+
let mut names = vec![FieldName::from("value")];
393+
let mut fields = vec![value];
394+
if let Some(typed_value) = typed_value {
395+
names.push(FieldName::from("typed_value"));
396+
fields.push(typed_value);
397+
}
398+
return Ok(StructArray::try_new(
399+
FieldNames::from_iter(names),
400+
fields,
401+
len,
402+
Validity::NonNullable,
403+
)?
404+
.into_array());
405+
}
406+
407+
// Fully shredded field: rebuild its typed subtree and wrap it in a `typed_value`-only shell.
408+
let typed_value = parquet_typed_value_from_logical_shredded(logical_field, ctx)?;
409+
Ok(StructArray::try_new(
410+
FieldNames::from_iter([FieldName::from("typed_value")]),
411+
vec![typed_value],
412+
len,
413+
Validity::NonNullable,
414+
)?
415+
.into_array())
416+
}
417+
310418
/// Accessors and Arrow conversion for Parquet Variant storage arrays.
311419
pub trait ParquetVariantArrayExt: TypedArrayRef<ParquetVariant> {
312420
/// Returns the non-nullable Parquet Variant metadata child.
@@ -389,20 +497,26 @@ mod tests {
389497
use arrow_array::Array as _;
390498
use arrow_array::ArrayRef as ArrowArrayRef;
391499
use arrow_array::Int32Array;
500+
use arrow_array::StringArray;
392501
use arrow_array::StructArray;
393502
use arrow_array::builder::BinaryViewBuilder;
394503
use arrow_buffer::NullBuffer;
395504
use arrow_schema::DataType;
396505
use arrow_schema::Field;
397506
use arrow_schema::Fields;
398507
use parquet_variant::Variant as PqVariant;
508+
use parquet_variant_compute::ShreddedSchemaBuilder;
399509
use parquet_variant_compute::VariantArray as ArrowVariantArray;
400510
use parquet_variant_compute::VariantArrayBuilder;
511+
use parquet_variant_compute::json_to_variant;
512+
use parquet_variant_compute::shred_variant;
513+
use vortex_array::Canonical;
401514
use vortex_array::IntoArray;
402515
use vortex_array::LEGACY_SESSION;
403516
use vortex_array::VortexSessionExecute;
404517
use vortex_array::arrays::PrimitiveArray;
405518
use vortex_array::arrays::VarBinViewArray;
519+
use vortex_array::arrays::variant::VariantArrayExt;
406520
use vortex_array::assert_arrays_eq;
407521
use vortex_array::dtype::DType;
408522
use vortex_array::dtype::Nullability;
@@ -413,6 +527,7 @@ mod tests {
413527

414528
use crate::ParquetVariant;
415529
use crate::array::ParquetVariantArrayExt;
530+
use crate::array::parquet_typed_value_from_logical_shredded;
416531

417532
fn assert_arrow_variant_storage_roundtrip(struct_array: StructArray) -> VortexResult<()> {
418533
let arrow_variant = ArrowVariantArray::try_new(&struct_array)?;
@@ -678,4 +793,64 @@ mod tests {
678793

679794
assert_arrow_variant_storage_roundtrip(struct_array)
680795
}
796+
797+
/// `parquet_typed_value_from_logical_shredded` must invert the wrapper-stripping that
798+
/// canonicalization performs: an object-shredded Parquet variant, once canonicalized and then
799+
/// rebuilt, must produce the same per-row values as the original.
800+
#[test]
801+
fn parquet_typed_value_inverse_roundtrips_object_shredding() -> VortexResult<()> {
802+
// Shred `$.a` as Int32 over conforming, non-conforming, and missing-field rows.
803+
let json: ArrowArrayRef = Arc::new(StringArray::from(vec![
804+
Some(r#"{"a":1,"b":"x"}"#),
805+
Some(r#"{"a":"not-a-number","b":"y"}"#),
806+
Some(r#"{"b":"z"}"#),
807+
]));
808+
let shredding = ShreddedSchemaBuilder::new()
809+
.with_path("a", &DataType::Int32)?
810+
.build();
811+
let shredded = shred_variant(&json_to_variant(&json)?, &shredding)?;
812+
let original = ParquetVariant::from_arrow_variant(&shredded)?;
813+
assert!(
814+
original
815+
.as_opt::<ParquetVariant>()
816+
.ok_or_else(|| vortex_err!("expected parquet variant"))?
817+
.typed_value_array()
818+
.is_some(),
819+
"fixture must be shredded"
820+
);
821+
822+
// Canonicalize: the forward transform lifts `typed_value` into a logical shredded child.
823+
let mut ctx = LEGACY_SESSION.create_execution_ctx();
824+
let Canonical::Variant(canonical) = original.clone().execute::<Canonical>(&mut ctx)? else {
825+
return Err(vortex_err!("expected canonical variant"));
826+
};
827+
let core = canonical
828+
.core_storage()
829+
.as_opt::<ParquetVariant>()
830+
.ok_or_else(|| vortex_err!("expected parquet variant core storage"))?;
831+
let logical = canonical
832+
.shredded()
833+
.ok_or_else(|| vortex_err!("expected canonical shredded child"))?
834+
.clone();
835+
836+
// Inverse transform: rebuild a Parquet `typed_value` and reattach it.
837+
let rebuilt_typed_value = parquet_typed_value_from_logical_shredded(logical, &mut ctx)?;
838+
let rebuilt = ParquetVariant::try_new(
839+
ParquetVariantArrayExt::validity(&core),
840+
core.metadata_array().clone(),
841+
core.value_array().cloned(),
842+
Some(rebuilt_typed_value),
843+
)?
844+
.into_array();
845+
846+
assert_eq!(rebuilt.len(), original.len());
847+
for idx in 0..original.len() {
848+
assert_eq!(
849+
rebuilt.execute_scalar(idx, &mut ctx)?,
850+
original.execute_scalar(idx, &mut ctx)?,
851+
"row {idx}"
852+
);
853+
}
854+
Ok(())
855+
}
681856
}

encodings/parquet-variant/src/arrow.rs

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ use vortex_session::registry::Id;
3636

3737
use crate::ParquetVariant;
3838
use crate::ParquetVariantArrayExt;
39+
use crate::array::parquet_typed_value_from_logical_shredded;
3940

4041
/// Arrow canonical extension name for Parquet Variant storage.
4142
const PARQUET_VARIANT_ARROW_EXTENSION_NAME: &str = "arrow.parquet.variant";
@@ -138,11 +139,16 @@ pub(crate) fn parquet_variant_for_export(
138139
return Ok(core_storage);
139140
};
140141

142+
// The canonical shredded child has had its Parquet `value`/`typed_value` wrapper shells
143+
// stripped; rebuild them so the reattached `typed_value` is valid Parquet storage that
144+
// `to_arrow` and `unshred_variant` can consume.
145+
let typed_value = parquet_typed_value_from_logical_shredded(shredded.clone(), ctx)?;
146+
141147
ParquetVariant::try_new(
142148
ParquetVariantArrayExt::validity(&parquet_core),
143149
parquet_core.metadata_array().clone(),
144150
parquet_core.value_array().cloned(),
145-
Some(shredded.clone()),
151+
Some(typed_value),
146152
)
147153
.map(IntoArray::into_array)
148154
}

0 commit comments

Comments
 (0)