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 c9cf6c82 [python] Add batch numeric extraction for Variant fields
(#997)
c9cf6c82 is described below
commit c9cf6c82e2f74221220ff32c298582a7cd0c69ad
Author: XiaoHongbo <[email protected]>
AuthorDate: Wed Sep 30 20:46:55 2026 +0800
[python] Add batch numeric extraction for Variant fields (#997)
---
bindings/python/python/pypaimon_rust/data.pyi | 26 +++
bindings/python/src/lib.rs | 2 +
bindings/python/src/variant.rs | 55 +++++
bindings/python/tests/test_variant.py | 65 ++++++
crates/paimon/src/arrow/mod.rs | 2 +
crates/paimon/src/arrow/variant.rs | 284 ++++++++++++++++++++++++++
crates/paimon/src/variant.rs | 21 +-
crates/paimon/src/variant/numeric.rs | 246 ++++++++++++++++++++++
8 files changed, 693 insertions(+), 8 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/data.pyi
b/bindings/python/python/pypaimon_rust/data.pyi
new file mode 100644
index 00000000..c46aa2bb
--- /dev/null
+++ b/bindings/python/python/pypaimon_rust/data.pyi
@@ -0,0 +1,26 @@
+# 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.
+
+from typing import Sequence
+
+import pyarrow
+
+def variant_get_numeric_fields(
+ column: pyarrow.Array, fields: Sequence[str]
+) -> pyarrow.FixedSizeListArray:
+ """Extract literal top-level numeric fields as float32 in the requested
order."""
+ ...
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index 89f690a6..310cf78a 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/lib.rs
@@ -26,6 +26,7 @@ mod schema;
mod table;
mod udf;
mod update_assignment;
+mod variant;
mod write;
// ---- #285: observability ----
mod partition;
@@ -35,5 +36,6 @@ mod tag;
#[pymodule]
fn pypaimon_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
context::register_module(py, m)?;
+ variant::register_module(py, m)?;
Ok(())
}
diff --git a/bindings/python/src/variant.rs b/bindings/python/src/variant.rs
new file mode 100644
index 00000000..f59ae351
--- /dev/null
+++ b/bindings/python/src/variant.rs
@@ -0,0 +1,55 @@
+// 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.
+
+use arrow::array::{make_array, Array, ArrayData, StructArray};
+use arrow::pyarrow::{FromPyArrow, ToPyArrow};
+use pyo3::prelude::*;
+
+use crate::error::to_py_err;
+
+/// Extract literal top-level numeric Variant fields into a row-major array.
+#[pyfunction]
+fn variant_get_numeric_fields(
+ py: Python<'_>,
+ column: &Bound<'_, PyAny>,
+ fields: Vec<String>,
+) -> PyResult<Py<PyAny>> {
+ let column = make_array(ArrayData::from_pyarrow_bound(column)?);
+ let output = py
+ .detach(|| {
+ let column = column
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .ok_or_else(|| paimon::Error::DataInvalid {
+ message: "Expected a PyArrow Variant
StructArray".to_string(),
+ source: None,
+ })?;
+ paimon::arrow::variant_get_numeric_fields(column, &fields)
+ })
+ .map_err(to_py_err)?;
+ Ok(output.to_data().to_pyarrow(py)?.unbind())
+}
+
+pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) ->
PyResult<()> {
+ let this = PyModule::new(py, "data")?;
+ this.add_function(wrap_pyfunction!(variant_get_numeric_fields, &this)?)?;
+ m.add_submodule(&this)?;
+ py.import("sys")?
+ .getattr("modules")?
+ .set_item("pypaimon_rust.data", this)?;
+ Ok(())
+}
diff --git a/bindings/python/tests/test_variant.py
b/bindings/python/tests/test_variant.py
new file mode 100644
index 00000000..41025c98
--- /dev/null
+++ b/bindings/python/tests/test_variant.py
@@ -0,0 +1,65 @@
+# 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.
+
+import pyarrow as pa
+import pytest
+
+from pypaimon_rust.data import variant_get_numeric_fields
+
+
+def test_variant_get_numeric_fields_returns_ordered_fixed_size_lists():
+ # Java-compatible encoding for {"age": 27, "city": "Beijing"}.
+ value = bytes(
+ [
+ 0x02,
+ 0x02,
+ 0x00,
+ 0x01,
+ 0x00,
+ 0x02,
+ 0x0A,
+ 0x0C,
+ 0x1B,
+ 0x1D,
+ ord("B"),
+ ord("e"),
+ ord("i"),
+ ord("j"),
+ ord("i"),
+ ord("n"),
+ ord("g"),
+ ]
+ )
+ metadata = bytes([0x01, 0x02, 0x00, 0x03, 0x07]) + b"agecity"
+ column = pa.StructArray.from_arrays(
+ [pa.array([value, b""], pa.binary()), pa.array([metadata, b""],
pa.binary())],
+ fields=[
+ pa.field("value", pa.binary(), nullable=False),
+ pa.field("metadata", pa.binary(), nullable=False),
+ ],
+ mask=pa.array([False, True]),
+ )
+
+ output = variant_get_numeric_fields(column, ["missing", "age"])
+
+ assert output.type == pa.list_(pa.field("item", pa.float32()), 2)
+ assert output.to_pylist() == [[None, 27.0], None]
+
+ with pytest.raises(NotImplementedError, match="city"):
+ variant_get_numeric_fields(column, ["city"])
+ with pytest.raises(ValueError, match="must not be empty"):
+ variant_get_numeric_fields(column, [])
diff --git a/crates/paimon/src/arrow/mod.rs b/crates/paimon/src/arrow/mod.rs
index 637d6ee2..0f65e4a6 100644
--- a/crates/paimon/src/arrow/mod.rs
+++ b/crates/paimon/src/arrow/mod.rs
@@ -24,9 +24,11 @@ pub(crate) mod residual;
mod row_filter;
pub(crate) mod schema_evolution;
pub(crate) mod shredding;
+mod variant;
pub use read_budget::ReadBudget;
pub use row_filter::{RowFilter, RowFilterContext, RowFilterFactory};
+pub use variant::variant_get_numeric_fields;
use crate::spec::{
ArrayType, BigIntType, BooleanType, DataField, DataType as PaimonDataType,
DateType,
diff --git a/crates/paimon/src/arrow/variant.rs
b/crates/paimon/src/arrow/variant.rs
new file mode 100644
index 00000000..9534d86a
--- /dev/null
+++ b/crates/paimon/src/arrow/variant.rs
@@ -0,0 +1,284 @@
+// 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.
+
+use std::sync::Arc;
+
+use arrow_array::builder::{FixedSizeListBuilder, Float32Builder};
+use arrow_array::{Array, BinaryArray, FixedSizeListArray, StructArray};
+use arrow_schema::{DataType, Field};
+
+use crate::variant::VariantFloat32Projection;
+use crate::{Error, Result};
+
+/// Extracts literal top-level numeric fields from an Arrow Variant column.
+///
+/// The result is row-major and preserves `fields` order. Missing fields,
+/// non-object roots, and Variant nulls become child nulls; SQL-null input rows
+/// remain parent nulls. Numeric values are converted to `f32`, which may lose
+/// precision.
+pub fn variant_get_numeric_fields(
+ column: &StructArray,
+ fields: &[String],
+) -> Result<FixedSizeListArray> {
+ if fields.is_empty() {
+ return data_invalid("Variant numeric field list must not be empty");
+ }
+
+ let width = i32::try_from(fields.len()).map_err(|_|
Error::ResourceExhausted {
+ message: "Variant numeric field count exceeds Arrow
limits".to_string(),
+ })?;
+ let value_capacity =
+ column
+ .len()
+ .checked_mul(fields.len())
+ .ok_or_else(|| Error::ResourceExhausted {
+ message: "Variant numeric output exceeds addressable
memory".to_string(),
+ })?;
+
+ let value_column = variant_binary_child(column, 0, "value")?;
+ let metadata_column = variant_binary_child(column, 1, "metadata")?;
+ let mut builder = FixedSizeListBuilder::with_capacity(
+ Float32Builder::with_capacity(value_capacity),
+ width,
+ column.len(),
+ )
+ .with_field(Arc::new(Field::new("item", DataType::Float32, true)));
+ let mut projection_metadata = None;
+ let mut projection = None;
+ let mut offsets = Vec::new();
+ let mut extracted = vec![None; fields.len()];
+
+ for row in 0..column.len() {
+ if column.is_null(row) {
+ append_null_row(&mut builder, fields.len());
+ continue;
+ }
+ if value_column.is_null(row) || metadata_column.is_null(row) {
+ return data_invalid(format!(
+ "Variant row {row} has a null value or metadata child"
+ ));
+ }
+
+ let metadata = metadata_column.value(row);
+ if projection_metadata != Some(metadata) {
+ projection = Some(VariantFloat32Projection::new(metadata,
fields)?);
+ projection_metadata = Some(metadata);
+ }
+ projection.as_mut().unwrap().extract_float32(
+ value_column.value(row),
+ metadata,
+ &mut offsets,
+ &mut extracted,
+ )?;
+ for value in &extracted {
+ builder.values().append_option(*value);
+ }
+ builder.append(true);
+ }
+
+ Ok(builder.finish())
+}
+
+fn variant_binary_child<'a>(
+ column: &'a StructArray,
+ index: usize,
+ expected_name: &str,
+) -> Result<&'a BinaryArray> {
+ let Some(field) = column.fields().get(index) else {
+ return data_invalid("Expected Variant struct fields value and
metadata");
+ };
+ if column.num_columns() != 2
+ || field.name() != expected_name
+ || field.data_type() != &DataType::Binary
+ {
+ return data_invalid("Expected Variant struct fields value and
metadata");
+ }
+ column
+ .column(index)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!("Variant {expected_name} field must be Binary"),
+ source: None,
+ })
+}
+
+fn append_null_row(builder: &mut FixedSizeListBuilder<Float32Builder>,
field_count: usize) {
+ for _ in 0..field_count {
+ builder.values().append_null();
+ }
+ builder.append(false);
+}
+
+fn data_invalid<T>(message: impl Into<String>) -> Result<T> {
+ Err(Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::variant::GenericVariant;
+ use arrow_array::builder::BinaryBuilder;
+ use arrow_array::cast::AsArray;
+ use arrow_buffer::{BooleanBuffer, NullBuffer};
+
+ fn variant_column(rows: &[Option<&str>]) -> StructArray {
+ let mut values = BinaryBuilder::new();
+ let mut metadata = BinaryBuilder::new();
+ let mut validity = Vec::with_capacity(rows.len());
+ for row in rows {
+ match row {
+ Some(json) => {
+ let variant = GenericVariant::parse_json(json).unwrap();
+ values.append_value(variant.value());
+ metadata.append_value(variant.metadata());
+ validity.push(true);
+ }
+ None => {
+ values.append_value([]);
+ metadata.append_value([]);
+ validity.push(false);
+ }
+ }
+ }
+ StructArray::new(
+ match super::super::variant_arrow_type() {
+ DataType::Struct(fields) => fields,
+ _ => unreachable!(),
+ },
+ vec![Arc::new(values.finish()), Arc::new(metadata.finish())],
+ Some(NullBuffer::new(BooleanBuffer::from(validity))),
+ )
+ }
+
+ #[test]
+ fn extracts_numeric_fields_in_requested_order() {
+ let input = variant_column(&[
+ Some(r#"{"i":7,"f":1.25e0,"d":12.50,"missing":null}"#),
+ Some(r#"{"i":-2,"f":-0.0,"d":0}"#),
+ None,
+ ]);
+ let fields = vec![
+ "d".to_string(),
+ "i".to_string(),
+ "f".to_string(),
+ "missing".to_string(),
+ "absent".to_string(),
+ ];
+
+ let output = variant_get_numeric_fields(&input, &fields).unwrap();
+ assert_eq!(output.len(), 3);
+ assert!(output.is_valid(0));
+ assert!(output.is_valid(1));
+ assert!(output.is_null(2));
+ let values = output
+ .values()
+ .as_primitive::<arrow_array::types::Float32Type>();
+ assert_eq!(values.value(0), 12.5);
+ assert_eq!(values.value(1), 7.0);
+ assert_eq!(values.value(2), 1.25);
+ assert!(values.is_null(3));
+ assert!(values.is_null(4));
+ assert_eq!(values.value(5), 0.0);
+ assert_eq!(values.value(6), -2.0);
+ assert_eq!(values.value(7), 0.0);
+ assert!(values.is_null(8));
+ assert!(values.is_null(9));
+ }
+
+ #[test]
+ fn treats_field_names_as_literals() {
+ let input = variant_column(&[Some(r#"{"a.b":3,"a":{"b":9}}"#)]);
+ let output = variant_get_numeric_fields(&input,
&["a.b".to_string()]).unwrap();
+ let values = output
+ .values()
+ .as_primitive::<arrow_array::types::Float32Type>();
+ assert_eq!(values.value(0), 3.0);
+ }
+
+ #[test]
+ fn rejects_non_numeric_fields() {
+ let input = variant_column(&[Some(r#"{"value":"3"}"#)]);
+ let err = variant_get_numeric_fields(&input,
&["value".to_string()]).unwrap_err();
+ assert!(matches!(err, Error::Unsupported { .. }));
+ }
+
+ #[test]
+ fn rejects_empty_field_list() {
+ let input = variant_column(&[Some(r#"{"value":3}"#)]);
+ let err = variant_get_numeric_fields(&input, &[]).unwrap_err();
+ assert!(matches!(err, Error::DataInvalid { .. }));
+ }
+
+ #[test]
+ fn returns_null_fields_for_non_object_roots() {
+ let input = variant_column(&[
+ Some(r#"{"value":3}"#),
+ Some("null"),
+ Some("3"),
+ Some("[1,2]"),
+ None,
+ ]);
+ let output = variant_get_numeric_fields(&input,
&["value".to_string()]).unwrap();
+ let values = output
+ .values()
+ .as_primitive::<arrow_array::types::Float32Type>();
+
+ assert_eq!(values.value(0), 3.0);
+ assert!(values.is_null(1));
+ assert!(values.is_null(2));
+ assert!(values.is_null(3));
+ assert!(output.is_valid(0));
+ assert!(output.is_valid(1));
+ assert!(output.is_valid(2));
+ assert!(output.is_valid(3));
+ assert!(output.is_null(4));
+ }
+
+ #[test]
+ fn rejects_malformed_payloads() {
+ let mut values = BinaryBuilder::new();
+ values.append_value([0x02]);
+ let mut metadata = BinaryBuilder::new();
+ metadata.append_value([0x01, 0x00, 0x00]);
+ let input = StructArray::new(
+ match super::super::variant_arrow_type() {
+ DataType::Struct(fields) => fields,
+ _ => unreachable!(),
+ },
+ vec![Arc::new(values.finish()), Arc::new(metadata.finish())],
+ None,
+ );
+
+ let err = variant_get_numeric_fields(&input,
&["value".to_string()]).unwrap_err();
+ assert!(matches!(err, Error::DataInvalid { .. }));
+ }
+
+ #[test]
+ fn respects_sliced_array_offsets() {
+ let input = variant_column(&[Some(r#"{"v":1}"#), Some(r#"{"v":2}"#)]);
+ let sliced = input.slice(1, 1);
+ let output = variant_get_numeric_fields(&sliced,
&["v".to_string()]).unwrap();
+ let values = output
+ .values()
+ .as_primitive::<arrow_array::types::Float32Type>();
+ assert_eq!(values.value(0), 2.0);
+ }
+}
diff --git a/crates/paimon/src/variant.rs b/crates/paimon/src/variant.rs
index 2199e621..0fb8b9a2 100644
--- a/crates/paimon/src/variant.rs
+++ b/crates/paimon/src/variant.rs
@@ -27,6 +27,9 @@ use crate::spec::{
use crate::{Error, Result};
use base64::{engine::general_purpose, Engine as _};
+mod numeric;
+pub(crate) use numeric::VariantFloat32Projection;
+
const BASIC_TYPE_BITS: u8 = 2;
const BASIC_TYPE_MASK: u8 = 0x3;
const TYPE_INFO_MASK: u8 = 0x3f;
@@ -228,7 +231,7 @@ impl<'a> VariantRef<'a> {
layout.id_start + layout.id_size * i,
layout.id_size,
)?;
- if key == get_metadata_key(self.metadata, id)? {
+ if key == get_metadata_key_ref(self.metadata, id)? {
let offset = read_unsigned(
self.value,
layout.offset_start + layout.offset_size * i,
@@ -252,7 +255,7 @@ impl<'a> VariantRef<'a> {
layout.id_start + layout.id_size * mid,
layout.id_size,
)?;
- match java_string_cmp(&get_metadata_key(self.metadata, id)?,
key) {
+ match java_string_cmp(get_metadata_key_ref(self.metadata,
id)?, key) {
std::cmp::Ordering::Less => low = mid + 1,
std::cmp::Ordering::Greater => high = mid,
std::cmp::Ordering::Equal => {
@@ -951,6 +954,10 @@ fn metadata_offset_size(metadata: &[u8]) -> Result<usize> {
}
fn get_metadata_key(metadata: &[u8], id: usize) -> Result<String> {
+ Ok(get_metadata_key_ref(metadata, id)?.to_string())
+}
+
+fn get_metadata_key_ref(metadata: &[u8], id: usize) -> Result<&str> {
let offset_size = metadata_offset_size(metadata)?;
let dict_size = read_unsigned(metadata, 1, offset_size)?;
if id >= dict_size {
@@ -964,12 +971,10 @@ fn get_metadata_key(metadata: &[u8], id: usize) ->
Result<String> {
}
check_range(metadata, string_start + offset, next_offset - offset)?;
let bytes = &metadata[string_start + offset..string_start + next_offset];
- std::str::from_utf8(bytes)
- .map(|v| v.to_string())
- .map_err(|e| Error::DataInvalid {
- message: "Malformed Variant metadata UTF-8".to_string(),
- source: Some(Box::new(e)),
- })
+ std::str::from_utf8(bytes).map_err(|e| Error::DataInvalid {
+ message: "Malformed Variant metadata UTF-8".to_string(),
+ source: Some(Box::new(e)),
+ })
}
fn get_boolean(value: &[u8], pos: usize) -> Result<bool> {
diff --git a/crates/paimon/src/variant/numeric.rs
b/crates/paimon/src/variant/numeric.rs
new file mode 100644
index 00000000..34ab0bea
--- /dev/null
+++ b/crates/paimon/src/variant/numeric.rs
@@ -0,0 +1,246 @@
+// 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.
+
+use super::*;
+
+pub(crate) struct VariantFloat32Projection {
+ fields: HashMap<usize, ProjectedField>,
+ output_width: usize,
+ last_layout: Option<ObjectLayoutProjection>,
+}
+
+struct ObjectLayoutProjection {
+ field_count: usize,
+ id_size: usize,
+ ids: Vec<u8>,
+ selected: Vec<(usize, usize)>,
+}
+
+struct ProjectedField {
+ name: String,
+ outputs: Vec<usize>,
+}
+
+impl VariantFloat32Projection {
+ pub(crate) fn new(metadata: &[u8], fields: &[String]) -> Result<Self> {
+ validate_metadata(metadata)?;
+ let mut requested = HashMap::<&str, Vec<usize>>::new();
+ for (index, field) in fields.iter().enumerate() {
+ requested.entry(field).or_default().push(index);
+ }
+
+ let offset_size = metadata_offset_size(metadata)?;
+ let dict_size = read_unsigned(metadata, 1, offset_size)?;
+ let mut projected = HashMap::with_capacity(fields.len());
+ for id in 0..dict_size {
+ let name = get_metadata_key_ref(metadata, id)?;
+ if let Some(outputs) = requested.get(name) {
+ projected.insert(
+ id,
+ ProjectedField {
+ name: name.to_string(),
+ outputs: outputs.clone(),
+ },
+ );
+ }
+ }
+ Ok(Self {
+ fields: projected,
+ output_width: fields.len(),
+ last_layout: None,
+ })
+ }
+
+ pub(crate) fn extract_float32(
+ &mut self,
+ value: &[u8],
+ metadata: &[u8],
+ offsets: &mut Vec<usize>,
+ output: &mut [Option<f32>],
+ ) -> Result<()> {
+ if output.len() != self.output_width {
+ return data_invalid("Invalid Variant numeric output width");
+ }
+ output.fill(None);
+ if value_kind(value, 0)? != VariantKind::Object {
+ validate_payload(value, metadata)?;
+ return Ok(());
+ }
+
+ let layout = object_layout(value, 0)?;
+ offsets.clear();
+ offsets.reserve(layout.size + 1);
+ for index in 0..=layout.size {
+ offsets.push(read_unsigned(
+ value,
+ layout.offset_start + layout.offset_size * index,
+ layout.offset_size,
+ )?);
+ }
+ let data_size = offsets[layout.size];
+ if layout.data_start.checked_add(data_size) != Some(value.len()) {
+ return data_invalid("Malformed Variant root size");
+ }
+
+ let monotonic =
+ offsets.first() == Some(&0) && offsets.windows(2).all(|pair|
pair[0] < pair[1]);
+ let sorted_offsets = if monotonic {
+ None
+ } else {
+ let mut sorted = offsets.clone();
+ sorted.sort_unstable();
+ if sorted.first() != Some(&0)
+ || sorted.last() != Some(&data_size)
+ || sorted.windows(2).any(|pair| pair[0] == pair[1])
+ {
+ return data_invalid("Malformed Variant object offsets");
+ }
+ Some(sorted)
+ };
+
+ self.prepare_layout(value, metadata, &layout)?;
+ for (index, id) in &self.last_layout.as_ref().unwrap().selected {
+ let projected = self.fields.get(id).unwrap();
+ let start = offsets[*index];
+ let end = match &sorted_offsets {
+ None => offsets[*index + 1],
+ Some(sorted) => {
+ let next = sorted.partition_point(|offset| *offset <=
start);
+ *sorted.get(next).ok_or_else(|| Error::DataInvalid {
+ message: "Malformed Variant object
offsets".to_string(),
+ source: None,
+ })?
+ }
+ };
+ let child_pos =
+ layout
+ .data_start
+ .checked_add(start)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Malformed Variant object
offsets".to_string(),
+ source: None,
+ })?;
+ let child_size = validate_value(value, metadata, child_pos, 1)?;
+ if child_size != end - start {
+ return data_invalid("Malformed Variant child size");
+ }
+ let child = VariantRef::new_at(value, metadata, child_pos)?;
+ let numeric = numeric_to_float32(child, &projected.name)?;
+ for output_index in &projected.outputs {
+ output[*output_index] = numeric;
+ }
+ }
+ Ok(())
+ }
+
+ fn prepare_layout(
+ &mut self,
+ value: &[u8],
+ metadata: &[u8],
+ layout: &ObjectLayout,
+ ) -> Result<()> {
+ let ids = &value[layout.id_start..layout.offset_start];
+ if self.last_layout.as_ref().is_some_and(|cached| {
+ cached.field_count == layout.size
+ && cached.id_size == layout.id_size
+ && cached.ids == ids
+ }) {
+ return Ok(());
+ }
+
+ let mut previous_key = None;
+ let mut selected = Vec::with_capacity(self.fields.len());
+ for index in 0..layout.size {
+ let id = read_unsigned(
+ value,
+ layout.id_start + layout.id_size * index,
+ layout.id_size,
+ )?;
+ let key = get_metadata_key_ref(metadata, id)?;
+ if previous_key
+ .is_some_and(|previous| java_string_cmp(previous, key) !=
std::cmp::Ordering::Less)
+ {
+ return data_invalid("Malformed Variant object key order");
+ }
+ previous_key = Some(key);
+ if self.fields.contains_key(&id) {
+ selected.push((index, id));
+ }
+ }
+ self.last_layout = Some(ObjectLayoutProjection {
+ field_count: layout.size,
+ id_size: layout.id_size,
+ ids: ids.to_vec(),
+ selected,
+ });
+ Ok(())
+ }
+}
+
+fn numeric_to_float32(value: VariantRef<'_>, field: &str) ->
Result<Option<f32>> {
+ Ok(match value.kind()? {
+ VariantKind::Null => None,
+ VariantKind::Long => Some(value.get_long()? as f32),
+ VariantKind::Float => Some(value.get_float()?),
+ VariantKind::Double => Some(value.get_double()? as f32),
+ VariantKind::Decimal => {
+ let decimal = value.get_decimal()?;
+ Some((decimal.unscaled as f64 / 10f64.powi(decimal.scale as i32))
as f32)
+ }
+ kind => {
+ return Err(Error::Unsupported {
+ message: format!("Variant field '{field}' has non-numeric type
{kind:?}"),
+ });
+ }
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn handles_non_monotonic_object_offsets() {
+ let variant = GenericVariant::parse_json(r#"{"a":1,"b":2}"#).unwrap();
+ let mut value = variant.value().to_vec();
+ let layout = object_layout(&value, 0).unwrap();
+ let first = read_unsigned(&value, layout.offset_start,
layout.offset_size).unwrap();
+ let second = read_unsigned(
+ &value,
+ layout.offset_start + layout.offset_size,
+ layout.offset_size,
+ )
+ .unwrap();
+ write_le_at(&mut value, layout.offset_start, second,
layout.offset_size);
+ write_le_at(
+ &mut value,
+ layout.offset_start + layout.offset_size,
+ first,
+ layout.offset_size,
+ );
+ validate_payload(&value, variant.metadata()).unwrap();
+
+ let fields = vec!["a".to_string(), "b".to_string(), "a".to_string()];
+ let mut projection = VariantFloat32Projection::new(variant.metadata(),
&fields).unwrap();
+ let mut offsets = Vec::new();
+ let mut output = vec![None; fields.len()];
+ projection
+ .extract_float32(&value, variant.metadata(), &mut offsets, &mut
output)
+ .unwrap();
+ assert_eq!(output, vec![Some(2.0), Some(1.0), Some(2.0)]);
+ }
+}
