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 4bf77696 [python] Push batched Variant extraction into native reads
(#1009)
4bf77696 is described below
commit 4bf7769613e565e393b3b54d60a8a8eb549aef0b
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Oct 4 22:23:21 2026 +0800
[python] Push batched Variant extraction into native reads (#1009)
---
.../python/python/pypaimon_rust/datafusion.pyi | 4 +
bindings/python/src/read.rs | 63 ++++-
bindings/python/src/schema.rs | 4 +
bindings/python/tests/test_read.py | 175 ++++++++++++
.../datafusion/src/variant_pushdown.rs | 8 +-
.../datafusion/tests/variant_pushdown.rs | 42 ++-
crates/paimon/src/arrow/shredding/variant.rs | 302 ++++++++++++++++++++-
crates/paimon/src/arrow/variant.rs | 42 ++-
crates/paimon/src/spec/variant_metadata.rs | 115 ++++++--
crates/paimon/src/table/data_file_reader.rs | 19 +-
crates/paimon/src/table/kv_file_reader.rs | 270 +++++++++++++++++-
crates/paimon/src/variant.rs | 103 ++++++-
crates/paimon/src/variant/numeric.rs | 65 ++++-
13 files changed, 1138 insertions(+), 74 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 5b94ab35..31e4120f 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -28,6 +28,7 @@ InputFieldsLike: TypeAlias = Union[ArrowTypeLike,
Sequence[ArrowTypeLike]]
VolatilityLike: TypeAlias = Union[str, Any]
class DataField:
+ def id(self) -> int: ...
def name(self) -> str: ...
def field_type(self) -> str: ...
def is_nullable(self) -> bool: ...
@@ -87,6 +88,9 @@ class TableRead:
def read(self, splits: Sequence[Split]) -> List[pyarrow.RecordBatch]: ...
class ReadBuilder:
+ def with_read_type(self, read_type_json: str) -> "ReadBuilder":
+ """Set a complete Paimon ROW read type serialized as schema JSON."""
+ ...
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
diff --git a/bindings/python/src/read.rs b/bindings/python/src/read.rs
index 9a25b7fd..53e88384 100644
--- a/bindings/python/src/read.rs
+++ b/bindings/python/src/read.rs
@@ -22,7 +22,7 @@ use std::sync::{Arc, Mutex};
use arrow::pyarrow::ToPyArrow;
use arrow::record_batch::RecordBatch;
use futures::TryStreamExt;
-use paimon::spec::{DataField, DataType, Predicate, RowType};
+use paimon::spec::{is_variant_extraction_row, DataField, DataType, Predicate,
RowType};
use paimon::table::{ArrowRecordBatchStream, DataSplit, IncrementalScanMode,
RowRange, Table};
use paimon_datafusion::runtime::runtime;
use pyo3::exceptions::{PyRuntimeError, PyTypeError, PyValueError};
@@ -36,6 +36,11 @@ 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>>,
+}
+
/// Time-travel selector option names, in the core's resolution priority order.
const TIME_TRAVEL_SELECTORS: [&str; 6] = [
"scan.timestamp-millis",
@@ -73,17 +78,16 @@ fn find_time_travel_selector(opts: &HashMap<String,
String>) -> Option<(&str, &s
/// Apply common scan/read config onto a core ReadBuilder.
fn apply_read_config(
builder: &mut paimon::table::ReadBuilder<'_>,
- projection: &Option<Vec<String>>,
- read_type: &Option<Vec<DataField>>,
+ projection: PyReadProjection<'_>,
limit: Option<usize>,
filter: &Option<Predicate>,
case_sensitive: bool,
) -> PyResult<()> {
builder.with_case_sensitive(case_sensitive);
- if let Some(read_type) = read_type {
+ if let Some(read_type) = projection.read_type {
builder.with_read_type(read_type.clone());
- } else if let Some(projection) = projection {
- let cols: Vec<&str> = projection.iter().map(String::as_str).collect();
+ } else if let Some(columns) = projection.columns {
+ let cols: Vec<&str> = columns.iter().map(String::as_str).collect();
builder.with_projection(&cols).map_err(to_py_err)?;
}
if let Some(limit) = limit {
@@ -293,12 +297,47 @@ impl PyReadBuilder {
#[pymethods]
impl PyReadBuilder {
+ /// Project top-level columns by name.
fn with_projection(mut slf: PyRefMut<'_, Self>, columns: Vec<String>) ->
PyRefMut<'_, Self> {
slf.projection = Some(columns);
slf.read_type = None;
slf
}
+ /// Set the complete Paimon ROW read type, including nested field metadata.
+ /// The JSON representation is the same as RowType in a Paimon schema.
+ fn with_read_type<'py>(
+ mut slf: PyRefMut<'py, Self>,
+ read_type_json: &str,
+ ) -> PyResult<PyRefMut<'py, Self>> {
+ let read_type: RowType = serde_json::from_str(read_type_json)
+ .map_err(|e| PyValueError::new_err(format!("invalid Paimon read
type: {e}")))?;
+ for field in read_type.fields() {
+ let source = slf
+ .table
+ .schema()
+ .fields()
+ .iter()
+ .find(|source| source.id() == field.id());
+ if let (Some(source), DataType::Row(row)) = (source,
field.data_type()) {
+ if matches!(source.data_type(), DataType::Variant(_))
+ && is_variant_extraction_row(row)
+ && row
+ .fields()
+ .iter()
+ .any(|child| !matches!(child.data_type(),
DataType::Float(_)))
+ {
+ return Err(PyValueError::new_err(
+ "Variant extraction target type must be float32",
+ ));
+ }
+ }
+ }
+ slf.read_type = Some(read_type.fields().to_vec());
+ slf.projection = None;
+ 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(
@@ -551,8 +590,10 @@ impl PyTableScan {
let mut builder = self.table.new_read_builder();
apply_read_config(
&mut builder,
- &self.projection,
- &self.read_type,
+ PyReadProjection {
+ columns: &self.projection,
+ read_type: &self.read_type,
+ },
self.limit,
&self.filter,
self.case_sensitive,
@@ -665,8 +706,10 @@ impl PyTableRead {
let mut builder = self.table.new_read_builder();
apply_read_config(
&mut builder,
- &self.projection,
- &self.read_type,
+ PyReadProjection {
+ columns: &self.projection,
+ read_type: &self.read_type,
+ },
self.limit,
&self.filter,
self.case_sensitive,
diff --git a/bindings/python/src/schema.rs b/bindings/python/src/schema.rs
index 3783e3a1..a40eeb06 100644
--- a/bindings/python/src/schema.rs
+++ b/bindings/python/src/schema.rs
@@ -72,6 +72,10 @@ impl PyDataField {
#[pymethods]
impl PyDataField {
+ fn id(&self) -> i32 {
+ self.inner.id()
+ }
+
fn name(&self) -> String {
self.inner.name().to_string()
}
diff --git a/bindings/python/tests/test_read.py
b/bindings/python/tests/test_read.py
index 17721de7..6d72e482 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -35,6 +35,48 @@ def _make_table_with_data(warehouse):
return catalog.get_table("rdb.t")
+def _make_variant_table_with_data(warehouse):
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.vdb")
+ ctx.sql("CREATE TABLE paimon.vdb.t (id INT, payload VARIANT)")
+ ctx.sql(
+ """
+ INSERT INTO paimon.vdb.t
+ SELECT 1, parse_json('{"age":27,"ratio":1.25}')
+ UNION ALL
+ SELECT 2, parse_json('{"age":"32","ratio":"invalid"}')
+ """
+ )
+ return PaimonCatalog({"warehouse": warehouse}).get_table("vdb.t")
+
+
+def _variant_read_type(table, columns, paths, fail_on_error=False):
+ fields = []
+ by_name = {field.name(): field for field in table.schema().fields()}
+ for name in columns:
+ field = by_name[name]
+ data_type = field.field_type()
+ if name == "payload":
+ data_type = {
+ "type": "ROW",
+ "fields": [
+ {
+ "id": index,
+ "name": str(index),
+ "type": "FLOAT",
+ "description": (
+ "__VARIANT_METADATA%s;%s;UTC"
+ % (path, str(fail_on_error).lower())
+ ),
+ }
+ for index, path in enumerate(paths)
+ ],
+ }
+ fields.append({"id": field.id(), "name": name, "type": data_type})
+ return json.dumps({"type": "ROW", "fields": fields})
+
+
def test_read_builder_chain_exists():
with tempfile.TemporaryDirectory() as warehouse:
table = _make_table_with_data(warehouse)
@@ -59,6 +101,139 @@ def test_with_projection():
assert plan is not None
+def test_with_read_type_plain_projection():
+ with tempfile.TemporaryDirectory() as warehouse:
+ table = _make_table_with_data(warehouse)
+ field = next(field for field in table.schema().fields() if
field.name() == "id")
+ read_type = json.dumps({
+ "type": "ROW",
+ "fields": [{
+ "id": field.id(), "name": field.name(),
+ "type": field.field_type(),
+ }],
+ })
+ builder = table.new_read_builder().with_read_type(read_type)
+ plan = builder.new_scan().plan()
+ result = pa.Table.from_batches(builder.new_read().read(plan.splits()))
+ assert result.schema.names == ["id"]
+ assert sorted(result.column("id").to_pylist()) == [1, 2, 3]
+
+
+def test_with_read_type_extracts_variant_fields():
+ with tempfile.TemporaryDirectory() as warehouse:
+ table = _make_variant_table_with_data(warehouse)
+ builder = table.new_read_builder().with_read_type(_variant_read_type(
+ table, ["id", "payload"], ["$.ratio", "$.age", "$.missing"]))
+ plan = builder.new_scan().plan()
+ result =
pa.Table.from_batches(builder.new_read().read(plan.splits())).sort_by("id")
+
+ assert result.schema.names == ["id", "payload"]
+ assert result.field("payload").type == pa.struct(
+ [
+ pa.field("0", pa.float32()),
+ pa.field("1", pa.float32()),
+ pa.field("2", pa.float32()),
+ ]
+ )
+ assert result.column("payload").combine_chunks().to_pylist() == [
+ {"0": 1.25, "1": 27.0, "2": None},
+ {"0": None, "1": 32.0, "2": None},
+ ]
+
+
+def test_with_read_type_variant_fields_honors_fail_on_error():
+ 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"], ["$.ratio"], fail_on_error=True))
+ plan = builder.new_scan().plan()
+ with pytest.raises(ValueError, match="Cannot cast Variant value to"):
+ builder.new_read().read(plan.splits())
+
+
+def test_with_read_type_variant_float_whitespace_matches_java():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.vdb")
+ ctx.sql("CREATE TABLE paimon.vdb.t (id INT, payload VARIANT)")
+ ascii_value = json.dumps({"x": " 1.5 "})
+ nbsp_value = json.dumps({"x": "\u00a01.5\u00a0"}, ensure_ascii=False)
+ ctx.sql(
+ "INSERT INTO paimon.vdb.t "
+ "SELECT 1, parse_json('%s') UNION ALL "
+ "SELECT 2, parse_json('%s')" % (ascii_value, nbsp_value)
+ )
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("vdb.t")
+
+ builder = table.new_read_builder().with_read_type(_variant_read_type(
+ table, ["id", "payload"], ["$.x"]))
+ plan = builder.new_scan().plan()
+ result = pa.Table.from_batches(
+ builder.new_read().read(plan.splits())
+ ).sort_by("id")
+ assert result.column("payload").combine_chunks().to_pylist() == [
+ {"0": 1.5}, {"0": None},
+ ]
+
+ strict = table.new_read_builder().with_read_type(_variant_read_type(
+ table, ["payload"], ["$.x"], fail_on_error=True))
+ with pytest.raises(ValueError, match="Cannot cast Variant value to"):
+ strict.new_read().read(strict.new_scan().plan().splits())
+
+
+def test_with_read_type_variant_fields_keeps_explicit_null():
+ with tempfile.TemporaryDirectory() as warehouse:
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": warehouse})
+ ctx.sql("CREATE SCHEMA paimon.vdb")
+ ctx.sql("CREATE TABLE paimon.vdb.t (id INT, payload VARIANT)")
+ ctx.sql("INSERT INTO paimon.vdb.t SELECT 1,
parse_json('{\"x\":null}')")
+ table = PaimonCatalog({"warehouse": warehouse}).get_table("vdb.t")
+ builder = table.new_read_builder().with_read_type(_variant_read_type(
+ table, ["payload"], ["$.x"], fail_on_error=True))
+ plan = builder.new_scan().plan()
+ result = pa.Table.from_batches(builder.new_read().read(plan.splits()))
+ assert result.column("payload").combine_chunks().to_pylist() == [{"0":
None}]
+
+
+def test_with_read_type_rejects_invalid_schema():
+ with tempfile.TemporaryDirectory() as warehouse:
+ table = _make_variant_table_with_data(warehouse)
+ with pytest.raises(ValueError, match="invalid Paimon read type"):
+ table.new_read_builder().with_read_type("not JSON")
+
+
+def test_with_read_type_rejects_unsupported_variant_target():
+ with tempfile.TemporaryDirectory() as warehouse:
+ table = _make_variant_table_with_data(warehouse)
+ read_type = json.loads(_variant_read_type(table, ["payload"],
["$.age"]))
+ read_type["fields"][0]["type"]["fields"][0]["type"] = "INT"
+ with pytest.raises(ValueError, match="must be float32"):
+ table.new_read_builder().with_read_type(json.dumps(read_type))
+
+
+def test_with_nested_projection_replaces_read_type():
+ 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"]])
+ plan = builder.new_scan().plan()
+ result = pa.Table.from_batches(builder.new_read().read(plan.splits()))
+ assert result.schema.names == ["id"]
+ assert sorted(result.column("id").to_pylist()) == [1, 2]
+
+
+def test_with_projection_replaces_read_type():
+ 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_projection(["id"])
+ plan = builder.new_scan().plan()
+ result = pa.Table.from_batches(builder.new_read().read(plan.splits()))
+ assert result.schema.names == ["id"]
+
+
def test_with_limit():
with tempfile.TemporaryDirectory() as warehouse:
table = _make_table_with_data(warehouse)
diff --git a/crates/integrations/datafusion/src/variant_pushdown.rs
b/crates/integrations/datafusion/src/variant_pushdown.rs
index ba3c7cb1..c0a79bf1 100644
--- a/crates/integrations/datafusion/src/variant_pushdown.rs
+++ b/crates/integrations/datafusion/src/variant_pushdown.rs
@@ -347,7 +347,8 @@ fn build_projection_rewrite(
"UTC".to_string(),
)
}),
- );
+ )
+ .map_err(to_datafusion_error)?;
read_type.push(
DataField::new(
field.id(),
@@ -541,6 +542,11 @@ fn parse_variant_get(expr: &Expr) ->
DFResult<VariantGetParse> {
let Some(type_name) = func.args.get(2).and_then(string_literal) else {
return Ok(VariantGetParse::FullVariant(column.clone()));
};
+ // Java's Variant field metadata uses ';' as a delimiter. Keep the full
+ // column so the original expression can handle paths containing it.
+ if path.contains(';') {
+ return Ok(VariantGetParse::FullVariant(column.clone()));
+ }
let Some(data_type) = paimon_type_for_variant_get(&type_name)? else {
return Ok(VariantGetParse::FullVariant(column.clone()));
};
diff --git a/crates/integrations/datafusion/tests/variant_pushdown.rs
b/crates/integrations/datafusion/tests/variant_pushdown.rs
index 2fb60653..bb8b50e7 100644
--- a/crates/integrations/datafusion/tests/variant_pushdown.rs
+++ b/crates/integrations/datafusion/tests/variant_pushdown.rs
@@ -334,7 +334,7 @@ async fn
try_variant_get_invalid_path_pushdown_returns_null() {
}
#[tokio::test]
-async fn variant_get_path_with_semicolon_pushes_extraction() {
+async fn variant_get_path_with_semicolon_falls_back() {
let (_tmp, sql_context) = setup_shredded_variant_table_with_rows().await;
let sql = r#"
SELECT variant_get(payload, '$.a;b', 'int') AS value
@@ -346,8 +346,8 @@ async fn
variant_get_path_with_semicolon_pushes_extraction() {
let plan = df.create_physical_plan().await.unwrap();
let plan_text = displayable(plan.as_ref()).indent(true).to_string();
assert!(
- plan_text.contains("PushedVariants=[payload=[$.a;b]]"),
- "plan should push semicolon-path variant extraction, got:\n{plan_text}"
+ !plan_text.contains("PushedVariants="),
+ "plan should not push an unencodable semicolon path, got:\n{plan_text}"
);
let batches = sql_context.sql(sql).await.unwrap().collect().await.unwrap();
@@ -360,3 +360,39 @@ async fn
variant_get_path_with_semicolon_pushes_extraction() {
.unwrap();
assert_eq!(values.values(), &[11, 22]);
}
+
+#[tokio::test]
+async fn variant_get_mixed_paths_with_semicolon_keep_full_column() {
+ let (_tmp, sql_context) = setup_shredded_variant_table_with_rows().await;
+ let sql = r#"
+ SELECT variant_get(payload, '$.age', 'int') AS age,
+ variant_get(payload, '$.a;b', 'int') AS value
+ FROM paimon.test_db.t
+ ORDER BY id
+ "#;
+
+ let df = sql_context.sql(sql).await.unwrap();
+ let plan = df.create_physical_plan().await.unwrap();
+ let plan_text = displayable(plan.as_ref()).indent(true).to_string();
+ assert!(
+ !plan_text.contains("PushedVariants="),
+ "plan should keep the full Variant column for mixed paths,
got:\n{plan_text}"
+ );
+
+ let batches = sql_context.sql(sql).await.unwrap().collect().await.unwrap();
+ assert_eq!(batches.len(), 1);
+ let ages = batches[0]
+ .column_by_name("age")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batches[0]
+ .column_by_name("value")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ assert_eq!(ages.values(), &[27, 32]);
+ assert_eq!(values.values(), &[11, 22]);
+}
diff --git a/crates/paimon/src/arrow/shredding/variant.rs
b/crates/paimon/src/arrow/shredding/variant.rs
index e71774ce..9f75d406 100644
--- a/crates/paimon/src/arrow/shredding/variant.rs
+++ b/crates/paimon/src/arrow/shredding/variant.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use super::super::variant::{extract_numeric_fields, NumericFieldMode};
use super::{
option_bool, option_f64, option_usize, FieldMetadata, ShreddingWritePlan,
ShreddingWritePlanFactory,
@@ -27,9 +28,10 @@ use crate::spec::{
};
use crate::variant::{
build_variant_schema, cast_shredded, cast_variant_to_shredded_value,
- infer_variant_shredding_schema, rebuild_shredded, variant_shredding_type,
GenericVariant,
- ShreddedRow, ShreddedValue, VariantShreddingInferConfig,
VariantShreddingInferenceSession,
- VARIANT_METADATA_FIELD_NAME, VARIANT_TYPED_VALUE_FIELD_NAME,
VARIANT_VALUE_FIELD_NAME,
+ infer_variant_shredding_schema, rebuild_shredded, top_level_path_key,
variant_shredding_type,
+ GenericVariant, ShreddedRow, ShreddedValue, VariantShreddingInferConfig,
+ VariantShreddingInferenceSession, VARIANT_METADATA_FIELD_NAME,
VARIANT_TYPED_VALUE_FIELD_NAME,
+ VARIANT_VALUE_FIELD_NAME,
};
use crate::{Error, Result};
use arrow_array::{
@@ -869,7 +871,11 @@ fn assemble_array_to_logical(
}
}
-fn assemble_variant_extraction_array(array: &dyn Array, row_type: &RowType) ->
Result<ArrayRef> {
+/// Extract projected fields from a logical Variant column after PK merge.
+pub(crate) fn assemble_variant_extraction_array(
+ array: &dyn Array,
+ row_type: &RowType,
+) -> Result<ArrayRef> {
let input = array
.as_any()
.downcast_ref::<StructArray>()
@@ -891,6 +897,10 @@ fn assemble_variant_extraction_array(array: &dyn Array,
row_type: &RowType) -> R
})
.collect::<Result<Vec<_>>>()?;
+ if let Some(projected) = assemble_plain_variant_projection(input, fields,
&metadata)? {
+ return Ok(projected);
+ }
+
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() {
@@ -953,6 +963,78 @@ fn assemble_variant_extraction_array(array: &dyn Array,
row_type: &RowType) -> R
))
}
+fn assemble_plain_variant_projection(
+ input: &StructArray,
+ fields: &[DataField],
+ metadata: &[crate::spec::VariantFieldMetadata],
+) -> Result<Option<ArrayRef>> {
+ if !is_plain_variant_array(input)
+ || fields
+ .iter()
+ .any(|field| !matches!(field.data_type(), DataType::Float(_)))
+ {
+ return Ok(None);
+ }
+
+ let mut keys = Vec::with_capacity(fields.len());
+ let mut fail_on_error = Vec::with_capacity(fields.len());
+ for field_metadata in metadata {
+ let Ok(Some(key)) = top_level_path_key(field_metadata.path()) else {
+ return Ok(None);
+ };
+ keys.push(key);
+ fail_on_error.push(field_metadata.fail_on_error());
+ }
+
+ let projected = extract_numeric_fields(
+ input,
+ &keys,
+ NumericFieldMode::Cast {
+ fail_on_error: &fail_on_error,
+ },
+ )?;
+ let values = projected
+ .values()
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .expect("float32 Variant projection must contain Float32 values");
+ let width = fields.len();
+ let columns = (0..width)
+ .map(|field_index| {
+ Arc::new(Float32Array::from_iter((0..projected.len()).map(
+ |row_index| {
+ let value_index = row_index * width + field_index;
+ if values.is_null(value_index) {
+ None
+ } else {
+ Some(values.value(value_index))
+ }
+ },
+ ))) as ArrayRef
+ })
+ .collect::<Vec<_>>();
+ let arrow_fields: Fields = fields
+ .iter()
+ .map(|field| {
+ ArrowField::new(
+ field.name(),
+ ArrowDataType::Float32,
+ field.data_type().is_nullable(),
+ )
+ })
+ .collect::<Vec<_>>()
+ .into();
+
+ Ok(Some(Arc::new(
+ StructArray::try_new(arrow_fields, columns,
projected.nulls().cloned()).map_err(|e| {
+ Error::UnexpectedError {
+ message: format!("Failed to build Variant float32 projection:
{e}"),
+ source: Some(Box::new(e)),
+ }
+ })?,
+ )))
+}
+
fn variant_from_storage_row(input: &StructArray, row: usize) ->
Result<Option<GenericVariant>> {
if input.is_null(row) {
return Ok(None);
@@ -1507,7 +1589,9 @@ fn null_buffer(validities: Vec<bool>) -> NullBuffer {
mod tests {
use super::*;
use crate::arrow::variant_arrow_type;
- use crate::spec::{variant_extraction_row, BlobType, IntType, VarCharType,
VariantType};
+ use crate::spec::{
+ variant_extraction_row, BlobType, FloatType, IntType, VarCharType,
VariantType,
+ };
#[test]
fn factory_owns_activation_and_configured_schema_precedence() {
@@ -1765,7 +1849,8 @@ mod tests {
"UTC".to_string(),
),
],
- );
+ )
+ .unwrap();
let extracted = assemble_array_to_logical(array.as_ref(),
&DataType::Row(row_type))
.unwrap()
@@ -1790,6 +1875,208 @@ mod tests {
assert!(names.is_null(1));
}
+ #[test]
+ fn variant_extraction_row_batches_top_level_float32_fields() {
+ let variants = vec![
+ Some(
+ GenericVariant::parse_json(
+ r#"{"age":27,"ratio":1.25,"text":"3.5","enabled":true}"#,
+ )
+ .unwrap(),
+ ),
+ Some(
+ GenericVariant::parse_json(
+ r#"{"age":32,"ratio":null,"text":"bad","enabled":false}"#,
+ )
+ .unwrap(),
+ ),
+ None,
+ ];
+ let input = variant_array(variants).unwrap();
+ let row_type = variant_extraction_row(
+ true,
+ [
+ ("$.ratio", false),
+ ("$.age", false),
+ ("$.text", false),
+ ("$.enabled", false),
+ ("$.missing", false),
+ ]
+ .into_iter()
+ .map(|(path, fail_on_error)| {
+ (
+ DataType::Float(FloatType::new()),
+ path.to_string(),
+ fail_on_error,
+ "UTC".to_string(),
+ )
+ }),
+ )
+ .unwrap();
+
+ let projected = assemble_array_to_logical(input.as_ref(),
&DataType::Row(row_type))
+ .unwrap()
+ .expect("projected");
+ let projected =
projected.as_any().downcast_ref::<StructArray>().unwrap();
+ assert_eq!(projected.len(), 3);
+ assert!(projected.is_valid(0));
+ assert!(projected.is_valid(1));
+ assert!(projected.is_null(2));
+ let values = (0..5)
+ .map(|index| {
+ projected
+ .column(index)
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap()
+ .iter()
+ .collect::<Vec<_>>()
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(values[0], vec![Some(1.25), None, None]);
+ assert_eq!(values[1], vec![Some(27.0), Some(32.0), None]);
+ assert_eq!(values[2], vec![Some(3.5), None, None]);
+ assert_eq!(values[3], vec![Some(1.0), Some(0.0), None]);
+ assert_eq!(values[4], vec![None, None, None]);
+ }
+
+ #[test]
+ fn variant_extraction_row_float32_fast_path_honors_fail_on_error() {
+ let variant =
GenericVariant::parse_json(r#"{"value":"bad"}"#).unwrap();
+ let input = variant_array(vec![Some(variant)]).unwrap();
+ let row_type = variant_extraction_row(
+ true,
+ [(
+ DataType::Float(FloatType::new()),
+ "$.value".to_string(),
+ true,
+ "UTC".to_string(),
+ )],
+ )
+ .unwrap();
+
+ let error =
+ assemble_array_to_logical(input.as_ref(),
&DataType::Row(row_type)).unwrap_err();
+ assert!(matches!(error, Error::DataInvalid { ref message, .. }
+ if message.contains("Cannot cast Variant value to")));
+ }
+
+ #[test]
+ fn variant_extraction_row_preserves_null_with_fail_on_error() {
+ let variant = GenericVariant::parse_json(r#"{"value":null}"#).unwrap();
+ let input = variant_array(vec![Some(variant)]).unwrap();
+ let row_type = variant_extraction_row(
+ true,
+ [(
+ DataType::Float(FloatType::new()),
+ "$.value".to_string(),
+ true,
+ "UTC".to_string(),
+ )],
+ )
+ .unwrap();
+
+ let projected = assemble_array_to_logical(input.as_ref(),
&DataType::Row(row_type))
+ .unwrap()
+ .unwrap();
+ let projected =
projected.as_any().downcast_ref::<StructArray>().unwrap();
+ assert!(projected.is_valid(0));
+ assert!(projected.column(0).is_null(0));
+ }
+
+ #[test]
+ fn invalid_try_path_returns_null_for_plain_and_shredded_variant() {
+ let logical_fields = vec![DataField::new(
+ 1,
+ "v".to_string(),
+ DataType::Variant(VariantType::new()),
+ )];
+ let options = HashMap::from([(
+ "variant.shreddingSchema".to_string(),
+
r#"{"type":"ROW","fields":[{"name":"v","type":{"type":"ROW","fields":[{"name":"age","type":"INT"}]}}]}"#.to_string(),
+ )]);
+ let physical_fields =
configured_variant_shredding_fields(&logical_fields, &options)
+ .unwrap()
+ .unwrap();
+ let variant = GenericVariant::parse_json(r#"{"age":27}"#).unwrap();
+ let plain = variant_array_for_test(&[variant]);
+ let row_type = variant_extraction_row(
+ true,
+ [(
+ DataType::Float(FloatType::new()),
+ "$.".to_string(),
+ false,
+ "UTC".to_string(),
+ )],
+ )
+ .unwrap();
+
+ let plain_result =
+ assemble_array_to_logical(plain.as_ref(),
&DataType::Row(row_type.clone()))
+ .unwrap()
+ .unwrap();
+ let plain_result =
plain_result.as_any().downcast_ref::<StructArray>().unwrap();
+ assert!(plain_result.column(0).is_null(0));
+
+ let strict_row_type = variant_extraction_row(
+ true,
+ [(
+ DataType::Float(FloatType::new()),
+ "$.".to_string(),
+ true,
+ "UTC".to_string(),
+ )],
+ )
+ .unwrap();
+ assert!(
+ assemble_array_to_logical(plain.as_ref(),
&DataType::Row(strict_row_type)).is_err()
+ );
+
+ let batch = RecordBatch::try_new(
+ build_target_arrow_schema(&logical_fields).unwrap(),
+ vec![plain],
+ )
+ .unwrap();
+ let physical =
+ batch_to_shredded_physical(&batch, &logical_fields,
&physical_fields).unwrap();
+ let read_fields = vec![DataField::new(1, "v".to_string(),
DataType::Row(row_type))];
+ let shredded_result = assemble_shredded_variant_batch(physical,
&read_fields).unwrap();
+ let shredded_result = shredded_result
+ .column_by_name("v")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ assert!(shredded_result.column(0).is_null(0));
+ }
+
+ #[test]
+ fn nested_float32_variant_extraction_keeps_general_path_semantics() {
+ let variant =
GenericVariant::parse_json(r#"{"pose":{"x":"1.5"}}"#).unwrap();
+ let input = variant_array(vec![Some(variant)]).unwrap();
+ let row_type = variant_extraction_row(
+ true,
+ [(
+ DataType::Float(FloatType::new()),
+ "$.pose.x".to_string(),
+ true,
+ "UTC".to_string(),
+ )],
+ )
+ .unwrap();
+
+ let projected = assemble_array_to_logical(input.as_ref(),
&DataType::Row(row_type))
+ .unwrap()
+ .expect("projected");
+ let projected =
projected.as_any().downcast_ref::<StructArray>().unwrap();
+ let values = projected
+ .column(0)
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap();
+ assert_eq!(values.value(0), 1.5);
+ }
+
#[test]
fn variant_extraction_row_reads_shredded_variant_array() {
let logical_fields = vec![DataField::new(
@@ -1831,7 +2118,8 @@ mod tests {
"UTC".to_string(),
),
],
- );
+ )
+ .unwrap();
let read_fields = vec![DataField::new(
1,
"v".to_string(),
diff --git a/crates/paimon/src/arrow/variant.rs
b/crates/paimon/src/arrow/variant.rs
index 9534d86a..07619aa0 100644
--- a/crates/paimon/src/arrow/variant.rs
+++ b/crates/paimon/src/arrow/variant.rs
@@ -33,10 +33,29 @@ use crate::{Error, Result};
pub fn variant_get_numeric_fields(
column: &StructArray,
fields: &[String],
+) -> Result<FixedSizeListArray> {
+ extract_numeric_fields(column, fields, NumericFieldMode::Numeric)
+}
+
+#[derive(Clone, Copy)]
+pub(crate) enum NumericFieldMode<'a> {
+ Numeric,
+ Cast { fail_on_error: &'a [bool] },
+}
+
+pub(crate) fn extract_numeric_fields(
+ column: &StructArray,
+ fields: &[String],
+ mode: NumericFieldMode<'_>,
) -> Result<FixedSizeListArray> {
if fields.is_empty() {
return data_invalid("Variant numeric field list must not be empty");
}
+ if let NumericFieldMode::Cast { fail_on_error } = mode {
+ if fail_on_error.len() != fields.len() {
+ return data_invalid("Variant float32 projection width mismatch");
+ }
+ }
let width = i32::try_from(fields.len()).map_err(|_|
Error::ResourceExhausted {
message: "Variant numeric field count exceeds Arrow
limits".to_string(),
@@ -78,12 +97,23 @@ pub fn variant_get_numeric_fields(
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,
- )?;
+ match mode {
+ NumericFieldMode::Cast { fail_on_error } => {
+ projection.as_mut().unwrap().extract_float32_cast(
+ value_column.value(row),
+ metadata,
+ fail_on_error,
+ &mut offsets,
+ &mut extracted,
+ )?
+ }
+ NumericFieldMode::Numeric =>
projection.as_mut().unwrap().extract_float32(
+ value_column.value(row),
+ metadata,
+ &mut offsets,
+ &mut extracted,
+ )?,
+ }
for value in &extracted {
builder.values().append_option(*value);
}
diff --git a/crates/paimon/src/spec/variant_metadata.rs
b/crates/paimon/src/spec/variant_metadata.rs
index b172931d..02bef4b5 100644
--- a/crates/paimon/src/spec/variant_metadata.rs
+++ b/crates/paimon/src/spec/variant_metadata.rs
@@ -58,11 +58,21 @@ impl VariantFieldMetadata {
}
}
-pub fn build_variant_metadata(path: &str, fail_on_error: bool, time_zone_id:
&str) -> String {
- let metadata = VariantFieldMetadata::new(path, fail_on_error,
time_zone_id);
- let json = serde_json::to_string(&metadata)
- .expect("Variant metadata serialization should not fail for string and
bool fields");
- format!("{VARIANT_METADATA_KEY}{json}")
+pub fn build_variant_metadata(
+ path: &str,
+ fail_on_error: bool,
+ time_zone_id: &str,
+) -> Result<String> {
+ if path.contains(VARIANT_METADATA_DELIMITER) {
+ return Err(Error::ConfigInvalid {
+ message: format!(
+ "Variant extraction path must not contain
'{VARIANT_METADATA_DELIMITER}': {path}"
+ ),
+ });
+ }
+ Ok(format!(
+
"{VARIANT_METADATA_KEY}{path}{VARIANT_METADATA_DELIMITER}{fail_on_error}{VARIANT_METADATA_DELIMITER}{time_zone_id}"
+ ))
}
pub fn parse_variant_metadata(description: &str) ->
Result<VariantFieldMetadata> {
@@ -82,10 +92,10 @@ pub fn parse_variant_metadata(description: &str) ->
Result<VariantFieldMetadata>
return Ok(metadata);
}
- parse_legacy_variant_metadata(raw)
+ parse_java_variant_metadata(raw)
}
-fn parse_legacy_variant_metadata(raw: &str) -> Result<VariantFieldMetadata> {
+fn parse_java_variant_metadata(raw: &str) -> Result<VariantFieldMetadata> {
let mut parts = raw.split(VARIANT_METADATA_DELIMITER);
let path = parts.next().unwrap_or_default();
let fail_on_error = parts.next().ok_or_else(|| Error::DataInvalid {
@@ -147,27 +157,33 @@ pub fn is_variant_extraction_row(row_type: &RowType) ->
bool {
pub fn variant_extraction_row(
nullable: bool,
extractions: impl IntoIterator<Item = (DataType, String, bool, String)>,
-) -> RowType {
+) -> Result<RowType> {
let fields = extractions
.into_iter()
.enumerate()
.map(|(idx, (data_type, path, fail_on_error, time_zone_id))| {
- DataField::new(idx as i32, idx.to_string(),
data_type).with_description(Some(
- build_variant_metadata(&path, fail_on_error, &time_zone_id),
- ))
+ Ok(
+ DataField::new(idx as i32, idx.to_string(),
data_type).with_description(Some(
+ build_variant_metadata(&path, fail_on_error,
&time_zone_id)?,
+ )),
+ )
})
- .collect();
- RowType::with_nullable(nullable, fields)
+ .collect::<Result<Vec<_>>>()?;
+ Ok(RowType::with_nullable(nullable, fields))
}
#[cfg(test)]
mod tests {
use super::*;
- use crate::spec::{DataType, IntType, VarCharType};
+ use crate::spec::{DataType, FloatType, IntType, VarCharType};
+
+ // Serialized by Java RowType.serializeJson with VariantMetadataUtils
descriptions.
+ const JAVA_READ_TYPE_JSON: &str =
r#"{"type":"ROW","fields":[{"id":7,"name":"payload","type":{"type":"ROW","fields":[{"id":0,"name":"0","type":"FLOAT","description":"__VARIANT_METADATA$.x;false;Asia/Shanghai"},{"id":1,"name":"1","type":"FLOAT","description":"__VARIANT_METADATA$.y;true;UTC"}]}}]}"#;
#[test]
fn parses_variant_metadata_description() {
- let description = build_variant_metadata("$.a", false, "UTC");
+ let description = build_variant_metadata("$.a", false, "UTC").unwrap();
+ assert_eq!(description, "__VARIANT_METADATA$.a;false;UTC");
let metadata = parse_variant_metadata(&description).unwrap();
assert_eq!(metadata.path(), "$.a");
assert!(!metadata.fail_on_error());
@@ -175,16 +191,25 @@ mod tests {
}
#[test]
- fn parses_variant_metadata_description_with_delimiters() {
- let description = build_variant_metadata("$.a;b", true, "UTC;8");
- let metadata = parse_variant_metadata(&description).unwrap();
+ fn parses_previous_json_variant_metadata_description() {
+ let description =
+
r#"__VARIANT_METADATA{"path":"$.a;b","failOnError":true,"timeZoneId":"UTC"}"#;
+ let metadata = parse_variant_metadata(description).unwrap();
assert_eq!(metadata.path(), "$.a;b");
assert!(metadata.fail_on_error());
- assert_eq!(metadata.time_zone_id(), "UTC;8");
+ assert_eq!(metadata.time_zone_id(), "UTC");
+ }
+
+ #[test]
+ fn rejects_path_with_java_delimiter() {
+ assert!(build_variant_metadata("$.a;b", true, "UTC")
+ .unwrap_err()
+ .to_string()
+ .contains("must not contain ';'"));
}
#[test]
- fn parses_legacy_variant_metadata_description() {
+ fn parses_java_variant_metadata_description() {
let description = format!("{VARIANT_METADATA_KEY}$.a;false;UTC");
let metadata = parse_variant_metadata(&description).unwrap();
assert_eq!(metadata.path(), "$.a");
@@ -192,6 +217,53 @@ mod tests {
assert_eq!(metadata.time_zone_id(), "UTC");
}
+ #[test]
+ fn java_read_type_json_round_trips_through_rust() {
+ let java_row: RowType =
serde_json::from_str(JAVA_READ_TYPE_JSON).unwrap();
+ let DataType::Row(variant_row) = java_row.fields()[0].data_type() else
{
+ panic!("expected Variant extraction row");
+ };
+ let metadata = variant_row
+ .fields()
+ .iter()
+ .map(|field|
parse_variant_metadata(field.description().unwrap()).unwrap())
+ .collect::<Vec<_>>();
+ assert_eq!(
+ metadata[0],
+ VariantFieldMetadata::new("$.x", false, "Asia/Shanghai")
+ );
+ assert_eq!(metadata[1], VariantFieldMetadata::new("$.y", true, "UTC"));
+
+ let rust_row = RowType::new(vec![DataField::new(
+ 7,
+ "payload".to_string(),
+ DataType::Row(
+ variant_extraction_row(
+ true,
+ [
+ (
+ DataType::Float(FloatType::new()),
+ "$.x".to_string(),
+ false,
+ "Asia/Shanghai".to_string(),
+ ),
+ (
+ DataType::Float(FloatType::new()),
+ "$.y".to_string(),
+ true,
+ "UTC".to_string(),
+ ),
+ ],
+ )
+ .unwrap(),
+ ),
+ )]);
+ assert_eq!(
+ serde_json::to_value(rust_row).unwrap(),
+
serde_json::from_str::<serde_json::Value>(JAVA_READ_TYPE_JSON).unwrap()
+ );
+ }
+
#[test]
fn identifies_variant_extraction_row_type() {
let row = variant_extraction_row(
@@ -210,7 +282,8 @@ mod tests {
"UTC".to_string(),
),
],
- );
+ )
+ .unwrap();
assert!(is_variant_extraction_row(&row));
assert!(is_variant_extraction_row_type(&DataType::Row(row)));
}
diff --git a/crates/paimon/src/table/data_file_reader.rs
b/crates/paimon/src/table/data_file_reader.rs
index 3347bb12..8d9c5eb0 100644
--- a/crates/paimon/src/table/data_file_reader.rs
+++ b/crates/paimon/src/table/data_file_reader.rs
@@ -1524,15 +1524,18 @@ mod row_tests {
)]));
let physical_type = variant_shredding_type(&configured).unwrap();
let data_field = field(1, "v", physical_type);
- let extraction_type = DataType::Row(variant_extraction_row(
- true,
- vec![(
- DataType::Int(IntType::new()),
- "$.age".to_string(),
+ let extraction_type = DataType::Row(
+ variant_extraction_row(
true,
- "UTC".to_string(),
- )],
- ));
+ vec![(
+ DataType::Int(IntType::new()),
+ "$.age".to_string(),
+ true,
+ "UTC".to_string(),
+ )],
+ )
+ .unwrap(),
+ );
let expected_field = field(1, "v", extraction_type.clone());
let read_fields = read_data_fields(&[data_field], &[expected_field],
false).unwrap();
diff --git a/crates/paimon/src/table/kv_file_reader.rs
b/crates/paimon/src/table/kv_file_reader.rs
index 764d99cb..e36ed774 100644
--- a/crates/paimon/src/table/kv_file_reader.rs
+++ b/crates/paimon/src/table/kv_file_reader.rs
@@ -31,15 +31,17 @@ use super::sort_merge::{
PartialUpdateMergeFunction, SortMergeReaderBuilder,
};
use crate::arrow::format::MosaicPrefetchOptions;
+use crate::arrow::shredding::variant::assemble_variant_extraction_array;
use crate::arrow::{build_target_arrow_schema, ReadBudget};
use crate::deletion_vector::DeletionVectorFactory;
use crate::file_index::evaluator::evaluate_file_index;
use crate::file_index::file_index_result::FileIndexResult;
use crate::io::FileIO;
use crate::spec::{
- 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,
+ 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,
+ VALUE_KIND_FIELD_NAME,
};
use crate::table::schema_manager::SchemaManager;
use crate::table::ArrowRecordBatchStream;
@@ -392,6 +394,33 @@ impl KeyValueFileReader {
.map(|s| s.as_str())
.collect();
+ // 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 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
+ .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,
+ })
+ })
+ .collect::<crate::Result<_>>()?;
+
// Collect key fields from table schema.
let key_fields: Vec<DataField> = self
.config
@@ -423,10 +452,8 @@ impl KeyValueFileReader {
// + any sequence fields not already included. Physical
system
// fields are already the first two columns of every KV file.
let read_type_names: std::collections::HashSet<&str> =
- self.config.read_type.iter().map(|f| f.name()).collect();
- let mut user_fields: Vec<DataField> = self
- .config
- .read_type
+ merge_read_type.iter().map(|f| f.name()).collect();
+ let mut user_fields: Vec<DataField> = merge_read_type
.iter()
.filter(|field| !matches!(field.id(), SEQUENCE_NUMBER_FIELD_ID |
VALUE_KIND_FIELD_ID))
.cloned()
@@ -545,9 +572,11 @@ impl KeyValueFileReader {
reorder_map[out_idx] = key_pos;
} else {
// Find position in value_fields
+ // The internal field may use the current table name while the
+ // requested output retains a historical name for the same ID.
let val_pos = value_fields
.iter()
- .position(|vf| vf.name() == field.name())
+ .position(|vf| vf.id() == field.id())
.unwrap();
reorder_map[out_idx] = num_keys + val_pos;
}
@@ -781,8 +810,18 @@ impl KeyValueFileReader {
// Reorder columns from [keys..., values...] to
read_type order.
let columns: Vec<_> = reorder_map
.iter()
- .map(|&src| batch.column(src).clone())
- .collect();
+ .enumerate()
+ .map(|(out_idx, &src)| {
+ let column = batch.column(src);
+ match config.read_type[out_idx].data_type() {
+ PaimonDataType::Row(row) if
is_variant_extraction_row(row) =>
+ {
+
assemble_variant_extraction_array(column.as_ref(), row)
+ }
+ _ => Ok(column.clone()),
+ }
+ })
+ .collect::<crate::Result<Vec<_>>>()?;
// An explicit row count keeps empty projections
working
// (e.g. COUNT(*) reads no columns).
let options =
@@ -808,19 +847,22 @@ impl KeyValueFileReader {
#[cfg(test)]
mod tests {
use super::*;
- use crate::arrow::build_target_arrow_schema;
use crate::arrow::format::create_format_writer;
+ use crate::arrow::{build_target_arrow_schema, variant_arrow_type};
use crate::catalog::Identifier;
use crate::deletion_vector::DeletionVector;
use crate::io::FileIOBuilder;
use crate::spec::{
- stats::BinaryTableStats, BinaryRow, DataFileMeta, DataType, Datum,
IntType,
- PredicateBuilder, Schema, TableSchema, VarCharType,
+ stats::BinaryTableStats, variant_extraction_row, BinaryRow,
DataFileMeta, DataType, Datum,
+ FloatType, IntType, PredicateBuilder, Schema, TableSchema,
VarCharType, VariantType,
};
use crate::table::source::{DataSplitBuilder, DeletionFile};
use crate::table::table_commit::TableCommit;
use crate::table::{Table, TableWrite};
- use arrow_array::{Array, Int32Array, Int64Array, Int8Array, StringArray};
+ use arrow_array::{
+ Array, BinaryArray, Float32Array, Int32Array, Int64Array, Int8Array,
StringArray,
+ StructArray,
+ };
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema
as ArrowSchema};
use bytes::Bytes;
use futures::TryStreamExt;
@@ -831,6 +873,206 @@ mod tests {
use std::collections::HashMap;
use std::sync::Arc;
+ #[tokio::test]
+ async fn kv_variant_projection_casts_only_visible_rows() {
+ let file_io = test_file_io();
+ let table_path = "memory:/kv_variant_projection_visible_rows";
+ setup_dirs(&file_io, table_path).await;
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("payload", DataType::Variant(VariantType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .build()
+ .unwrap();
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "kv_variant_projection_visible_rows"),
+ table_path.to_string(),
+ TableSchema::new(0, &schema),
+ None,
+ );
+ let batch = |id, json| {
+ let variant =
crate::variant::GenericVariant::parse_json(json).unwrap();
+ let ArrowDataType::Struct(fields) = variant_arrow_type() else {
+ unreachable!()
+ };
+ let payload = StructArray::try_new(
+ fields,
+ vec![
+ Arc::new(BinaryArray::from(vec![Some(variant.value())])),
+
Arc::new(BinaryArray::from(vec![Some(variant.metadata())])),
+ ],
+ None,
+ )
+ .unwrap();
+ RecordBatch::try_new(
+ build_target_arrow_schema(table.schema().fields()).unwrap(),
+ vec![Arc::new(Int32Array::from(vec![id])), Arc::new(payload)],
+ )
+ .unwrap()
+ };
+
+ write_commit(&table, &batch(1, r#"{"x":"invalid"}"#)).await;
+ write_commit(&table, &batch(1, r#"{"x":1.5}"#)).await;
+
+ let mut read_builder = table.new_read_builder();
+ read_builder.with_read_type(vec![
+ table.schema().fields()[0].clone(),
+ DataField::new(
+ 1,
+ "payload".to_string(),
+ DataType::Row(
+ variant_extraction_row(
+ true,
+ [(
+ DataType::Float(FloatType::new()),
+ "$.x".to_string(),
+ true,
+ "UTC".to_string(),
+ )],
+ )
+ .unwrap(),
+ ),
+ ),
+ ]);
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ let batches = read_builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(int_column(&batches, "id"), vec![1]);
+ let payload = batches[0]
+ .column_by_name("payload")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let x = payload
+ .column(0)
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap();
+ assert_eq!(x.value(0), 1.5);
+
+ // A bad value that survives the merge must still fail strict
projection.
+ write_commit(&table, &batch(1, r#"{"x":"still invalid"}"#)).await;
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ let error = read_builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap_err();
+ assert!(matches!(error, Error::DataInvalid { ref message, .. }
+ if message.contains("Cannot cast Variant value")));
+ }
+
+ #[tokio::test]
+ async fn kv_variant_projection_preserves_requested_historical_field_name()
{
+ let file_io = test_file_io();
+ let table_path = "memory:/kv_variant_projection_renamed_field";
+ setup_dirs(&file_io, table_path).await;
+ let schema = |name: &str| {
+ Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column(name, DataType::Variant(VariantType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .build()
+ .unwrap()
+ };
+ let old_schema = TableSchema::new(0, &schema("payload"));
+ let new_schema = TableSchema::new(1, &schema("renamed_payload"));
+ let table = |schema: TableSchema| {
+ Table::new(
+ file_io.clone(),
+ Identifier::new("default",
"kv_variant_projection_renamed_field"),
+ table_path.to_string(),
+ schema,
+ None,
+ )
+ };
+ let old_table = table(old_schema.clone());
+ let new_table = table(new_schema.clone());
+ write_schema_file(&old_table, &old_schema).await;
+ write_schema_file(&new_table, &new_schema).await;
+
+ let batch = |table: &Table, json| {
+ let variant =
crate::variant::GenericVariant::parse_json(json).unwrap();
+ let ArrowDataType::Struct(fields) = variant_arrow_type() else {
+ unreachable!()
+ };
+ let payload = StructArray::try_new(
+ fields,
+ vec![
+ Arc::new(BinaryArray::from(vec![Some(variant.value())])),
+
Arc::new(BinaryArray::from(vec![Some(variant.metadata())])),
+ ],
+ None,
+ )
+ .unwrap();
+ RecordBatch::try_new(
+ build_target_arrow_schema(table.schema().fields()).unwrap(),
+ vec![Arc::new(Int32Array::from(vec![1])), Arc::new(payload)],
+ )
+ .unwrap()
+ };
+ write_commit(&old_table, &batch(&old_table,
r#"{"x":"invalid"}"#)).await;
+ write_commit(&new_table, &batch(&new_table, r#"{"x":1.5}"#)).await;
+
+ let mut read_builder = new_table.new_read_builder();
+ read_builder.with_read_type(vec![
+ new_table.schema().fields()[0].clone(),
+ DataField::new(
+ 1,
+ "payload".to_string(),
+ DataType::Row(
+ variant_extraction_row(
+ true,
+ [(
+ DataType::Float(FloatType::new()),
+ "$.x".to_string(),
+ true,
+ "UTC".to_string(),
+ )],
+ )
+ .unwrap(),
+ ),
+ ),
+ ]);
+ let plan = read_builder.new_scan().plan().await.unwrap();
+ let batches = read_builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+ assert_eq!(int_column(&batches, "id"), vec![1]);
+ let payload = batches[0]
+ .column_by_name("payload")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let x = payload
+ .column(0)
+ .as_any()
+ .downcast_ref::<Float32Array>()
+ .unwrap();
+ assert!(!payload.is_null(0));
+ assert!(!x.is_null(0));
+ assert_eq!(x.value(0), 1.5);
+ }
+
#[test]
fn kv_row_layout_preserves_file_schema_key_names_and_fields() {
let fields = vec![
diff --git a/crates/paimon/src/variant.rs b/crates/paimon/src/variant.rs
index a316c929..829e9e95 100644
--- a/crates/paimon/src/variant.rs
+++ b/crates/paimon/src/variant.rs
@@ -1297,6 +1297,19 @@ fn parse_path(path: &str) -> Result<Vec<PathSegment>> {
Ok(segments)
}
+/// Return the literal key selected by a one-segment object path.
+///
+/// Variant extraction read types carry SQL-style paths. The vectorized
+/// numeric projection can batch literal top-level keys, while nested paths
+/// continue through the general `variant_get` implementation.
+pub(crate) fn top_level_path_key(path: &str) -> Result<Option<String>> {
+ let segments = parse_path(path)?;
+ Ok(match segments.as_slice() {
+ [PathSegment::Key(key)] => Some(key.clone()),
+ _ => None,
+ })
+}
+
#[derive(Clone)]
struct FieldEntry {
key: String,
@@ -2708,9 +2721,7 @@ fn cast_variant_to_extraction_value(
.and_then(|value| i32::try_from(value).ok())
.map(ShreddedValue::Int32),
VariantScalarSchema::Int64 =>
cast_variant_to_i64(variant).map(ShreddedValue::Int64),
- VariantScalarSchema::Float32 => {
- cast_variant_to_f64(variant).map(|value|
ShreddedValue::Float32(value as f32))
- }
+ VariantScalarSchema::Float32 =>
cast_variant_to_f32(variant).map(ShreddedValue::Float32),
VariantScalarSchema::Float64 =>
cast_variant_to_f64(variant).map(ShreddedValue::Float64),
VariantScalarSchema::Decimal { precision, scale } => {
cast_variant_to_decimal(variant, *precision,
*scale).map(ShreddedValue::Decimal128)
@@ -2798,6 +2809,31 @@ fn cast_variant_to_f64(variant: VariantRef<'_>) ->
Option<f64> {
}
}
+pub(crate) fn cast_variant_to_f32(variant: VariantRef<'_>) -> Option<f32> {
+ match variant.kind().ok()? {
+ VariantKind::Boolean => Some(if variant.get_boolean().ok()? {
+ 1.0
+ } else {
+ 0.0
+ }),
+ VariantKind::Long => Some(variant.get_long().ok()? as f32),
+ VariantKind::Float => variant.get_float().ok(),
+ VariantKind::Double => Some(variant.get_double().ok()? as f32),
+ VariantKind::Decimal => {
+ let decimal = variant.get_decimal().ok()?;
+ Some((decimal.unscaled as f64 / 10f64.powi(decimal.scale as i32))
as f32)
+ }
+ // Java Float.parseFloat removes only leading/trailing characters <=
U+0020.
+ VariantKind::String => variant
+ .get_string()
+ .ok()?
+ .trim_matches(|c| c <= '\u{20}')
+ .parse::<f32>()
+ .ok(),
+ _ => None,
+ }
+}
+
fn cast_variant_to_string(variant: VariantRef<'_>) -> Option<String> {
match variant.kind().ok()? {
VariantKind::Object | VariantKind::Array => variant.to_json().ok(),
@@ -3637,4 +3673,65 @@ mod tests {
"2.5 truncates to 2"
);
}
+
+ #[test]
+ fn
float32_projection_casts_supported_sources_without_reinterpreting_temporal_values()
{
+ let variant = GenericVariant::parse_json(
+
r#"{"boolean":true,"integer":27,"decimal":1.25,"string":"3.5","invalid":"bad"}"#,
+ )
+ .unwrap();
+ for (path, expected) in [
+ ("$.boolean", Some(1.0)),
+ ("$.integer", Some(27.0)),
+ ("$.decimal", Some(1.25)),
+ ("$.string", Some(3.5)),
+ ("$.invalid", None),
+ ] {
+ assert_eq!(
+ cast_variant_to_f32(variant.get_path(path).unwrap().unwrap()),
+ expected
+ );
+ }
+
+ let mut date = VariantBuilder::new();
+ date.append_date(20_000);
+ assert_eq!(
+ cast_variant_to_f32(date.result().unwrap().as_ref().unwrap()),
+ None
+ );
+
+ let mut timestamp = VariantBuilder::new();
+ timestamp.append_timestamp(1_700_000_000_123_456);
+ assert_eq!(
+ cast_variant_to_f32(timestamp.result().unwrap().as_ref().unwrap()),
+ None
+ );
+ }
+
+ #[test]
+ fn float32_projection_uses_java_string_whitespace_rules() {
+ let variant = GenericVariant::parse_json(
+ r#"{"ascii":" \t1.5\n
","control":"\u00001.5\u001f","nbsp":"\u00a01.5\u00a0"}"#,
+ )
+ .unwrap();
+ let float_type = DataType::Float(crate::spec::FloatType::new());
+
+ for path in ["$.ascii", "$.control"] {
+ let value = variant.get_path(path).unwrap().unwrap();
+ assert_eq!(
+ cast_variant_to_shredded_value(value, &float_type,
true).unwrap(),
+ Some(ShreddedValue::Float32(1.5))
+ );
+ }
+
+ let nbsp = variant.get_path("$.nbsp").unwrap().unwrap();
+ assert_eq!(
+ cast_variant_to_shredded_value(nbsp, &float_type, false).unwrap(),
+ None
+ );
+ assert!(matches!(
+ cast_variant_to_shredded_value(nbsp, &float_type, true),
+ Err(Error::DataInvalid { .. })
+ ));
+ }
}
diff --git a/crates/paimon/src/variant/numeric.rs
b/crates/paimon/src/variant/numeric.rs
index 34ab0bea..8c797481 100644
--- a/crates/paimon/src/variant/numeric.rs
+++ b/crates/paimon/src/variant/numeric.rs
@@ -35,6 +35,12 @@ struct ProjectedField {
outputs: Vec<usize>,
}
+#[derive(Clone, Copy)]
+enum Float32ProjectionMode<'a> {
+ Numeric,
+ Cast { fail_on_error: &'a [bool] },
+}
+
impl VariantFloat32Projection {
pub(crate) fn new(metadata: &[u8], fields: &[String]) -> Result<Self> {
validate_metadata(metadata)?;
@@ -75,6 +81,43 @@ impl VariantFloat32Projection {
if output.len() != self.output_width {
return data_invalid("Invalid Variant numeric output width");
}
+ self.extract_float32_impl(
+ value,
+ metadata,
+ offsets,
+ output,
+ Float32ProjectionMode::Numeric,
+ )
+ }
+
+ pub(crate) fn extract_float32_cast(
+ &mut self,
+ value: &[u8],
+ metadata: &[u8],
+ fail_on_error: &[bool],
+ offsets: &mut Vec<usize>,
+ output: &mut [Option<f32>],
+ ) -> Result<()> {
+ if output.len() != self.output_width || fail_on_error.len() !=
self.output_width {
+ return data_invalid("Invalid Variant float32 projection width");
+ }
+ self.extract_float32_impl(
+ value,
+ metadata,
+ offsets,
+ output,
+ Float32ProjectionMode::Cast { fail_on_error },
+ )
+ }
+
+ fn extract_float32_impl(
+ &mut self,
+ value: &[u8],
+ metadata: &[u8],
+ offsets: &mut Vec<usize>,
+ output: &mut [Option<f32>],
+ mode: Float32ProjectionMode<'_>,
+ ) -> Result<()> {
output.fill(None);
if value_kind(value, 0)? != VariantKind::Object {
validate_payload(value, metadata)?;
@@ -139,7 +182,27 @@ impl VariantFloat32Projection {
return data_invalid("Malformed Variant child size");
}
let child = VariantRef::new_at(value, metadata, child_pos)?;
- let numeric = numeric_to_float32(child, &projected.name)?;
+ let numeric = match mode {
+ Float32ProjectionMode::Numeric => numeric_to_float32(child,
&projected.name)?,
+ Float32ProjectionMode::Cast { fail_on_error } => {
+ if child.is_null()? {
+ continue;
+ }
+ let numeric = cast_variant_to_f32(child);
+ if numeric.is_none()
+ && projected
+ .outputs
+ .iter()
+ .any(|output_index| fail_on_error[*output_index])
+ {
+ return data_invalid(format!(
+ "Cannot cast Variant value to {:?}",
+ DataType::Float(FloatType::new())
+ ));
+ }
+ numeric
+ }
+ };
for output_index in &projected.outputs {
output[*output_index] = numeric;
}
