[python] Add batch numeric extraction for Variant fields

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."""
+    ...

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(())
 }

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(())
+}

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, [])

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,