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 839448a Add Variant support and DataFusion functions (#458)
839448a is described below
commit 839448a04a4b0cec67197bd7cb63991565e0d835
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Jul 5 23:48:37 2026 +0800
Add Variant support and DataFusion functions (#458)
---
crates/integrations/datafusion/src/lib.rs | 2 +
crates/integrations/datafusion/src/sql_context.rs | 22 +-
.../datafusion/src/variant_functions.rs | 877 ++++++++++
crates/paimon/src/arrow/format/parquet.rs | 1 +
crates/paimon/src/arrow/format/row.rs | 253 ++-
crates/paimon/src/arrow/format/vortex.rs | 1 +
crates/paimon/src/arrow/mod.rs | 108 +-
crates/paimon/src/btree/key_serde.rs | 11 +-
crates/paimon/src/lib.rs | 1 +
crates/paimon/src/spec/binary_row.rs | 164 +-
crates/paimon/src/spec/data_type_casts.rs | 1 +
crates/paimon/src/spec/partition_utils.rs | 1 +
crates/paimon/src/spec/predicate.rs | 29 +
crates/paimon/src/spec/types.rs | 55 +
crates/paimon/src/variant.rs | 1807 ++++++++++++++++++++
crates/paimon/tests/fixtures/variant_type.json | 1 +
.../tests/fixtures/variant_type_nullable.json | 1 +
docs/src/sql.md | 154 ++
18 files changed, 3478 insertions(+), 11 deletions(-)
diff --git a/crates/integrations/datafusion/src/lib.rs
b/crates/integrations/datafusion/src/lib.rs
index de6e932..8ede2a0 100644
--- a/crates/integrations/datafusion/src/lib.rs
+++ b/crates/integrations/datafusion/src/lib.rs
@@ -54,6 +54,7 @@ mod system_tables;
mod table;
mod table_function_args;
mod update;
+mod variant_functions;
mod vector_search;
use std::collections::HashMap;
@@ -74,4 +75,5 @@ pub use physical_plan::PaimonTableScan;
pub use relation_planner::PaimonRelationPlanner;
pub use sql_context::SQLContext;
pub use table::PaimonTableProvider;
+pub use variant_functions::register_variant_functions;
pub use vector_search::{register_vector_search, VectorSearchFunction};
diff --git a/crates/integrations/datafusion/src/sql_context.rs
b/crates/integrations/datafusion/src/sql_context.rs
index 6170d6b..c550eff 100644
--- a/crates/integrations/datafusion/src/sql_context.rs
+++ b/crates/integrations/datafusion/src/sql_context.rs
@@ -62,7 +62,7 @@ use paimon::spec::{
ArrayType as PaimonArrayType, BigIntType, BlobType, BooleanType, DataField
as PaimonDataField,
DataType as PaimonDataType, DateType, Datum, DecimalType, DoubleType,
FloatType, IntType,
LocalZonedTimestampType, MapType as PaimonMapType, RowType as
PaimonRowType, SchemaChange,
- SmallIntType, TimestampType, TinyIntType, VarBinaryType, VarCharType,
+ SmallIntType, TimestampType, TinyIntType, VarBinaryType, VarCharType,
VariantType,
};
use crate::error::to_datafusion_error;
@@ -106,6 +106,7 @@ impl SQLContext {
))
.build();
let ctx = SessionContext::new_with_state(state);
+ crate::variant_functions::register_variant_functions(&ctx);
Self {
ctx,
catalogs: HashMap::new(),
@@ -1683,6 +1684,13 @@ fn sql_data_type_to_paimon_type(
))
}
SqlType::Blob(_) =>
Ok(PaimonDataType::Blob(BlobType::with_nullable(nullable))),
+ SqlType::Custom(name, modifiers)
+ if name.to_string().eq_ignore_ascii_case("VARIANT") &&
modifiers.is_empty() =>
+ {
+ Ok(PaimonDataType::Variant(VariantType::with_nullable(
+ nullable,
+ )))
+ }
SqlType::Date =>
Ok(PaimonDataType::Date(DateType::with_nullable(nullable))),
SqlType::Timestamp(precision, tz_info) => {
let precision = match precision {
@@ -2122,7 +2130,8 @@ fn datum_to_constant_array(
| Datum::Timestamp { .. }
| Datum::LocalZonedTimestamp { .. }
| Datum::Decimal { .. }
- | Datum::Bytes(_) => Err(DataFusionError::Plan(format!(
+ | Datum::Bytes(_)
+ | Datum::Variant { .. } => Err(DataFusionError::Plan(format!(
"Unsupported datum type for partition column: {d}"
))),
},
@@ -2835,6 +2844,15 @@ mod tests {
);
}
+ #[test]
+ fn test_sql_type_variant() {
+ use datafusion::sql::sqlparser::ast::{DataType as SqlType, Ident,
ObjectName};
+ assert_sql_type_to_paimon(
+ SqlType::Custom(ObjectName::from(Ident::new("VARIANT")), vec![]),
+ PaimonDataType::Variant(VariantType::new()),
+ );
+ }
+
#[test]
fn test_sql_type_date() {
use datafusion::sql::sqlparser::ast::DataType as SqlType;
diff --git a/crates/integrations/datafusion/src/variant_functions.rs
b/crates/integrations/datafusion/src/variant_functions.rs
new file mode 100644
index 0000000..6eb8ab8
--- /dev/null
+++ b/crates/integrations/datafusion/src/variant_functions.rs
@@ -0,0 +1,877 @@
+// 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 datafusion::arrow::array::{
+ Array, ArrayRef, BinaryArray, BinaryBuilder, BooleanBuilder,
LargeStringArray, StringArray,
+ StringViewArray, StructArray,
+};
+use datafusion::arrow::buffer::{BooleanBuffer, NullBuffer};
+use datafusion::arrow::datatypes::{DataType as ArrowDataType, Field, FieldRef,
Fields};
+use datafusion::common::{DataFusionError, Result as DFResult, ScalarValue};
+use datafusion::logical_expr::{
+ ColumnarValue, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDF,
ScalarUDFImpl, Signature,
+ Volatility,
+};
+use datafusion::prelude::SessionContext;
+use paimon::variant::{GenericVariant, VariantDecimal, VariantKind, VariantRef};
+
+pub fn register_variant_functions(ctx: &SessionContext) {
+ ctx.register_udf(ScalarUDF::from(ParseJsonFunc::new(false)));
+ ctx.register_udf(ScalarUDF::from(ParseJsonFunc::new(true)));
+ ctx.register_udf(ScalarUDF::from(IsVariantNullFunc::new()));
+ ctx.register_udf(ScalarUDF::from(VariantGetFunc::new(false)));
+ ctx.register_udf(ScalarUDF::from(VariantGetFunc::new(true)));
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct ParseJsonFunc {
+ try_parse: bool,
+ signature: Signature,
+}
+
+impl ParseJsonFunc {
+ fn new(try_parse: bool) -> Self {
+ Self {
+ try_parse,
+ signature: Signature::string(1, Volatility::Immutable),
+ }
+ }
+}
+
+impl ScalarUDFImpl for ParseJsonFunc {
+ fn name(&self) -> &str {
+ if self.try_parse {
+ "try_parse_json"
+ } else {
+ "parse_json"
+ }
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[ArrowDataType]) ->
DFResult<ArrowDataType> {
+ Ok(variant_arrow_type())
+ }
+
+ fn return_field_from_args(&self, _args: ReturnFieldArgs) ->
DFResult<FieldRef> {
+ Ok(Arc::new(Field::new(
+ self.name(),
+ variant_arrow_type(),
+ true,
+ )))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) ->
DFResult<ColumnarValue> {
+ if args.args.len() != 1 {
+ return plan_err(format!("{} expects 1 argument", self.name()));
+ }
+ let arrays = ColumnarValue::values_to_arrays(&args.args)?;
+ let input = arrays[0].as_ref();
+ let mut values = Vec::with_capacity(input.len());
+ for row in 0..input.len() {
+ let Some(json) = string_at(input, row)? else {
+ values.push(None);
+ continue;
+ };
+ match GenericVariant::parse_json(&json) {
+ Ok(variant) => values.push(Some(variant)),
+ Err(e) if self.try_parse => {
+ let _ = e;
+ values.push(None);
+ }
+ Err(e) => return Err(to_df_error(e)),
+ }
+ }
+ Ok(ColumnarValue::Array(variant_array(values)?))
+ }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct IsVariantNullFunc {
+ signature: Signature,
+}
+
+impl IsVariantNullFunc {
+ fn new() -> Self {
+ Self {
+ signature: Signature::any(1, Volatility::Immutable),
+ }
+ }
+}
+
+impl ScalarUDFImpl for IsVariantNullFunc {
+ fn name(&self) -> &str {
+ "is_variant_null"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[ArrowDataType]) ->
DFResult<ArrowDataType> {
+ Ok(ArrowDataType::Boolean)
+ }
+
+ fn return_field_from_args(&self, _args: ReturnFieldArgs) ->
DFResult<FieldRef> {
+ Ok(Arc::new(Field::new(
+ self.name(),
+ ArrowDataType::Boolean,
+ false,
+ )))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) ->
DFResult<ColumnarValue> {
+ if args.args.len() != 1 {
+ return plan_err("is_variant_null expects 1 argument");
+ }
+ let arrays = ColumnarValue::values_to_arrays(&args.args)?;
+ let input = arrays[0].as_ref();
+ let mut builder = BooleanBuilder::new();
+ let Some((values, metadata)) = variant_children(input)? else {
+ for _ in 0..input.len() {
+ builder.append_value(false);
+ }
+ return Ok(ColumnarValue::Array(Arc::new(builder.finish())));
+ };
+
+ for row in 0..input.len() {
+ if input.is_null(row) {
+ builder.append_value(false);
+ } else {
+ let variant = VariantRef::new(values.value(row),
metadata.value(row), 0)
+ .map_err(to_df_error)?;
+ builder.append_value(variant.is_null().map_err(to_df_error)?);
+ }
+ }
+ Ok(ColumnarValue::Array(Arc::new(builder.finish())))
+ }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+struct VariantGetFunc {
+ try_get: bool,
+ signature: Signature,
+}
+
+impl VariantGetFunc {
+ fn new(try_get: bool) -> Self {
+ Self {
+ try_get,
+ signature: Signature::variadic_any(Volatility::Immutable),
+ }
+ }
+}
+
+impl ScalarUDFImpl for VariantGetFunc {
+ fn name(&self) -> &str {
+ if self.try_get {
+ "try_variant_get"
+ } else {
+ "variant_get"
+ }
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[ArrowDataType]) ->
DFResult<ArrowDataType> {
+ internal_err("return_field_from_args should be used for variant_get")
+ }
+
+ fn return_field_from_args(&self, args: ReturnFieldArgs) ->
DFResult<FieldRef> {
+ if args.arg_fields.len() != 2 && args.arg_fields.len() != 3 {
+ return plan_err(format!("{} expects 2 or 3 arguments",
self.name()));
+ }
+ let output = match args.arg_fields.len() {
+ 2 => variant_get_output_type(None)?,
+ 3 => {
+ let Some(type_arg) = args.scalar_arguments.get(2).and_then(|v|
*v) else {
+ return plan_err("variant_get type argument must be a
string literal");
+ };
+ variant_get_output_type(Some(type_arg))?
+ }
+ _ => unreachable!("argument count checked above"),
+ };
+ Ok(Arc::new(Field::new(
+ self.name(),
+ output.arrow_type().clone(),
+ true,
+ )))
+ }
+
+ fn invoke_with_args(&self, args: ScalarFunctionArgs) ->
DFResult<ColumnarValue> {
+ if args.args.len() != 2 && args.args.len() != 3 {
+ return plan_err(format!("{} expects 2 or 3 arguments",
self.name()));
+ }
+ let output = if args.return_type() == &variant_arrow_type() {
+ VariantGetOutput::Variant
+ } else {
+ VariantGetOutput::Scalar(args.return_type().clone())
+ };
+ let arrays = ColumnarValue::values_to_arrays(&args.args)?;
+ let variants = arrays[0].as_ref();
+ let paths = arrays[1].as_ref();
+ let Some((values, metadata)) = variant_children(variants)? else {
+ return Ok(ColumnarValue::Array(null_array(
+ output.arrow_type(),
+ variants.len(),
+ )));
+ };
+
+ match output {
+ VariantGetOutput::Variant => {
+ let mut result = Vec::with_capacity(variants.len());
+ for row in 0..variants.len() {
+ result.push(self.variant_at_path(variants, values,
metadata, paths, row)?);
+ }
+ Ok(ColumnarValue::Array(variant_array(result)?))
+ }
+ VariantGetOutput::Scalar(data_type) => {
+ let mut scalars = Vec::with_capacity(variants.len());
+ for row in 0..variants.len() {
+ match self.variant_at_path_ref(variants, values, metadata,
paths, row)? {
+ Some(variant) => scalars.push(cast_variant_to_scalar(
+ variant,
+ &data_type,
+ !self.try_get,
+ )?),
+ None =>
scalars.push(ScalarValue::try_from(&data_type)?),
+ }
+ }
+ if scalars.is_empty() {
+ return Ok(ColumnarValue::Array(null_array(&data_type, 0)));
+ }
+ Ok(ColumnarValue::Array(ScalarValue::iter_to_array(scalars)?))
+ }
+ }
+ }
+}
+
+impl VariantGetFunc {
+ fn variant_at_path(
+ &self,
+ variants: &dyn Array,
+ values: &BinaryArray,
+ metadata: &BinaryArray,
+ paths: &dyn Array,
+ row: usize,
+ ) -> DFResult<Option<GenericVariant>> {
+ self.variant_at_path_ref(variants, values, metadata, paths, row)?
+ .map(|variant| variant.to_owned_variant().map_err(to_df_error))
+ .transpose()
+ }
+
+ fn variant_at_path_ref<'a>(
+ &self,
+ variants: &dyn Array,
+ values: &'a BinaryArray,
+ metadata: &'a BinaryArray,
+ paths: &dyn Array,
+ row: usize,
+ ) -> DFResult<Option<VariantRef<'a>>> {
+ if variants.is_null(row) || paths.is_null(row) {
+ return Ok(None);
+ }
+ let path = string_at(paths, row)?;
+ let Some(path) = path else {
+ return Ok(None);
+ };
+ let variant =
+ VariantRef::new(values.value(row), metadata.value(row),
0).map_err(to_df_error)?;
+ match variant.get_path(&path) {
+ Ok(value) => Ok(value),
+ Err(e) if self.try_get => {
+ let _ = e;
+ Ok(None)
+ }
+ Err(e) => Err(to_df_error(e)),
+ }
+ }
+}
+
+#[derive(Clone, Debug)]
+enum VariantGetOutput {
+ Variant,
+ Scalar(ArrowDataType),
+}
+
+impl VariantGetOutput {
+ fn arrow_type(&self) -> &ArrowDataType {
+ match self {
+ Self::Variant => {
+ static VARIANT_TYPE: std::sync::LazyLock<ArrowDataType> =
+ std::sync::LazyLock::new(variant_arrow_type);
+ &VARIANT_TYPE
+ }
+ Self::Scalar(data_type) => data_type,
+ }
+ }
+}
+
+fn variant_get_output_type(type_arg: Option<&ScalarValue>) ->
DFResult<VariantGetOutput> {
+ let Some(type_arg) = type_arg else {
+ return Ok(VariantGetOutput::Variant);
+ };
+ let type_name = match type_arg {
+ ScalarValue::Utf8(Some(v))
+ | ScalarValue::LargeUtf8(Some(v))
+ | ScalarValue::Utf8View(Some(v)) => v,
+ ScalarValue::Utf8(None) | ScalarValue::LargeUtf8(None) |
ScalarValue::Utf8View(None) => {
+ return plan_err("variant_get type argument must not be NULL");
+ }
+ _ => return plan_err("variant_get type argument must be a string
literal"),
+ };
+ parse_variant_get_type(type_name)
+}
+
+fn parse_variant_get_type(type_name: &str) -> DFResult<VariantGetOutput> {
+ let normalized = type_name.trim().to_ascii_lowercase();
+ match normalized.as_str() {
+ "variant" => Ok(VariantGetOutput::Variant),
+ "boolean" | "bool" =>
Ok(VariantGetOutput::Scalar(ArrowDataType::Boolean)),
+ "byte" | "tinyint" =>
Ok(VariantGetOutput::Scalar(ArrowDataType::Int8)),
+ "short" | "smallint" =>
Ok(VariantGetOutput::Scalar(ArrowDataType::Int16)),
+ "int" | "integer" =>
Ok(VariantGetOutput::Scalar(ArrowDataType::Int32)),
+ "long" | "bigint" =>
Ok(VariantGetOutput::Scalar(ArrowDataType::Int64)),
+ "float" | "real" =>
Ok(VariantGetOutput::Scalar(ArrowDataType::Float32)),
+ "double" => Ok(VariantGetOutput::Scalar(ArrowDataType::Float64)),
+ "string" | "varchar" | "text" =>
Ok(VariantGetOutput::Scalar(ArrowDataType::Utf8)),
+ "decimal" => Ok(VariantGetOutput::Scalar(ArrowDataType::Decimal128(10,
0))),
+ _ if normalized.starts_with("decimal(") && normalized.ends_with(')')
=> {
+ let inner = &normalized["decimal(".len()..normalized.len() - 1];
+ let Some((precision, scale)) = inner.split_once(',') else {
+ return plan_err(format!("Invalid decimal type for variant_get:
{type_name}"));
+ };
+ let precision = precision
+ .trim()
+ .parse::<u8>()
+ .map_err(|e| DataFusionError::Plan(format!("Invalid decimal
precision: {e}")))?;
+ let scale = scale
+ .trim()
+ .parse::<i8>()
+ .map_err(|e| DataFusionError::Plan(format!("Invalid decimal
scale: {e}")))?;
+ Ok(VariantGetOutput::Scalar(ArrowDataType::Decimal128(
+ precision, scale,
+ )))
+ }
+ _ => plan_err(format!("Unsupported variant_get type: {type_name}")),
+ }
+}
+
+fn cast_variant_to_scalar(
+ variant: VariantRef<'_>,
+ target: &ArrowDataType,
+ fail_on_error: bool,
+) -> DFResult<ScalarValue> {
+ if variant.is_null().map_err(to_df_error)? {
+ return ScalarValue::try_from(target);
+ }
+ let result = match target {
+ ArrowDataType::Boolean => cast_to_boolean(variant),
+ ArrowDataType::Int8 => cast_to_i64(variant).and_then(|v| {
+ i8::try_from(v)
+ .map(ScalarValue::from)
+ .map_err(|_| invalid_cast())
+ }),
+ ArrowDataType::Int16 => cast_to_i64(variant).and_then(|v| {
+ i16::try_from(v)
+ .map(ScalarValue::from)
+ .map_err(|_| invalid_cast())
+ }),
+ ArrowDataType::Int32 => cast_to_i64(variant).and_then(|v| {
+ i32::try_from(v)
+ .map(ScalarValue::from)
+ .map_err(|_| invalid_cast())
+ }),
+ ArrowDataType::Int64 => cast_to_i64(variant).map(ScalarValue::from),
+ ArrowDataType::Float32 => {
+ cast_to_f64(variant).map(|v| ScalarValue::Float32(Some(v as f32)))
+ }
+ ArrowDataType::Float64 => cast_to_f64(variant).map(ScalarValue::from),
+ ArrowDataType::Utf8 => cast_to_string(variant).map(ScalarValue::from),
+ ArrowDataType::Decimal128(precision, scale) =>
cast_to_decimal(variant, *precision, *scale),
+ _ => Err(invalid_cast()),
+ };
+
+ match result {
+ Ok(value) => Ok(value),
+ Err(e) if !fail_on_error => {
+ let _ = e;
+ ScalarValue::try_from(target)
+ }
+ Err(e) => Err(e),
+ }
+}
+
+fn cast_to_boolean(variant: VariantRef<'_>) -> DFResult<ScalarValue> {
+ match variant.kind().map_err(to_df_error)? {
+ VariantKind::Boolean => Ok(ScalarValue::Boolean(Some(
+ variant.get_boolean().map_err(to_df_error)?,
+ ))),
+ VariantKind::String => match variant
+ .get_string()
+ .map_err(to_df_error)?
+ .to_ascii_lowercase()
+ .as_str()
+ {
+ "true" => Ok(ScalarValue::Boolean(Some(true))),
+ "false" => Ok(ScalarValue::Boolean(Some(false))),
+ _ => Err(invalid_cast()),
+ },
+ _ => Err(invalid_cast()),
+ }
+}
+
+fn cast_to_i64(variant: VariantRef<'_>) -> DFResult<i64> {
+ match variant.kind().map_err(to_df_error)? {
+ VariantKind::Long
+ | VariantKind::Date
+ | VariantKind::Timestamp
+ | VariantKind::TimestampNtz => variant.get_long().map_err(to_df_error),
+ VariantKind::String => variant
+ .get_string()
+ .map_err(to_df_error)?
+ .parse::<i64>()
+ .map_err(|_| invalid_cast()),
+ VariantKind::Decimal => {
+ let decimal = variant.get_decimal().map_err(to_df_error)?;
+ rescale_decimal(decimal.unscaled, decimal.scale, 0)
+ .and_then(|v| i64::try_from(v).map_err(|_| invalid_cast()))
+ }
+ _ => Err(invalid_cast()),
+ }
+}
+
+fn cast_to_f64(variant: VariantRef<'_>) -> DFResult<f64> {
+ match variant.kind().map_err(to_df_error)? {
+ VariantKind::Long
+ | VariantKind::Date
+ | VariantKind::Timestamp
+ | VariantKind::TimestampNtz =>
Ok(variant.get_long().map_err(to_df_error)? as f64),
+ VariantKind::Double => variant.get_double().map_err(to_df_error),
+ VariantKind::Float => Ok(variant.get_float().map_err(to_df_error)? as
f64),
+ VariantKind::Decimal => {
+ let decimal = variant.get_decimal().map_err(to_df_error)?;
+ Ok(decimal.unscaled as f64 / 10f64.powi(decimal.scale as i32))
+ }
+ VariantKind::String => variant
+ .get_string()
+ .map_err(to_df_error)?
+ .parse::<f64>()
+ .map_err(|_| invalid_cast()),
+ _ => Err(invalid_cast()),
+ }
+}
+
+fn cast_to_string(variant: VariantRef<'_>) -> DFResult<String> {
+ match variant.kind().map_err(to_df_error)? {
+ VariantKind::Object | VariantKind::Array =>
variant.to_json().map_err(to_df_error),
+ VariantKind::Boolean =>
Ok(variant.get_boolean().map_err(to_df_error)?.to_string()),
+ VariantKind::Long
+ | VariantKind::Date
+ | VariantKind::Timestamp
+ | VariantKind::TimestampNtz =>
Ok(variant.get_long().map_err(to_df_error)?.to_string()),
+ VariantKind::String => variant.get_string().map_err(to_df_error),
+ VariantKind::Double =>
Ok(variant.get_double().map_err(to_df_error)?.to_string()),
+ VariantKind::Decimal => Ok(variant
+ .get_decimal()
+ .map_err(to_df_error)?
+ .to_plain_string()),
+ VariantKind::Float =>
Ok(variant.get_float().map_err(to_df_error)?.to_string()),
+ _ => variant.to_json().map_err(to_df_error),
+ }
+}
+
+fn cast_to_decimal(variant: VariantRef<'_>, precision: u8, scale: i8) ->
DFResult<ScalarValue> {
+ let unscaled = match variant.kind().map_err(to_df_error)? {
+ VariantKind::Long
+ | VariantKind::Date
+ | VariantKind::Timestamp
+ | VariantKind::TimestampNtz => {
+ rescale_decimal(variant.get_long().map_err(to_df_error)? as i128,
0, scale)?
+ }
+ VariantKind::Decimal => {
+ let decimal = variant.get_decimal().map_err(to_df_error)?;
+ rescale_decimal(decimal.unscaled, decimal.scale, scale)?
+ }
+ VariantKind::String => {
+ let parsed =
parse_decimal_string(&variant.get_string().map_err(to_df_error)?)
+ .ok_or_else(invalid_cast)?;
+ rescale_decimal(parsed.unscaled, parsed.scale, scale)?
+ }
+ _ => return Err(invalid_cast()),
+ };
+ if decimal_precision(unscaled) > precision {
+ return Err(invalid_cast());
+ }
+ Ok(ScalarValue::Decimal128(Some(unscaled), precision, scale))
+}
+
+fn rescale_decimal(unscaled: i128, from_scale: i8, to_scale: i8) ->
DFResult<i128> {
+ match to_scale.cmp(&from_scale) {
+ std::cmp::Ordering::Equal => Ok(unscaled),
+ std::cmp::Ordering::Greater => {
+ let factor = 10_i128
+ .checked_pow((to_scale - from_scale) as u32)
+ .ok_or_else(invalid_cast)?;
+ unscaled.checked_mul(factor).ok_or_else(invalid_cast)
+ }
+ std::cmp::Ordering::Less => {
+ let factor = 10_i128
+ .checked_pow((from_scale - to_scale) as u32)
+ .ok_or_else(invalid_cast)?;
+ if unscaled % factor == 0 {
+ Ok(unscaled / factor)
+ } else {
+ Err(invalid_cast())
+ }
+ }
+ }
+}
+
+fn parse_decimal_string(input: &str) -> Option<VariantDecimal> {
+ let input = input.trim();
+ if input.is_empty() || input.contains(['e', 'E']) {
+ return None;
+ }
+ let negative = input.starts_with('-');
+ let unsigned = input.strip_prefix('-').unwrap_or(input);
+ if unsigned.is_empty()
+ || unsigned.matches('.').count() > 1
+ || !unsigned.bytes().all(|ch| ch == b'.' || ch.is_ascii_digit())
+ {
+ return None;
+ }
+ let scale = unsigned
+ .split_once('.')
+ .map(|(_, fraction)| fraction.len())
+ .unwrap_or(0);
+ let digits: String = unsigned
+ .bytes()
+ .filter(|ch| *ch != b'.')
+ .map(char::from)
+ .collect();
+ let significant = digits.trim_start_matches('0');
+ let precision = if significant.is_empty() {
+ 1
+ } else {
+ significant.len()
+ };
+ if precision > 38 || scale > 38 {
+ return None;
+ }
+ let mut unscaled = digits.parse::<i128>().ok()?;
+ if negative {
+ unscaled = -unscaled;
+ }
+ Some(VariantDecimal {
+ unscaled,
+ precision: precision as u8,
+ scale: scale as i8,
+ })
+}
+
+fn decimal_precision(unscaled: i128) -> u8 {
+ let mut value = unscaled.unsigned_abs();
+ if value == 0 {
+ return 1;
+ }
+ let mut precision = 0;
+ while value > 0 {
+ precision += 1;
+ value /= 10;
+ }
+ precision
+}
+
+fn string_at(array: &dyn Array, row: usize) -> DFResult<Option<String>> {
+ if array.is_null(row) {
+ return Ok(None);
+ }
+ match array.data_type() {
+ ArrowDataType::Utf8 => Ok(Some(
+ array
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .ok_or_else(|| DataFusionError::Internal("Expected Utf8
array".to_string()))?
+ .value(row)
+ .to_string(),
+ )),
+ ArrowDataType::LargeUtf8 => Ok(Some(
+ array
+ .as_any()
+ .downcast_ref::<LargeStringArray>()
+ .ok_or_else(|| DataFusionError::Internal("Expected LargeUtf8
array".to_string()))?
+ .value(row)
+ .to_string(),
+ )),
+ ArrowDataType::Utf8View => Ok(Some(
+ array
+ .as_any()
+ .downcast_ref::<StringViewArray>()
+ .ok_or_else(|| DataFusionError::Internal("Expected Utf8View
array".to_string()))?
+ .value(row)
+ .to_string(),
+ )),
+ other => plan_err(format!("Expected string array, got {other:?}")),
+ }
+}
+
+fn variant_children(array: &dyn Array) -> DFResult<Option<(&BinaryArray,
&BinaryArray)>> {
+ let ArrowDataType::Struct(fields) = array.data_type() else {
+ return Ok(None);
+ };
+ if fields.len() != 2
+ || fields[0].name() != "value"
+ || fields[0].data_type() != &ArrowDataType::Binary
+ || fields[1].name() != "metadata"
+ || fields[1].data_type() != &ArrowDataType::Binary
+ {
+ return Ok(None);
+ }
+ let array = array
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .ok_or_else(|| DataFusionError::Internal("Expected Variant
StructArray".to_string()))?;
+ let values = array
+ .column(0)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .ok_or_else(|| {
+ DataFusionError::Internal("Expected Variant.value
BinaryArray".to_string())
+ })?;
+ let metadata = array
+ .column(1)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .ok_or_else(|| {
+ DataFusionError::Internal("Expected Variant.metadata
BinaryArray".to_string())
+ })?;
+ Ok(Some((values, metadata)))
+}
+
+fn variant_array(values: Vec<Option<GenericVariant>>) -> DFResult<ArrayRef> {
+ let len = values.len();
+ let mut value_builder = BinaryBuilder::new();
+ let mut metadata_builder = BinaryBuilder::new();
+ let mut validities = Vec::with_capacity(len);
+ for value in values {
+ match value {
+ Some(variant) => {
+ value_builder.append_value(variant.value());
+ metadata_builder.append_value(variant.metadata());
+ validities.push(true);
+ }
+ None => {
+ value_builder.append_value(&[] as &[u8]);
+ metadata_builder.append_value(&[] as &[u8]);
+ validities.push(false);
+ }
+ }
+ }
+ let nulls = if validities.iter().all(|valid| *valid) {
+ None
+ } else {
+ Some(NullBuffer::new(BooleanBuffer::from(validities)))
+ };
+ let array = StructArray::try_new(
+ variant_fields(),
+ vec![
+ Arc::new(value_builder.finish()),
+ Arc::new(metadata_builder.finish()),
+ ],
+ nulls,
+ )?;
+ Ok(Arc::new(array))
+}
+
+fn variant_arrow_type() -> ArrowDataType {
+ paimon::arrow::variant_arrow_type()
+}
+
+fn variant_fields() -> Fields {
+ match variant_arrow_type() {
+ ArrowDataType::Struct(fields) => fields,
+ _ => unreachable!("variant_arrow_type must be a struct"),
+ }
+}
+
+fn null_array(data_type: &ArrowDataType, len: usize) -> ArrayRef {
+ datafusion::arrow::array::new_null_array(data_type, len)
+}
+
+fn invalid_cast() -> DataFusionError {
+ DataFusionError::Execution("Invalid Variant cast".to_string())
+}
+
+fn to_df_error(error: paimon::Error) -> DataFusionError {
+ DataFusionError::External(Box::new(error))
+}
+
+fn plan_err<T>(message: impl Into<String>) -> DFResult<T> {
+ Err(DataFusionError::Plan(message.into()))
+}
+
+fn internal_err<T>(message: impl Into<String>) -> DFResult<T> {
+ Err(DataFusionError::Internal(message.into()))
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use datafusion::arrow::array::{BooleanArray, Int32Array, StringArray};
+
+ async fn collect_one(sql: &str) ->
datafusion::arrow::record_batch::RecordBatch {
+ let ctx = SessionContext::new();
+ register_variant_functions(&ctx);
+ let batches = ctx.sql(sql).await.unwrap().collect().await.unwrap();
+ assert_eq!(batches.len(), 1);
+ batches.into_iter().next().unwrap()
+ }
+
+ #[tokio::test]
+ async fn parse_json_and_variant_get_scalars() {
+ let batch = collect_one(
+ r#"
+ SELECT
+
variant_get(parse_json('{"age":26,"city":"Beijing","nested":{"name":"Alice"},"arr":[1,2,3]}'),
'$.age', 'int') AS age,
+
variant_get(parse_json('{"age":26,"city":"Beijing","nested":{"name":"Alice"},"arr":[1,2,3]}'),
'$.city', 'string') AS city,
+
variant_get(parse_json('{"age":26,"city":"Beijing","nested":{"name":"Alice"},"arr":[1,2,3]}'),
'$.nested.name', 'string') AS name,
+
variant_get(parse_json('{"age":26,"city":"Beijing","nested":{"name":"Alice"},"arr":[1,2,3]}'),
'$.arr[1]', 'int') AS arr_value
+ "#,
+ )
+ .await;
+
+ assert_eq!(
+ batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 26
+ );
+ assert_eq!(
+ batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ "Beijing"
+ );
+ assert_eq!(
+ batch
+ .column(2)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap()
+ .value(0),
+ "Alice"
+ );
+ assert_eq!(
+ batch
+ .column(3)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap()
+ .value(0),
+ 2
+ );
+ }
+
+ #[tokio::test]
+ async fn variant_null_is_distinct_from_sql_null() {
+ let batch = collect_one(
+ "SELECT is_variant_null(parse_json('null')) AS variant_null,
is_variant_null(NULL) AS sql_null",
+ )
+ .await;
+ let variant_null = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<BooleanArray>()
+ .unwrap();
+ let sql_null = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<BooleanArray>()
+ .unwrap();
+ assert!(variant_null.value(0));
+ assert!(!sql_null.value(0));
+ }
+
+ #[tokio::test]
+ async fn try_functions_return_null_on_invalid_input() {
+ let batch = collect_one(
+ r#"
+ SELECT
+ try_parse_json('{bad json') AS bad_json,
+ try_variant_get(parse_json('{"age":"not an int"}'), '$.age',
'int') AS bad_cast,
+ variant_get(parse_json('{}'), '$.missing', 'int') AS missing_path
+ "#,
+ )
+ .await;
+ assert!(batch.column(0).is_null(0));
+ assert!(batch.column(1).is_null(0));
+ assert!(batch.column(2).is_null(0));
+ }
+
+ #[tokio::test]
+ async fn strict_functions_surface_errors() {
+ let ctx = SessionContext::new();
+ register_variant_functions(&ctx);
+ let err = ctx
+ .sql("SELECT parse_json('{bad json')")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap_err();
+ assert!(err.to_string().contains("Expected"));
+
+ let err = ctx
+ .sql("SELECT variant_get(parse_json('{\"age\":\"not an int\"}'),
'$.age', 'int')")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap_err();
+ assert!(err.to_string().contains("Invalid Variant cast"));
+ }
+
+ #[tokio::test]
+ async fn variant_get_rejects_non_literal_type_argument() {
+ let ctx = SessionContext::new();
+ register_variant_functions(&ctx);
+ let sql = r#"
+ SELECT variant_get(parse_json('{"age":26}'), '$.age', type_name)
+ FROM (VALUES ('int')) AS t(type_name)
+ "#;
+ let err = match ctx.sql(sql).await {
+ Ok(df) => df.collect().await.unwrap_err(),
+ Err(err) => err,
+ };
+ assert!(err
+ .to_string()
+ .contains("variant_get type argument must be a string literal"));
+ }
+}
diff --git a/crates/paimon/src/arrow/format/parquet.rs
b/crates/paimon/src/arrow/format/parquet.rs
index 982f518..579cbeb 100644
--- a/crates/paimon/src/arrow/format/parquet.rs
+++ b/crates/paimon/src/arrow/format/parquet.rs
@@ -1215,6 +1215,7 @@ fn literal_scalar_for_parquet_filter(
DataType::Time(_)
| DataType::Timestamp(_)
| DataType::LocalZonedTimestamp(_)
+ | DataType::Variant(_)
| DataType::Blob(_)
| DataType::Array(_)
| DataType::Map(_)
diff --git a/crates/paimon/src/arrow/format/row.rs
b/crates/paimon/src/arrow/format/row.rs
index 0ffcfd9..211bc4d 100644
--- a/crates/paimon/src/arrow/format/row.rs
+++ b/crates/paimon/src/arrow/format/row.rs
@@ -21,9 +21,12 @@
//! `org.apache.paimon.format.row.RowFormatWriter` in paimon-java.
use super::{FilePredicates, FormatFileReader, FormatFileWriter};
-use crate::arrow::{arrow_to_paimon_type, build_target_arrow_schema,
paimon_type_to_arrow};
+use crate::arrow::{
+ arrow_to_paimon_type, build_target_arrow_schema, is_variant_arrow_fields,
paimon_type_to_arrow,
+ variant_arrow_type,
+};
use crate::io::{FileRead, FileWrite, OutputFile};
-use crate::spec::{DataField, DataType, IntType};
+use crate::spec::{DataField, DataType, IntType, VarBinaryType, VariantType};
use crate::table::{ArrowRecordBatchStream, RowRange};
use crate::Error;
use arrow_array::builder::{
@@ -403,6 +406,7 @@ fn validate_arrow_type_for_row_field(
)
| (ArrowDataType::Date32, DataType::Date(_))
| (ArrowDataType::Time32(TimeUnit::Millisecond), DataType::Time(_)) =>
true,
+ (ArrowDataType::Struct(fields), DataType::Variant(_)) =>
is_variant_arrow_fields(fields),
(ArrowDataType::Timestamp(unit, tz), DataType::Timestamp(t)) => {
*unit == timestamp_time_unit_for_precision(t.precision()) &&
tz.is_none()
}
@@ -520,6 +524,7 @@ fn validate_supported_type(data_type: &DataType) ->
crate::Result<()> {
| DataType::VarChar(_)
| DataType::Binary(_)
| DataType::VarBinary(_)
+ | DataType::Variant(_)
| DataType::Blob(_)
| DataType::Date(_)
| DataType::Time(_)
@@ -659,6 +664,10 @@ fn write_field_value(
out,
downcast::<BinaryArray>(array, data_type)?.value(row_idx),
),
+ DataType::Variant(_) => {
+ let row = downcast::<StructArray>(array, data_type)?;
+ write_variant_struct(out, row, row_idx)?;
+ }
DataType::Date(_) => out.extend_from_slice(
&downcast::<Date32Array>(array, data_type)?
.value(row_idx)
@@ -712,6 +721,50 @@ fn write_field_value(
Ok(())
}
+fn write_variant_struct(out: &mut Vec<u8>, row: &StructArray, row_idx: usize)
-> crate::Result<()> {
+ if row.num_columns() != 2 {
+ return Err(Error::DataInvalid {
+ message: format!(".row variant expected 2 columns, got {}",
row.num_columns()),
+ source: None,
+ });
+ }
+ let value = row.column(0);
+ let metadata = row.column(1);
+ if value.is_null(row_idx) || metadata.is_null(row_idx) {
+ return Err(Error::DataInvalid {
+ message: ".row variant value/metadata children must be
non-null".to_string(),
+ source: None,
+ });
+ }
+ let value = value
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ ".row variant value must be BinaryArray, got {:?}",
+ value.data_type()
+ ),
+ source: None,
+ })?;
+ let metadata = metadata
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .ok_or_else(|| Error::DataInvalid {
+ message: format!(
+ ".row variant metadata must be BinaryArray, got {:?}",
+ metadata.data_type()
+ ),
+ source: None,
+ })?;
+
+ let value = value.value(row_idx);
+ let metadata = metadata.value(row_idx);
+ VariantType::validate_payload(value, metadata)?;
+ write_bytes(out, value);
+ write_bytes(out, metadata);
+ Ok(())
+}
+
fn write_array_slice(
out: &mut Vec<u8>,
values: &ArrayRef,
@@ -1146,6 +1199,15 @@ impl ColumnBuilder {
DataType::Binary(_) | DataType::VarBinary(_) | DataType::Blob(_)
=> {
Self::Binary(BinaryBuilder::new())
}
+ DataType::Variant(_) => Self::Row {
+ fields: variant_arrow_fields(),
+ columns: vec![
+ ColumnBuilder::new(&variant_binary_type(), capacity)?,
+ ColumnBuilder::new(&variant_binary_type(), capacity)?,
+ ],
+ validities: Vec::with_capacity(capacity),
+ len: 0,
+ },
DataType::Date(_) =>
Self::Date(Date32Builder::with_capacity(capacity)),
DataType::Time(_) =>
Self::Time(Time32MillisecondBuilder::with_capacity(capacity)),
DataType::Timestamp(t) => timestamp_builder(t.precision(), false,
capacity),
@@ -1372,6 +1434,19 @@ impl ColumnBuilder {
push_nested_offset(offsets, key_count)?;
validities.push(true);
}
+ (
+ Self::Row {
+ columns,
+ validities,
+ len,
+ ..
+ },
+ DataType::Variant(_),
+ ) => {
+ read_variant_into(input, columns)?;
+ *len += 1;
+ validities.push(true);
+ }
(
Self::Row {
columns,
@@ -1565,6 +1640,44 @@ fn read_struct_into(
Ok(())
}
+fn read_variant_into(
+ input: &mut BlockInput<'_>,
+ builders: &mut [ColumnBuilder],
+) -> crate::Result<()> {
+ if builders.len() != 2 {
+ return Err(Error::DataInvalid {
+ message: format!(
+ ".row variant reader expected 2 builders, got {}",
+ builders.len()
+ ),
+ source: None,
+ });
+ }
+ let value = input.read_bytes()?;
+ let metadata = input.read_bytes()?;
+ VariantType::validate_payload(&value, &metadata)?;
+ append_binary_value(&mut builders[0], value, "Variant.value")?;
+ append_binary_value(&mut builders[1], metadata, "Variant.metadata")?;
+ Ok(())
+}
+
+fn append_binary_value(
+ builder: &mut ColumnBuilder,
+ value: Vec<u8>,
+ label: &str,
+) -> crate::Result<()> {
+ match builder {
+ ColumnBuilder::Binary(b) => {
+ b.append_value(value);
+ Ok(())
+ }
+ _ => Err(Error::DataInvalid {
+ message: format!(".row {label} builder must be Binary"),
+ source: None,
+ }),
+ }
+}
+
fn arrow_fields_for_row(fields: &[DataField]) -> crate::Result<Fields> {
let fields = fields
.iter()
@@ -1579,6 +1692,20 @@ fn arrow_fields_for_row(fields: &[DataField]) ->
crate::Result<Fields> {
Ok(fields.into())
}
+fn variant_arrow_fields() -> Fields {
+ match variant_arrow_type() {
+ ArrowDataType::Struct(fields) => fields,
+ _ => unreachable!("variant_arrow_type always returns a Struct"),
+ }
+}
+
+fn variant_binary_type() -> DataType {
+ DataType::VarBinary(
+ VarBinaryType::try_new(false, VarBinaryType::MAX_LENGTH)
+ .expect("variant binary child length is valid"),
+ )
+}
+
fn map_entries_field(
key_type: &DataType,
value_type: &DataType,
@@ -2187,8 +2314,9 @@ mod tests {
use crate::spec::{
ArrayType, BigIntType, BooleanType, DataType, DateType, DecimalType,
DoubleType, FloatType,
IntType, MapType, MultisetType, RowType, TimeType, TimestampType,
VarBinaryType,
- VarCharType,
+ VarCharType, VariantType,
};
+ use crate::variant::GenericVariant;
use futures::TryStreamExt;
use std::ops::Range;
use std::sync::atomic::{AtomicUsize, Ordering};
@@ -2342,6 +2470,125 @@ mod tests {
assert_eq!(names.value(1), "ccc");
}
+ #[tokio::test]
+ async fn row_writer_reader_roundtrip_variant() {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let path = "memory:/row-variant/data.row";
+ let output = file_io.new_output(path).unwrap();
+ let fields = vec![
+ DataField::new(0, "id".to_string(), DataType::Int(IntType::new())),
+ DataField::new(
+ 1,
+ "payload".to_string(),
+ DataType::Variant(VariantType::new()),
+ ),
+ ];
+ let schema = build_target_arrow_schema(&fields).unwrap();
+ let variant_fields = match variant_arrow_type() {
+ ArrowDataType::Struct(fields) => fields,
+ other => panic!("expected variant Struct, got {other:?}"),
+ };
+ let first_variant = GenericVariant::parse_json("2").unwrap();
+ let second_variant = GenericVariant::parse_json(r#"{"a":3}"#).unwrap();
+ let variant_array = StructArray::try_new(
+ variant_fields,
+ vec![
+ Arc::new(BinaryArray::from(vec![
+ Some(first_variant.value()),
+ Some(second_variant.value()),
+ ])) as ArrayRef,
+ Arc::new(BinaryArray::from(vec![
+ Some(first_variant.metadata()),
+ Some(second_variant.metadata()),
+ ])) as ArrayRef,
+ ],
+ None,
+ )
+ .unwrap();
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])),
+ Arc::new(variant_array),
+ ],
+ )
+ .unwrap();
+ let mut writer = RowFormatWriter::new(&output, schema, fields.clone(),
1)
+ .await
+ .unwrap();
+ writer.write(&batch).await.unwrap();
+ Box::new(writer).close().await.unwrap();
+
+ let bytes = file_io.new_input(path).unwrap().read().await.unwrap();
+ let batches = RowFormatReader
+ .read_batch_stream(
+ Box::new(BytesFileRead(bytes.clone())),
+ bytes.len() as u64,
+ &fields,
+ None,
+ Some(8),
+ None,
+ )
+ .await
+ .unwrap()
+ .try_collect::<Vec<_>>()
+ .await
+ .unwrap();
+
+ assert_eq!(batches.len(), 1);
+ let payload = batches[0]
+ .column(1)
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .unwrap();
+ let value = payload
+ .column(0)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ let metadata = payload
+ .column(1)
+ .as_any()
+ .downcast_ref::<BinaryArray>()
+ .unwrap();
+ assert_eq!(value.value(0), first_variant.value());
+ assert_eq!(metadata.value(0), first_variant.metadata());
+ assert_eq!(value.value(1), second_variant.value());
+ assert_eq!(metadata.value(1), second_variant.metadata());
+ }
+
+ #[test]
+ fn row_variant_field_encoding_matches_java() {
+ let fields = match variant_arrow_type() {
+ ArrowDataType::Struct(fields) => fields,
+ other => panic!("expected variant Struct, got {other:?}"),
+ };
+ let variant = GenericVariant::parse_json("2").unwrap();
+ let variant_array = StructArray::try_new(
+ fields,
+ vec![
+ Arc::new(BinaryArray::from(vec![Some(variant.value())])) as
ArrayRef,
+ Arc::new(BinaryArray::from(vec![Some(variant.metadata())])) as
ArrayRef,
+ ],
+ None,
+ )
+ .unwrap();
+ let array = Arc::new(variant_array) as ArrayRef;
+ let mut encoded = Vec::new();
+ write_field_value(
+ &mut encoded,
+ &array,
+ 0,
+ &DataType::Variant(VariantType::new()),
+ )
+ .unwrap();
+
+ let mut expected = Vec::new();
+ write_bytes(&mut expected, variant.value());
+ write_bytes(&mut expected, variant.metadata());
+ assert_eq!(encoded, expected);
+ }
+
#[tokio::test]
async fn row_reader_selection_across_blocks() {
let fields = vec to an Arrow
[`DataType`](ArrowDataType).
pub fn paimon_type_to_arrow(dt: &PaimonDataType) ->
crate::Result<ArrowDataType> {
Ok(match dt {
@@ -42,6 +46,7 @@ pub fn paimon_type_to_arrow(dt: &PaimonDataType) ->
crate::Result<ArrowDataType>
PaimonDataType::Binary(_) | PaimonDataType::VarBinary(_) |
PaimonDataType::Blob(_) => {
ArrowDataType::Binary
}
+ PaimonDataType::Variant(_) => variant_arrow_type(),
PaimonDataType::Date(_) => ArrowDataType::Date32,
PaimonDataType::Time(_) =>
ArrowDataType::Time32(TimeUnit::Millisecond),
PaimonDataType::Timestamp(t) => {
@@ -228,6 +233,11 @@ pub fn arrow_to_paimon_type(
})
}
ArrowDataType::Struct(fields) => {
+ if is_variant_arrow_fields(fields) &&
has_variant_arrow_field_ids(fields) {
+ return Ok(PaimonDataType::Variant(VariantType::with_nullable(
+ nullable,
+ )));
+ }
let field_slice: Vec<ArrowField> = fields.iter().map(|f|
f.as_ref().clone()).collect();
let paimon_fields = arrow_fields_to_paimon(&field_slice)?;
Ok(PaimonDataType::Row(RowType::with_nullable(
@@ -252,6 +262,51 @@ pub fn arrow_to_paimon_type(
}
}
+pub fn variant_arrow_type() -> ArrowDataType {
+ ArrowDataType::Struct(
+ vec![
+ arrow_field_with_paimon_id("value", ArrowDataType::Binary, false,
0),
+ arrow_field_with_paimon_id("metadata", ArrowDataType::Binary,
false, 1),
+ ]
+ .into(),
+ )
+}
+
+pub(crate) fn is_variant_arrow_fields(fields: &arrow_schema::Fields) -> bool {
+ fields.len() == 2
+ && fields[0].name() == "value"
+ && fields[0].data_type() == &ArrowDataType::Binary
+ && !fields[0].is_nullable()
+ && fields[1].name() == "metadata"
+ && fields[1].data_type() == &ArrowDataType::Binary
+ && !fields[1].is_nullable()
+}
+
+fn has_variant_arrow_field_ids(fields: &arrow_schema::Fields) -> bool {
+ fields.len() == 2
+ && arrow_field_id(&fields[0]) == Some(0)
+ && arrow_field_id(&fields[1]) == Some(1)
+}
+
+fn arrow_field_id(field: &ArrowField) -> Option<i32> {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)?
+ .parse()
+ .ok()
+}
+
+fn arrow_field_with_paimon_id(
+ name: impl Into<String>,
+ data_type: ArrowDataType,
+ nullable: bool,
+ id: i32,
+) -> ArrowField {
+ let mut metadata = HashMap::new();
+ metadata.insert(PARQUET_FIELD_ID_META_KEY.to_string(), id.to_string());
+ ArrowField::new(name, data_type, nullable).with_metadata(metadata)
+}
+
/// Convert Arrow fields to Paimon [`DataField`]s with auto-assigned IDs
starting from 0.
pub fn arrow_fields_to_paimon(fields: &[ArrowField]) ->
crate::Result<Vec<DataField>> {
fields
@@ -270,10 +325,11 @@ pub fn build_target_arrow_schema(fields: &[DataField]) ->
crate::Result<Arc<Arro
.iter()
.map(|f| {
let arrow_type = paimon_type_to_arrow(f.data_type())?;
- Ok(ArrowField::new(
+ Ok(arrow_field_with_paimon_id(
f.name(),
arrow_type,
f.data_type().is_nullable(),
+ f.id(),
))
})
.collect::<crate::Result<Vec<_>>>()?;
@@ -379,6 +435,54 @@ mod tests {
assert_arrow_to_paimon(&ArrowDataType::Binary, true, &varbinary);
}
+ #[test]
+ fn test_variant_roundtrip() {
+ let variant = PaimonDataType::Variant(VariantType::new());
+ let arrow = variant_arrow_type();
+ assert_paimon_to_arrow(&variant, &arrow);
+ assert_arrow_to_paimon(&arrow, true, &variant);
+ }
+
+ #[test]
+ fn test_plain_value_metadata_struct_stays_row() {
+ let arrow = ArrowDataType::Struct(
+ vec![
+ ArrowField::new("value", ArrowDataType::Binary, false),
+ ArrowField::new("metadata", ArrowDataType::Binary, false),
+ ]
+ .into(),
+ );
+ let paimon = arrow_to_paimon_type(&arrow, true).unwrap();
+
+ assert!(matches!(paimon, PaimonDataType::Row(_)));
+ }
+
+ #[test]
+ fn test_variant_arrow_field_ids_for_parquet() {
+ let schema = build_target_arrow_schema(&[DataField::new(
+ 7,
+ "payload".to_string(),
+ PaimonDataType::Variant(VariantType::new()),
+ )])
+ .unwrap();
+ let field = schema.field(0);
+ assert_eq!(
+ field.metadata().get(PARQUET_FIELD_ID_META_KEY),
+ Some(&"7".to_string())
+ );
+ let ArrowDataType::Struct(fields) = field.data_type() else {
+ panic!("expected variant Struct");
+ };
+ assert_eq!(
+ fields[0].metadata().get(PARQUET_FIELD_ID_META_KEY),
+ Some(&"0".to_string())
+ );
+ assert_eq!(
+ fields[1].metadata().get(PARQUET_FIELD_ID_META_KEY),
+ Some(&"1".to_string())
+ );
+ }
+
#[test]
fn test_timestamp_roundtrip() {
// millisecond precision
diff --git a/crates/paimon/src/btree/key_serde.rs
b/crates/paimon/src/btree/key_serde.rs
index 2d7b37e..22d7254 100644
--- a/crates/paimon/src/btree/key_serde.rs
+++ b/crates/paimon/src/btree/key_serde.rs
@@ -20,7 +20,7 @@
//! Reference:
[org.apache.paimon.globalindex.btree.KeySerializer](https://github.com/apache/paimon/blob/master/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/KeySerializer.java)
use crate::btree::var_len::{decode_var_int_from_slice, encode_var_int};
-use crate::spec::{DataType, Datum};
+use crate::spec::{DataType, Datum, VariantType};
use std::cmp::Ordering;
/// Timestamp precision <= 3 is compact (millis only).
@@ -149,5 +149,14 @@ pub fn serialize_datum(datum: &Datum, data_type:
&DataType) -> Vec<u8> {
}
}
Datum::Bytes(v) => v.clone(),
+ Datum::Variant { value, metadata } => {
+ VariantType::validate_payload(value, metadata)
+ .expect("invalid Variant payload for BTree key");
+ let mut bytes = Vec::with_capacity(4 + value.len() +
metadata.len());
+ bytes.extend_from_slice(&(value.len() as u32).to_le_bytes());
+ bytes.extend_from_slice(value);
+ bytes.extend_from_slice(metadata);
+ bytes
+ }
}
}
diff --git a/crates/paimon/src/lib.rs b/crates/paimon/src/lib.rs
index 8c26fe6..da34c7d 100644
--- a/crates/paimon/src/lib.rs
+++ b/crates/paimon/src/lib.rs
@@ -37,6 +37,7 @@ pub mod spec;
pub mod table;
#[cfg(feature = "fulltext")]
pub mod tantivy;
+pub mod variant;
pub mod vector_search;
pub mod vindex;
diff --git a/crates/paimon/src/spec/binary_row.rs
b/crates/paimon/src/spec/binary_row.rs
index 145f64c..cc6601b 100644
--- a/crates/paimon/src/spec/binary_row.rs
+++ b/crates/paimon/src/spec/binary_row.rs
@@ -19,7 +19,7 @@
//! and BinaryRowBuilder for constructing BinaryRow instances.
use crate::spec::murmur_hash::hash_by_words;
-use crate::spec::{DataType, Datum};
+use crate::spec::{DataType, Datum, VariantType};
use arrow_array::RecordBatch;
use serde::{Deserialize, Serialize};
use std::sync::LazyLock;
@@ -343,6 +343,10 @@ impl BinaryRow {
DataType::Binary(_) | DataType::VarBinary(_) => {
Datum::Bytes(self.get_binary(pos)?.to_vec())
}
+ DataType::Variant(_) => {
+ let (value, metadata) =
decode_variant_bytes(self.get_binary(pos)?)?;
+ Datum::Variant { value, metadata }
+ }
DataType::Decimal(dt) => {
let unscaled = self.get_decimal_unscaled(pos, dt.precision())?;
Datum::Decimal {
@@ -647,6 +651,11 @@ impl BinaryRowBuilder {
self.write_binary(pos, b);
}
}
+ Datum::Variant { value, metadata } => {
+ let bytes = encode_variant_bytes(value, metadata)
+ .expect("invalid Variant payload for BinaryRow");
+ self.write_binary(pos, &bytes);
+ }
}
}
}
@@ -772,6 +781,7 @@ pub fn extract_datum_from_arrow(
.ok_or_else(|| type_mismatch_err("Binary", col_idx))?;
Datum::Bytes(arr.value(row_idx).to_vec())
}
+ DataType::Variant(_) => extract_variant_datum_from_arrow(col, row_idx,
col_idx)?,
DataType::Timestamp(ts) => {
if ts.precision() <= 3 {
let arr = col
@@ -829,6 +839,104 @@ pub fn extract_datum_from_arrow(
Ok(Some(datum))
}
+fn encode_variant_bytes(value: &[u8], metadata: &[u8]) ->
crate::Result<Vec<u8>> {
+ VariantType::validate_payload(value, metadata)?;
+ let mut bytes = Vec::with_capacity(4 + value.len() + metadata.len());
+ bytes.extend_from_slice(&(value.len() as u32).to_le_bytes());
+ bytes.extend_from_slice(value);
+ bytes.extend_from_slice(metadata);
+ Ok(bytes)
+}
+
+fn decode_variant_bytes(bytes: &[u8]) -> crate::Result<(Vec<u8>, Vec<u8>)> {
+ if bytes.len() < 4 {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Variant bytes too short: {} bytes", bytes.len()),
+ source: None,
+ });
+ }
+ let value_len = u32::from_le_bytes(bytes[0..4].try_into().unwrap()) as
usize;
+ let value_end = 4usize
+ .checked_add(value_len)
+ .ok_or_else(|| crate::Error::DataInvalid {
+ message: "Variant value length overflows".to_string(),
+ source: None,
+ })?;
+ if value_end > bytes.len() {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Variant value length {value_len} exceeds payload length {}",
+ bytes.len()
+ ),
+ source: None,
+ });
+ }
+ let value = bytes[4..value_end].to_vec();
+ let metadata = bytes[value_end..].to_vec();
+ VariantType::validate_payload(&value, &metadata)?;
+ Ok((value, metadata))
+}
+
+fn extract_variant_datum_from_arrow(
+ col: &std::sync::Arc<dyn arrow_array::Array>,
+ row_idx: usize,
+ col_idx: usize,
+) -> crate::Result<Datum> {
+ use arrow_array::Array;
+
+ let arr = col
+ .as_any()
+ .downcast_ref::<arrow_array::StructArray>()
+ .ok_or_else(|| type_mismatch_err("Variant", col_idx))?;
+ validate_variant_struct_array(arr, col_idx)?;
+ let value = arr.column(0);
+ let metadata = arr.column(1);
+ if value.is_null(row_idx) || metadata.is_null(row_idx) {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Variant Arrow struct at column {col_idx} has
null child value"),
+ source: None,
+ });
+ }
+ let value = value
+ .as_any()
+ .downcast_ref::<arrow_array::BinaryArray>()
+ .ok_or_else(|| type_mismatch_err("Variant.value", col_idx))?
+ .value(row_idx)
+ .to_vec();
+ let metadata = metadata
+ .as_any()
+ .downcast_ref::<arrow_array::BinaryArray>()
+ .ok_or_else(|| type_mismatch_err("Variant.metadata", col_idx))?
+ .value(row_idx)
+ .to_vec();
+ VariantType::validate_payload(&value, &metadata)?;
+ Ok(Datum::Variant { value, metadata })
+}
+
+fn validate_variant_struct_array(
+ arr: &arrow_array::StructArray,
+ col_idx: usize,
+) -> crate::Result<()> {
+ if arr.num_columns() != 2 {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "Variant Arrow struct at column {col_idx} must have 2 fields,
got {}",
+ arr.num_columns()
+ ),
+ source: None,
+ });
+ }
+ arr.column(0)
+ .as_any()
+ .downcast_ref::<arrow_array::BinaryArray>()
+ .ok_or_else(|| type_mismatch_err("Variant.value", col_idx))?;
+ arr.column(1)
+ .as_any()
+ .downcast_ref::<arrow_array::BinaryArray>()
+ .ok_or_else(|| type_mismatch_err("Variant.metadata", col_idx))?;
+ Ok(())
+}
+
fn type_mismatch_err(expected: &str, col_idx: usize) -> crate::Error {
crate::Error::DataInvalid {
message: format!(
@@ -858,6 +966,7 @@ enum TypedColumn<'a> {
Date32(&'a arrow_array::Date32Array),
Decimal128(&'a arrow_array::Decimal128Array, u32, u32), // (array,
precision, scale)
Binary(&'a arrow_array::BinaryArray),
+ Variant(&'a arrow_array::StructArray),
TimestampMs(&'a arrow_array::TimestampMillisecondArray),
TimestampUs(&'a arrow_array::TimestampMicrosecondArray),
}
@@ -943,6 +1052,14 @@ fn downcast_columns<'a>(
.downcast_ref()
.ok_or_else(|| type_mismatch_err("Binary",
col_idx))?,
),
+ DataType::Variant(_) => {
+ let arr = col
+ .as_any()
+ .downcast_ref()
+ .ok_or_else(|| type_mismatch_err("Variant",
col_idx))?;
+ validate_variant_struct_array(arr, col_idx)?;
+ TypedColumn::Variant(arr)
+ }
DataType::Timestamp(ts) => {
if ts.precision() <= 3 {
TypedColumn::TimestampMs(
@@ -990,7 +1107,7 @@ fn write_typed_value(
row_idx: usize,
typed_col: &TypedColumn,
_data_type: &DataType,
-) {
+) -> crate::Result<()> {
use arrow_array::Array;
match typed_col {
TypedColumn::Boolean(arr) => {
@@ -1109,6 +1226,32 @@ fn write_typed_value(
}
}
}
+ TypedColumn::Variant(arr) => {
+ if arr.is_null(row_idx) {
+ builder.set_null_at(pos);
+ } else {
+ let value = arr.column(0);
+ let metadata = arr.column(1);
+ if value.is_null(row_idx) || metadata.is_null(row_idx) {
+ return Err(crate::Error::DataInvalid {
+ message: "Variant Arrow struct has null child
value".to_string(),
+ source: None,
+ });
+ } else if let (Some(value), Some(metadata)) = (
+ value.as_any().downcast_ref::<arrow_array::BinaryArray>(),
+
metadata.as_any().downcast_ref::<arrow_array::BinaryArray>(),
+ ) {
+ let bytes =
+ encode_variant_bytes(value.value(row_idx),
metadata.value(row_idx))?;
+ builder.write_binary(pos, &bytes);
+ } else {
+ return Err(crate::Error::DataInvalid {
+ message: "Variant Arrow struct children must be
BinaryArray".to_string(),
+ source: None,
+ });
+ }
+ }
+ }
TypedColumn::TimestampMs(arr) => {
if arr.is_null(row_idx) {
builder.set_null_at(pos);
@@ -1127,6 +1270,7 @@ fn write_typed_value(
}
}
}
+ Ok(())
}
/// Build BinaryRows for all rows in the batch for the given field indices.
@@ -1144,7 +1288,7 @@ pub(crate) fn batch_build_binary_rows(
for row_idx in 0..num_rows {
let mut builder = BinaryRowBuilder::new(arity);
for (pos, (typed_col, field)) in typed_columns.iter().enumerate() {
- write_typed_value(&mut builder, pos, row_idx, typed_col,
field.data_type());
+ write_typed_value(&mut builder, pos, row_idx, typed_col,
field.data_type())?;
}
rows.push(builder.build());
}
@@ -1180,6 +1324,7 @@ pub fn batch_hash_codes(
#[cfg(test)]
mod tests {
use super::*;
+ use crate::variant::GenericVariant;
#[test]
fn test_empty_binary_row() {
@@ -1338,6 +1483,19 @@ mod tests {
assert_eq!(row.get_binary(0).unwrap(), &[0x00, 0x01, 0x02, 0x03]);
}
+ #[test]
+ fn test_variant_datum_roundtrip() {
+ let data_type = DataType::Variant(crate::spec::VariantType::new());
+ let variant = GenericVariant::parse_json(r#"{"a":1}"#).unwrap();
+ let datum = Datum::Variant {
+ value: variant.value().to_vec(),
+ metadata: variant.metadata().to_vec(),
+ };
+ let row = BinaryRow::from_datums(&[(Some(&datum), &data_type)]);
+
+ assert_eq!(row.get_datum(0, &data_type).unwrap(), Some(datum));
+ }
+
#[test]
fn test_mixed_types_partition_row() {
let mut builder = BinaryRowBuilder::new(2);
diff --git a/crates/paimon/src/spec/data_type_casts.rs
b/crates/paimon/src/spec/data_type_casts.rs
index c44c4a4..c445204 100644
--- a/crates/paimon/src/spec/data_type_casts.rs
+++ b/crates/paimon/src/spec/data_type_casts.rs
@@ -162,6 +162,7 @@ fn explicit_cast_supported(source: &DataType, target:
&DataType) -> bool {
is_datetime(source) || is_character_string(source) ||
is_numeric(source)
}
DataType::Blob(_)
+ | DataType::Variant(_)
| DataType::Array(_)
| DataType::Map(_)
| DataType::Multiset(_)
diff --git a/crates/paimon/src/spec/partition_utils.rs
b/crates/paimon/src/spec/partition_utils.rs
index db0f2c2..3db1884 100644
--- a/crates/paimon/src/spec/partition_utils.rs
+++ b/crates/paimon/src/spec/partition_utils.rs
@@ -265,6 +265,7 @@ fn format_partition_value(
| DataType::Double(_)
| DataType::Binary(_)
| DataType::VarBinary(_)
+ | DataType::Variant(_)
| DataType::Blob(_)
| DataType::Array(_)
| DataType::Map(_)
diff --git a/crates/paimon/src/spec/predicate.rs
b/crates/paimon/src/spec/predicate.rs
index a419f77..5fb641d 100644
--- a/crates/paimon/src/spec/predicate.rs
+++ b/crates/paimon/src/spec/predicate.rs
@@ -76,6 +76,10 @@ pub enum Datum {
scale: u32,
},
Bytes(Vec<u8>),
+ Variant {
+ value: Vec<u8>,
+ metadata: Vec<u8>,
+ },
}
impl fmt::Display for Datum {
@@ -97,6 +101,14 @@ impl fmt::Display for Datum {
unscaled, scale, ..
} => write!(f, "DEC({unscaled},s{scale})"),
Self::Bytes(v) => write!(f, "BYTES(len={})", v.len()),
+ Self::Variant { value, metadata } => {
+ write!(
+ f,
+ "VARIANT(value_len={},metadata_len={})",
+ value.len(),
+ metadata.len()
+ )
+ }
}
}
}
@@ -162,6 +174,22 @@ pub(crate) fn datum_cmp(lhs: &Datum, rhs: &Datum) ->
Option<Ordering> {
},
) => decimal_cmp(*ua, *sa, *ub, *sb),
(Datum::Bytes(a), Datum::Bytes(b)) => Some(java_bytes_cmp(a, b)),
+ (
+ Datum::Variant {
+ value: va,
+ metadata: ma,
+ },
+ Datum::Variant {
+ value: vb,
+ metadata: mb,
+ },
+ ) => {
+ if va == vb && ma == mb {
+ Some(Ordering::Equal)
+ } else {
+ None
+ }
+ }
_ => None,
}
}
@@ -864,6 +892,7 @@ fn validate_datum_matches_type(datum: &Datum, data_type:
&DataType) -> Result<()
| (Datum::Decimal { .. }, DataType::Decimal(_))
| (Datum::Bytes(_), DataType::Binary(_))
| (Datum::Bytes(_), DataType::VarBinary(_))
+ | (Datum::Variant { .. }, DataType::Variant(_))
);
if !ok {
return Err(Error::ConfigInvalid {
diff --git a/crates/paimon/src/spec/types.rs b/crates/paimon/src/spec/types.rs
index 7269aff..b1e5ebb 100644
--- a/crates/paimon/src/spec/types.rs
+++ b/crates/paimon/src/spec/types.rs
@@ -76,6 +76,8 @@ pub enum DataType {
Binary(BinaryType),
/// Data type of a variable-length binary string (=a sequence of bytes).
VarBinary(VarBinaryType),
+ /// Data type of semi-structured values encoded as Variant value +
metadata.
+ Variant(VariantType),
/// Data type of binary large object.
Blob(BlobType),
/// Data type of a fixed-length character string.
@@ -139,6 +141,7 @@ impl DataType {
DataType::Float(v) => v.nullable,
DataType::Binary(v) => v.nullable,
DataType::VarBinary(v) => v.nullable,
+ DataType::Variant(v) => v.nullable,
DataType::Blob(v) => v.nullable,
DataType::Char(v) => v.nullable,
DataType::VarChar(v) => v.nullable,
@@ -176,6 +179,7 @@ impl DataType {
DataType::VarBinary(v) => {
DataType::VarBinary(VarBinaryType::try_new(nullable,
v.length())?)
}
+ DataType::Variant(_) =>
DataType::Variant(VariantType::with_nullable(nullable)),
DataType::Blob(_) =>
DataType::Blob(BlobType::with_nullable(nullable)),
DataType::Char(v) =>
DataType::Char(CharType::with_nullable(nullable, v.length())?),
DataType::VarChar(v) => {
@@ -632,6 +636,44 @@ impl BlobType {
}
}
+/// VariantType for paimon.
+///
+/// Data type of semi-structured values encoded as two binary buffers: `value`
+/// and `metadata`.
+///
+/// Impl Reference:
<https://github.com/apache/paimon/blob/master/paimon-api/src/main/java/org/apache/paimon/types/VariantType.java>.
+#[serde_as]
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct VariantType {
+ #[serde_as(as =
"FromInto<serde_utils::NullableType<serde_utils::VARIANT>>")]
+ nullable: bool,
+}
+
+impl Default for VariantType {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl VariantType {
+ pub fn new() -> Self {
+ Self::with_nullable(true)
+ }
+
+ pub fn with_nullable(nullable: bool) -> Self {
+ Self { nullable }
+ }
+
+ pub fn family(&self) -> DataTypeFamily {
+ DataTypeFamily::PREDEFINED
+ }
+
+ pub(crate) fn validate_payload(value: &[u8], metadata: &[u8]) ->
crate::Result<()> {
+ crate::variant::validate_payload(value, metadata)
+ }
+}
+
/// CharType for paimon.
///
/// Data type of a fixed-length character string.
@@ -1732,6 +1774,11 @@ mod serde_utils {
const NAME: &'static str = "BLOB";
}
+ pub struct VARIANT;
+ impl DataTypeName for VARIANT {
+ const NAME: &'static str = "VARIANT";
+ }
+
pub struct BINARY;
impl DataTypeName for BINARY {
const NAME: &'static str = "BINARY";
@@ -2171,6 +2218,14 @@ mod tests {
length: 233,
}),
),
+ (
+ "variant_type",
+ DataType::Variant(VariantType { nullable: false }),
+ ),
+ (
+ "variant_type_nullable",
+ DataType::Variant(VariantType { nullable: true }),
+ ),
(
"varchar_type",
DataType::VarChar(VarCharType {
diff --git a/crates/paimon/src/variant.rs b/crates/paimon/src/variant.rs
new file mode 100644
index 0000000..537a51a
--- /dev/null
+++ b/crates/paimon/src/variant.rs
@@ -0,0 +1,1807 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Java-compatible helpers for Paimon Variant values.
+
+use std::collections::{HashMap, HashSet};
+
+use crate::{Error, Result};
+use base64::{engine::general_purpose, Engine as _};
+
+const BASIC_TYPE_BITS: u8 = 2;
+const BASIC_TYPE_MASK: u8 = 0x3;
+const TYPE_INFO_MASK: u8 = 0x3f;
+const MAX_SHORT_STR_SIZE: usize = 0x3f;
+
+const PRIMITIVE: u8 = 0;
+const SHORT_STR: u8 = 1;
+const OBJECT: u8 = 2;
+const ARRAY: u8 = 3;
+
+const NULL: u8 = 0;
+const TRUE: u8 = 1;
+const FALSE: u8 = 2;
+const INT1: u8 = 3;
+const INT2: u8 = 4;
+const INT4: u8 = 5;
+const INT8: u8 = 6;
+const DOUBLE: u8 = 7;
+const DECIMAL4: u8 = 8;
+const DECIMAL8: u8 = 9;
+const DECIMAL16: u8 = 10;
+const DATE: u8 = 11;
+const TIMESTAMP: u8 = 12;
+const TIMESTAMP_NTZ: u8 = 13;
+const FLOAT: u8 = 14;
+const BINARY: u8 = 15;
+const LONG_STR: u8 = 16;
+const UUID: u8 = 20;
+
+const VERSION: u8 = 1;
+const VERSION_MASK: u8 = 0x0f;
+
+const U8_MAX: usize = 0xff;
+const U16_MAX: usize = 0xffff;
+const U24_MAX: usize = 0xff_ffff;
+const U32_SIZE: usize = 4;
+const SIZE_LIMIT: usize = 128 * 1024 * 1024;
+const MAX_NESTING_DEPTH: usize = 1000;
+
+const MAX_DECIMAL4_PRECISION: u8 = 9;
+const MAX_DECIMAL8_PRECISION: u8 = 18;
+const MAX_DECIMAL16_PRECISION: u8 = 38;
+const BINARY_SEARCH_THRESHOLD: usize = 32;
+
+/// An owned Paimon Variant value encoded as Java-compatible `value` and
`metadata` buffers.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct GenericVariant {
+ value: Vec<u8>,
+ metadata: Vec<u8>,
+}
+
+impl GenericVariant {
+ /// Parse a JSON string into the same binary Variant representation
produced by Java Paimon.
+ pub fn parse_json(json: &str) -> Result<Self> {
+ JsonVariantParser::new(json).parse()
+ }
+
+ /// Build an owned Variant from existing buffers after validating the
metadata header.
+ pub fn from_parts(value: Vec<u8>, metadata: Vec<u8>) -> Result<Self> {
+ VariantRef::new(&value, &metadata, 0)?;
+ Ok(Self { value, metadata })
+ }
+
+ pub fn value(&self) -> &[u8] {
+ &self.value
+ }
+
+ pub fn metadata(&self) -> &[u8] {
+ &self.metadata
+ }
+
+ pub fn as_ref(&self) -> Result<VariantRef<'_>> {
+ VariantRef::new(&self.value, &self.metadata, 0)
+ }
+
+ pub fn is_variant_null(&self) -> Result<bool> {
+ self.as_ref()?.is_null()
+ }
+
+ pub fn get_path(&self, path: &str) -> Result<Option<VariantRef<'_>>> {
+ self.as_ref()?.get_path(path)
+ }
+
+ pub fn to_json(&self) -> Result<String> {
+ self.as_ref()?.to_json()
+ }
+}
+
+/// A borrowed view into a Variant buffer. Sub-values share the original
metadata dictionary.
+#[derive(Clone, Copy, Debug)]
+pub struct VariantRef<'a> {
+ value: &'a [u8],
+ metadata: &'a [u8],
+ pos: usize,
+}
+
+impl<'a> VariantRef<'a> {
+ pub fn new(value: &'a [u8], metadata: &'a [u8], pos: usize) ->
Result<Self> {
+ validate_payload(value, metadata)?;
+ Self::new_at(value, metadata, pos)
+ }
+
+ fn new_at(value: &'a [u8], metadata: &'a [u8], pos: usize) -> Result<Self>
{
+ check_index(value, pos)?;
+ Ok(Self {
+ value,
+ metadata,
+ pos,
+ })
+ }
+
+ pub fn kind(&self) -> Result<VariantKind> {
+ value_kind(self.value, self.pos)
+ }
+
+ pub fn is_null(&self) -> Result<bool> {
+ Ok(self.kind()? == VariantKind::Null)
+ }
+
+ pub fn value_slice(&self) -> Result<&'a [u8]> {
+ let size = value_size(self.value, self.pos)?;
+ check_index(self.value, self.pos + size - 1)?;
+ Ok(&self.value[self.pos..self.pos + size])
+ }
+
+ pub fn metadata(&self) -> &'a [u8] {
+ self.metadata
+ }
+
+ pub fn to_owned_variant(&self) -> Result<GenericVariant> {
+ Ok(GenericVariant {
+ value: self.value_slice()?.to_vec(),
+ metadata: self.metadata.to_vec(),
+ })
+ }
+
+ pub fn get_path(&self, path: &str) -> Result<Option<VariantRef<'a>>> {
+ let mut current = *self;
+ for segment in parse_path(path)? {
+ match (segment, current.kind()?) {
+ (PathSegment::Key(key), VariantKind::Object) => {
+ let Some(next) = current.get_field_by_key(&key)? else {
+ return Ok(None);
+ };
+ current = next;
+ }
+ (PathSegment::Index(index), VariantKind::Array) => {
+ let Some(next) = current.get_element_at_index(index)? else
{
+ return Ok(None);
+ };
+ current = next;
+ }
+ _ => return Ok(None),
+ }
+ }
+ Ok(Some(current))
+ }
+
+ pub fn get_boolean(&self) -> Result<bool> {
+ get_boolean(self.value, self.pos)
+ }
+
+ pub fn get_long(&self) -> Result<i64> {
+ get_long(self.value, self.pos)
+ }
+
+ pub fn get_double(&self) -> Result<f64> {
+ get_double(self.value, self.pos)
+ }
+
+ pub fn get_float(&self) -> Result<f32> {
+ get_float(self.value, self.pos)
+ }
+
+ pub fn get_string(&self) -> Result<String> {
+ get_string(self.value, self.pos)
+ }
+
+ pub fn get_decimal(&self) -> Result<VariantDecimal> {
+ get_decimal(self.value, self.pos)
+ }
+
+ pub fn to_json(&self) -> Result<String> {
+ let mut out = String::new();
+ write_json(self.value, self.metadata, self.pos, &mut out)?;
+ Ok(out)
+ }
+
+ fn get_field_by_key(&self, key: &str) -> Result<Option<VariantRef<'a>>> {
+ let layout = object_layout(self.value, self.pos)?;
+ if layout.size < BINARY_SEARCH_THRESHOLD {
+ for i in 0..layout.size {
+ let id = read_unsigned(
+ self.value,
+ layout.id_start + layout.id_size * i,
+ layout.id_size,
+ )?;
+ if key == get_metadata_key(self.metadata, id)? {
+ let offset = read_unsigned(
+ self.value,
+ layout.offset_start + layout.offset_size * i,
+ layout.offset_size,
+ )?;
+ return VariantRef::new_at(
+ self.value,
+ self.metadata,
+ layout.data_start + offset,
+ )
+ .map(Some);
+ }
+ }
+ } else {
+ let mut low = 0usize;
+ let mut high = layout.size;
+ while low < high {
+ let mid = low + (high - low) / 2;
+ let id = read_unsigned(
+ self.value,
+ layout.id_start + layout.id_size * mid,
+ layout.id_size,
+ )?;
+ match java_string_cmp(&get_metadata_key(self.metadata, id)?,
key) {
+ std::cmp::Ordering::Less => low = mid + 1,
+ std::cmp::Ordering::Greater => high = mid,
+ std::cmp::Ordering::Equal => {
+ let offset = read_unsigned(
+ self.value,
+ layout.offset_start + layout.offset_size * mid,
+ layout.offset_size,
+ )?;
+ return VariantRef::new_at(
+ self.value,
+ self.metadata,
+ layout.data_start + offset,
+ )
+ .map(Some);
+ }
+ }
+ }
+ }
+ Ok(None)
+ }
+
+ fn get_element_at_index(&self, index: usize) ->
Result<Option<VariantRef<'a>>> {
+ let layout = array_layout(self.value, self.pos)?;
+ if index >= layout.size {
+ return Ok(None);
+ }
+ let offset = read_unsigned(
+ self.value,
+ layout.offset_start + layout.offset_size * index,
+ layout.offset_size,
+ )?;
+ VariantRef::new_at(self.value, self.metadata, layout.data_start +
offset).map(Some)
+ }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum VariantKind {
+ Object,
+ Array,
+ Null,
+ Boolean,
+ Long,
+ String,
+ Double,
+ Decimal,
+ Date,
+ Timestamp,
+ TimestampNtz,
+ Float,
+ Binary,
+ Uuid,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct VariantDecimal {
+ pub unscaled: i128,
+ pub precision: u8,
+ pub scale: i8,
+}
+
+impl VariantDecimal {
+ pub fn to_plain_string(self) -> String {
+ decimal_to_plain_string(self.unscaled, self.scale, true)
+ }
+}
+
+pub fn validate_payload(value: &[u8], metadata: &[u8]) -> Result<()> {
+ if metadata.is_empty() || (metadata[0] & VERSION_MASK) != VERSION {
+ return data_invalid("Malformed Variant metadata version");
+ }
+ if value.len() > SIZE_LIMIT || metadata.len() > SIZE_LIMIT {
+ return data_invalid("Variant value or metadata exceeds constructor
size limit");
+ }
+ validate_metadata(metadata)?;
+ let root_size = validate_value(value, metadata, 0, 0)?;
+ if root_size != value.len() {
+ return data_invalid("Malformed Variant root size");
+ }
+ Ok(())
+}
+
+fn validate_metadata(metadata: &[u8]) -> Result<()> {
+ let offset_size = metadata_offset_size(metadata)?;
+ let dict_size = read_unsigned(metadata, 1, offset_size)?;
+ let offset_start = 1 + offset_size;
+ let string_start = offset_start + (dict_size + 1) * offset_size;
+ check_range(metadata, offset_start, (dict_size + 1) * offset_size)?;
+ let string_size = read_unsigned(
+ metadata,
+ offset_start + dict_size * offset_size,
+ offset_size,
+ )?;
+ if string_start.checked_add(string_size) != Some(metadata.len()) {
+ return data_invalid("Malformed Variant metadata size");
+ }
+
+ let mut previous_offset = 0usize;
+ let mut seen = HashSet::with_capacity(dict_size);
+ for id in 0..dict_size {
+ let offset = read_unsigned(metadata, offset_start + id * offset_size,
offset_size)?;
+ let next_offset =
+ read_unsigned(metadata, offset_start + (id + 1) * offset_size,
offset_size)?;
+ if offset != previous_offset || offset > next_offset || next_offset >
string_size {
+ return data_invalid("Malformed Variant metadata offsets");
+ }
+ previous_offset = next_offset;
+ let bytes = &metadata[string_start + offset..string_start +
next_offset];
+ let key = std::str::from_utf8(bytes).map_err(|e| Error::DataInvalid {
+ message: "Malformed Variant metadata UTF-8".to_string(),
+ source: Some(Box::new(e)),
+ })?;
+ if !seen.insert(key) {
+ return data_invalid("Malformed Variant metadata duplicate key");
+ }
+ }
+ Ok(())
+}
+
+fn validate_value(value: &[u8], metadata: &[u8], pos: usize, depth: usize) ->
Result<usize> {
+ if depth > MAX_NESTING_DEPTH {
+ return data_invalid("Malformed Variant nesting depth");
+ }
+
+ match value_kind(value, pos)? {
+ VariantKind::Object => validate_object(value, metadata, pos, depth),
+ VariantKind::Array => validate_array(value, metadata, pos, depth),
+ VariantKind::Null => validate_sized_value(value, pos,
value_size(value, pos)?),
+ VariantKind::Boolean => {
+ get_boolean(value, pos)?;
+ validate_sized_value(value, pos, value_size(value, pos)?)
+ }
+ VariantKind::Long
+ | VariantKind::Date
+ | VariantKind::Timestamp
+ | VariantKind::TimestampNtz => {
+ get_long(value, pos)?;
+ validate_sized_value(value, pos, value_size(value, pos)?)
+ }
+ VariantKind::String => {
+ get_string(value, pos)?;
+ validate_sized_value(value, pos, value_size(value, pos)?)
+ }
+ VariantKind::Double => {
+ get_double(value, pos)?;
+ validate_sized_value(value, pos, value_size(value, pos)?)
+ }
+ VariantKind::Decimal => {
+ get_decimal(value, pos)?;
+ validate_sized_value(value, pos, value_size(value, pos)?)
+ }
+ VariantKind::Float => {
+ get_float(value, pos)?;
+ validate_sized_value(value, pos, value_size(value, pos)?)
+ }
+ VariantKind::Binary => {
+ get_binary(value, pos)?;
+ validate_sized_value(value, pos, value_size(value, pos)?)
+ }
+ VariantKind::Uuid => {
+ get_uuid(value, pos)?;
+ validate_sized_value(value, pos, value_size(value, pos)?)
+ }
+ }
+}
+
+fn validate_sized_value(value: &[u8], pos: usize, size: usize) ->
Result<usize> {
+ check_range(value, pos, size)?;
+ Ok(size)
+}
+
+fn validate_object(value: &[u8], metadata: &[u8], pos: usize, depth: usize) ->
Result<usize> {
+ let layout = object_layout(value, pos)?;
+ let data_size = read_unsigned(
+ value,
+ layout.offset_start + layout.size * layout.offset_size,
+ layout.offset_size,
+ )?;
+ let data_end = layout
+ .data_start
+ .checked_add(data_size)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Malformed Variant object offsets".to_string(),
+ source: None,
+ })?;
+ if data_end > value.len() {
+ return data_invalid("Malformed Variant object offsets");
+ }
+
+ let mut ranges = Vec::with_capacity(layout.size);
+ let mut previous_key: Option<String> = None;
+ for i in 0..layout.size {
+ let id = read_unsigned(value, layout.id_start + layout.id_size * i,
layout.id_size)?;
+ let key = get_metadata_key(metadata, id)?;
+ if previous_key
+ .as_deref()
+ .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);
+
+ let offset = read_unsigned(
+ value,
+ layout.offset_start + layout.offset_size * i,
+ layout.offset_size,
+ )?;
+ ranges.push(validate_child_range(
+ value,
+ metadata,
+ layout.data_start,
+ data_size,
+ offset,
+ depth + 1,
+ )?);
+ }
+
+ let final_offset = read_unsigned(
+ value,
+ layout.offset_start + layout.offset_size * layout.size,
+ layout.offset_size,
+ )?;
+ if final_offset != data_size {
+ return data_invalid("Malformed Variant object offsets");
+ }
+ validate_ranges_cover_data(&mut ranges, data_size)?;
+ Ok(layout.data_start - pos + data_size)
+}
+
+fn validate_array(value: &[u8], metadata: &[u8], pos: usize, depth: usize) ->
Result<usize> {
+ let layout = array_layout(value, pos)?;
+ let data_size = read_unsigned(
+ value,
+ layout.offset_start + layout.size * layout.offset_size,
+ layout.offset_size,
+ )?;
+ let data_end = layout
+ .data_start
+ .checked_add(data_size)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Malformed Variant array offsets".to_string(),
+ source: None,
+ })?;
+ if data_end > value.len() {
+ return data_invalid("Malformed Variant array offsets");
+ }
+
+ let mut previous_offset = 0usize;
+ for i in 0..layout.size {
+ let offset = read_unsigned(
+ value,
+ layout.offset_start + layout.offset_size * i,
+ layout.offset_size,
+ )?;
+ let next_offset = read_unsigned(
+ value,
+ layout.offset_start + layout.offset_size * (i + 1),
+ layout.offset_size,
+ )?;
+ validate_child_value(
+ value,
+ metadata,
+ layout.data_start,
+ data_size,
+ offset,
+ next_offset,
+ depth + 1,
+ )?;
+ previous_offset = validate_offset_order(previous_offset, offset,
next_offset, data_size)?;
+ }
+
+ let final_offset = read_unsigned(
+ value,
+ layout.offset_start + layout.offset_size * layout.size,
+ layout.offset_size,
+ )?;
+ if final_offset != data_size || final_offset < previous_offset {
+ return data_invalid("Malformed Variant array offsets");
+ }
+ Ok(layout.data_start - pos + data_size)
+}
+
+fn validate_child_value(
+ value: &[u8],
+ metadata: &[u8],
+ data_start: usize,
+ data_size: usize,
+ offset: usize,
+ next_offset: usize,
+ depth: usize,
+) -> Result<()> {
+ if offset > next_offset || next_offset > data_size {
+ return data_invalid("Malformed Variant offsets");
+ }
+ let child_pos = data_start
+ .checked_add(offset)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Malformed Variant offsets".to_string(),
+ source: None,
+ })?;
+ let child_size = validate_value(value, metadata, child_pos, depth)?;
+ if child_size != next_offset - offset {
+ return data_invalid("Malformed Variant child size");
+ }
+ Ok(())
+}
+
+fn validate_child_range(
+ value: &[u8],
+ metadata: &[u8],
+ data_start: usize,
+ data_size: usize,
+ offset: usize,
+ depth: usize,
+) -> Result<(usize, usize)> {
+ if offset > data_size {
+ return data_invalid("Malformed Variant offsets");
+ }
+ let child_pos = data_start
+ .checked_add(offset)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Malformed Variant offsets".to_string(),
+ source: None,
+ })?;
+ let child_size = validate_value(value, metadata, child_pos, depth)?;
+ let end = offset
+ .checked_add(child_size)
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Malformed Variant offsets".to_string(),
+ source: None,
+ })?;
+ if end > data_size {
+ return data_invalid("Malformed Variant offsets");
+ }
+ Ok((offset, end))
+}
+
+fn validate_ranges_cover_data(ranges: &mut [(usize, usize)], data_size: usize)
-> Result<()> {
+ ranges.sort_unstable_by_key(|(start, _)| *start);
+ let mut expected_start = 0usize;
+ for (start, end) in ranges {
+ if *start != expected_start || *end < *start {
+ return data_invalid("Malformed Variant offsets");
+ }
+ expected_start = *end;
+ }
+ if expected_start != data_size {
+ return data_invalid("Malformed Variant offsets");
+ }
+ Ok(())
+}
+
+fn validate_offset_order(
+ previous_offset: usize,
+ offset: usize,
+ next_offset: usize,
+ data_size: usize,
+) -> Result<usize> {
+ if offset != previous_offset || offset > next_offset || next_offset >
data_size {
+ return data_invalid("Malformed Variant offsets");
+ }
+ Ok(next_offset)
+}
+
+fn data_invalid<T>(message: impl Into<String>) -> Result<T> {
+ Err(Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ })
+}
+
+fn primitive_header(type_info: u8) -> u8 {
+ (type_info << BASIC_TYPE_BITS) | PRIMITIVE
+}
+
+fn short_str_header(size: usize) -> u8 {
+ ((size as u8) << BASIC_TYPE_BITS) | SHORT_STR
+}
+
+fn object_header(large_size: bool, id_size: usize, offset_size: usize) -> u8 {
+ (((large_size as u8) << (BASIC_TYPE_BITS + 4))
+ | (((id_size - 1) as u8) << (BASIC_TYPE_BITS + 2))
+ | (((offset_size - 1) as u8) << BASIC_TYPE_BITS))
+ | OBJECT
+}
+
+fn array_header(large_size: bool, offset_size: usize) -> u8 {
+ (((large_size as u8) << (BASIC_TYPE_BITS + 2)) | (((offset_size - 1) as
u8) << BASIC_TYPE_BITS))
+ | ARRAY
+}
+
+fn integer_size(value: usize) -> Result<usize> {
+ if value > SIZE_LIMIT {
+ return data_invalid("Variant value exceeds size limit");
+ }
+ Ok(if value <= U8_MAX {
+ 1
+ } else if value <= U16_MAX {
+ 2
+ } else if value <= U24_MAX {
+ 3
+ } else {
+ 4
+ })
+}
+
+fn check_index(bytes: &[u8], pos: usize) -> Result<()> {
+ if pos >= bytes.len() {
+ return data_invalid("Malformed Variant");
+ }
+ Ok(())
+}
+
+fn check_range(bytes: &[u8], pos: usize, len: usize) -> Result<()> {
+ if len == 0 {
+ return Ok(());
+ }
+ let end = pos
+ .checked_add(len)
+ .and_then(|v| v.checked_sub(1))
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Malformed Variant".to_string(),
+ source: None,
+ })?;
+ check_index(bytes, end)
+}
+
+fn write_le_at(bytes: &mut [u8], pos: usize, value: usize, num_bytes: usize) {
+ for i in 0..num_bytes {
+ bytes[pos + i] = ((value >> (8 * i)) & 0xff) as u8;
+ }
+}
+
+fn push_signed_le(bytes: &mut Vec<u8>, value: i128, num_bytes: usize) {
+ for i in 0..num_bytes {
+ bytes.push(((value >> (8 * i)) & 0xff) as u8);
+ }
+}
+
+fn read_unsigned(bytes: &[u8], pos: usize, num_bytes: usize) -> Result<usize> {
+ check_range(bytes, pos, num_bytes)?;
+ let mut result = 0usize;
+ for i in 0..num_bytes {
+ result |= (bytes[pos + i] as usize) << (8 * i);
+ }
+ Ok(result)
+}
+
+fn read_signed(bytes: &[u8], pos: usize, num_bytes: usize) -> Result<i64> {
+ check_range(bytes, pos, num_bytes)?;
+ let mut result = 0i64;
+ for i in 0..num_bytes {
+ result |= (bytes[pos + i] as i64) << (8 * i);
+ }
+ let shift = 64 - num_bytes * 8;
+ Ok((result << shift) >> shift)
+}
+
+fn value_kind(value: &[u8], pos: usize) -> Result<VariantKind> {
+ check_index(value, pos)?;
+ let basic_type = value[pos] & BASIC_TYPE_MASK;
+ let type_info = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ Ok(match basic_type {
+ SHORT_STR => VariantKind::String,
+ OBJECT => VariantKind::Object,
+ ARRAY => VariantKind::Array,
+ _ => match type_info {
+ NULL => VariantKind::Null,
+ TRUE | FALSE => VariantKind::Boolean,
+ INT1 | INT2 | INT4 | INT8 => VariantKind::Long,
+ DOUBLE => VariantKind::Double,
+ DECIMAL4 | DECIMAL8 | DECIMAL16 => VariantKind::Decimal,
+ DATE => VariantKind::Date,
+ TIMESTAMP => VariantKind::Timestamp,
+ TIMESTAMP_NTZ => VariantKind::TimestampNtz,
+ FLOAT => VariantKind::Float,
+ BINARY => VariantKind::Binary,
+ LONG_STR => VariantKind::String,
+ UUID => VariantKind::Uuid,
+ _ => return data_invalid(format!("Unknown primitive type in
Variant: {type_info}")),
+ },
+ })
+}
+
+fn value_size(value: &[u8], pos: usize) -> Result<usize> {
+ check_index(value, pos)?;
+ let basic_type = value[pos] & BASIC_TYPE_MASK;
+ let type_info = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ match basic_type {
+ SHORT_STR => Ok(1 + type_info as usize),
+ OBJECT => {
+ let layout = object_layout(value, pos)?;
+ let data_size = read_unsigned(
+ value,
+ layout.offset_start + layout.size * layout.offset_size,
+ layout.offset_size,
+ )?;
+ Ok(layout.data_start - pos + data_size)
+ }
+ ARRAY => {
+ let layout = array_layout(value, pos)?;
+ let data_size = read_unsigned(
+ value,
+ layout.offset_start + layout.size * layout.offset_size,
+ layout.offset_size,
+ )?;
+ Ok(layout.data_start - pos + data_size)
+ }
+ _ => match type_info {
+ NULL | TRUE | FALSE => Ok(1),
+ INT1 => Ok(2),
+ INT2 => Ok(3),
+ INT4 | DATE | FLOAT => Ok(5),
+ INT8 | DOUBLE | TIMESTAMP | TIMESTAMP_NTZ => Ok(9),
+ DECIMAL4 => Ok(6),
+ DECIMAL8 => Ok(10),
+ DECIMAL16 => Ok(18),
+ BINARY | LONG_STR => {
+ let len = read_unsigned(value, pos + 1, U32_SIZE)?;
+ Ok(1 + U32_SIZE + len)
+ }
+ UUID => Ok(17),
+ _ => data_invalid(format!("Unknown primitive type in Variant:
{type_info}")),
+ },
+ }
+}
+
+#[derive(Clone, Copy)]
+struct ObjectLayout {
+ size: usize,
+ id_size: usize,
+ offset_size: usize,
+ id_start: usize,
+ offset_start: usize,
+ data_start: usize,
+}
+
+fn object_layout(value: &[u8], pos: usize) -> Result<ObjectLayout> {
+ check_index(value, pos)?;
+ if (value[pos] & BASIC_TYPE_MASK) != OBJECT {
+ return data_invalid("Expected Variant object");
+ }
+ let type_info = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ let large_size = ((type_info >> 4) & 0x1) != 0;
+ let size_bytes = if large_size { U32_SIZE } else { 1 };
+ let size = read_unsigned(value, pos + 1, size_bytes)?;
+ let id_size = ((type_info >> 2) & 0x3) as usize + 1;
+ let offset_size = (type_info & 0x3) as usize + 1;
+ let id_start = pos + 1 + size_bytes;
+ let offset_start = id_start + size * id_size;
+ let data_start = offset_start + (size + 1) * offset_size;
+ check_range(value, id_start, size * id_size)?;
+ check_range(value, offset_start, (size + 1) * offset_size)?;
+ Ok(ObjectLayout {
+ size,
+ id_size,
+ offset_size,
+ id_start,
+ offset_start,
+ data_start,
+ })
+}
+
+#[derive(Clone, Copy)]
+struct ArrayLayout {
+ size: usize,
+ offset_size: usize,
+ offset_start: usize,
+ data_start: usize,
+}
+
+fn array_layout(value: &[u8], pos: usize) -> Result<ArrayLayout> {
+ check_index(value, pos)?;
+ if (value[pos] & BASIC_TYPE_MASK) != ARRAY {
+ return data_invalid("Expected Variant array");
+ }
+ let type_info = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ let large_size = ((type_info >> 2) & 0x1) != 0;
+ let size_bytes = if large_size { U32_SIZE } else { 1 };
+ let size = read_unsigned(value, pos + 1, size_bytes)?;
+ let offset_size = (type_info & 0x3) as usize + 1;
+ let offset_start = pos + 1 + size_bytes;
+ let data_start = offset_start + (size + 1) * offset_size;
+ check_range(value, offset_start, (size + 1) * offset_size)?;
+ Ok(ArrayLayout {
+ size,
+ offset_size,
+ offset_start,
+ data_start,
+ })
+}
+
+fn metadata_offset_size(metadata: &[u8]) -> Result<usize> {
+ check_index(metadata, 0)?;
+ if (metadata[0] & VERSION_MASK) != VERSION {
+ return data_invalid("Malformed Variant metadata version");
+ }
+ Ok(((metadata[0] >> 6) & 0x3) as usize + 1)
+}
+
+fn get_metadata_key(metadata: &[u8], id: usize) -> Result<String> {
+ let offset_size = metadata_offset_size(metadata)?;
+ let dict_size = read_unsigned(metadata, 1, offset_size)?;
+ if id >= dict_size {
+ return data_invalid("Malformed Variant metadata dictionary id");
+ }
+ let string_start = 1 + (dict_size + 2) * offset_size;
+ let offset = read_unsigned(metadata, 1 + (id + 1) * offset_size,
offset_size)?;
+ let next_offset = read_unsigned(metadata, 1 + (id + 2) * offset_size,
offset_size)?;
+ if offset > next_offset {
+ return data_invalid("Malformed Variant metadata offsets");
+ }
+ 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)),
+ })
+}
+
+fn get_boolean(value: &[u8], pos: usize) -> Result<bool> {
+ check_index(value, pos)?;
+ let basic_type = value[pos] & BASIC_TYPE_MASK;
+ let type_info = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ if basic_type != PRIMITIVE || (type_info != TRUE && type_info != FALSE) {
+ return data_invalid("Expected Variant boolean");
+ }
+ Ok(type_info == TRUE)
+}
+
+fn get_long(value: &[u8], pos: usize) -> Result<i64> {
+ check_index(value, pos)?;
+ if (value[pos] & BASIC_TYPE_MASK) != PRIMITIVE {
+ return data_invalid("Expected Variant long/date/timestamp");
+ }
+ match (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK {
+ INT1 => read_signed(value, pos + 1, 1),
+ INT2 => read_signed(value, pos + 1, 2),
+ INT4 | DATE => read_signed(value, pos + 1, 4),
+ INT8 | TIMESTAMP | TIMESTAMP_NTZ => read_signed(value, pos + 1, 8),
+ _ => data_invalid("Expected Variant long/date/timestamp"),
+ }
+}
+
+fn get_double(value: &[u8], pos: usize) -> Result<f64> {
+ check_index(value, pos)?;
+ if (value[pos] & BASIC_TYPE_MASK) != PRIMITIVE
+ || ((value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK) != DOUBLE
+ {
+ return data_invalid("Expected Variant double");
+ }
+ Ok(f64::from_bits(read_signed(value, pos + 1, 8)? as u64))
+}
+
+fn get_float(value: &[u8], pos: usize) -> Result<f32> {
+ check_index(value, pos)?;
+ if (value[pos] & BASIC_TYPE_MASK) != PRIMITIVE
+ || ((value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK) != FLOAT
+ {
+ return data_invalid("Expected Variant float");
+ }
+ Ok(f32::from_bits(read_signed(value, pos + 1, 4)? as u32))
+}
+
+fn get_string(value: &[u8], pos: usize) -> Result<String> {
+ check_index(value, pos)?;
+ let basic_type = value[pos] & BASIC_TYPE_MASK;
+ let type_info = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ let (start, len) = if basic_type == SHORT_STR {
+ (pos + 1, type_info as usize)
+ } else if basic_type == PRIMITIVE && type_info == LONG_STR {
+ (pos + 1 + U32_SIZE, read_unsigned(value, pos + 1, U32_SIZE)?)
+ } else {
+ return data_invalid("Expected Variant string");
+ };
+ check_range(value, start, len)?;
+ std::str::from_utf8(&value[start..start + len])
+ .map(|v| v.to_string())
+ .map_err(|e| Error::DataInvalid {
+ message: "Malformed Variant string UTF-8".to_string(),
+ source: Some(Box::new(e)),
+ })
+}
+
+fn get_binary(value: &[u8], pos: usize) -> Result<&[u8]> {
+ check_index(value, pos)?;
+ if (value[pos] & BASIC_TYPE_MASK) != PRIMITIVE
+ || ((value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK) != BINARY
+ {
+ return data_invalid("Expected Variant binary");
+ }
+ let len = read_unsigned(value, pos + 1, U32_SIZE)?;
+ let start = pos + 1 + U32_SIZE;
+ check_range(value, start, len)?;
+ Ok(&value[start..start + len])
+}
+
+fn get_uuid(value: &[u8], pos: usize) -> Result<uuid::Uuid> {
+ check_index(value, pos)?;
+ if (value[pos] & BASIC_TYPE_MASK) != PRIMITIVE
+ || ((value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK) != UUID
+ {
+ return data_invalid("Expected Variant UUID");
+ }
+ check_range(value, pos + 1, 16)?;
+ let mut bytes = [0u8; 16];
+ bytes.copy_from_slice(&value[pos + 1..pos + 17]);
+ Ok(uuid::Uuid::from_bytes(bytes))
+}
+
+fn get_decimal(value: &[u8], pos: usize) -> Result<VariantDecimal> {
+ check_index(value, pos)?;
+ if (value[pos] & BASIC_TYPE_MASK) != PRIMITIVE {
+ return data_invalid("Expected Variant decimal");
+ }
+ let type_info = (value[pos] >> BASIC_TYPE_BITS) & TYPE_INFO_MASK;
+ let scale = value
+ .get(pos + 1)
+ .copied()
+ .ok_or_else(|| Error::DataInvalid {
+ message: "Malformed Variant decimal".to_string(),
+ source: None,
+ })? as i8;
+ let (unscaled, max_precision) = match type_info {
+ DECIMAL4 => (
+ read_signed(value, pos + 2, 4)? as i128,
+ MAX_DECIMAL4_PRECISION,
+ ),
+ DECIMAL8 => (
+ read_signed(value, pos + 2, 8)? as i128,
+ MAX_DECIMAL8_PRECISION,
+ ),
+ DECIMAL16 => {
+ check_range(value, pos + 2, 16)?;
+ let mut bytes = [0u8; 16];
+ bytes.copy_from_slice(&value[pos + 2..pos + 18]);
+ (i128::from_le_bytes(bytes), MAX_DECIMAL16_PRECISION)
+ }
+ _ => return data_invalid("Expected Variant decimal"),
+ };
+ let precision = decimal_precision(unscaled);
+ if precision > max_precision || scale < 0 || scale as u8 > max_precision {
+ return data_invalid("Malformed Variant decimal precision or scale");
+ }
+ Ok(VariantDecimal {
+ unscaled,
+ precision,
+ scale,
+ })
+}
+
+fn write_json(value: &[u8], metadata: &[u8], pos: usize, out: &mut String) ->
Result<()> {
+ match value_kind(value, pos)? {
+ VariantKind::Object => {
+ let layout = object_layout(value, pos)?;
+ out.push('{');
+ for i in 0..layout.size {
+ if i != 0 {
+ out.push(',');
+ }
+ let id =
+ read_unsigned(value, layout.id_start + layout.id_size * i,
layout.id_size)?;
+ let key = get_metadata_key(metadata, id)?;
+ out.push_str(
+ &serde_json::to_string(&key).map_err(|e|
Error::DataInvalid {
+ message: "Failed to escape Variant object
key".to_string(),
+ source: Some(Box::new(e)),
+ })?,
+ );
+ out.push(':');
+ let offset = read_unsigned(
+ value,
+ layout.offset_start + layout.offset_size * i,
+ layout.offset_size,
+ )?;
+ write_json(value, metadata, layout.data_start + offset, out)?;
+ }
+ out.push('}');
+ }
+ VariantKind::Array => {
+ let layout = array_layout(value, pos)?;
+ out.push('[');
+ for i in 0..layout.size {
+ if i != 0 {
+ out.push(',');
+ }
+ let offset = read_unsigned(
+ value,
+ layout.offset_start + layout.offset_size * i,
+ layout.offset_size,
+ )?;
+ write_json(value, metadata, layout.data_start + offset, out)?;
+ }
+ out.push(']');
+ }
+ VariantKind::Null => out.push_str("null"),
+ VariantKind::Boolean => out.push_str(if get_boolean(value, pos)? {
+ "true"
+ } else {
+ "false"
+ }),
+ VariantKind::Long => out.push_str(&get_long(value, pos)?.to_string()),
+ VariantKind::String => out.push_str(
+ &serde_json::to_string(&get_string(value, pos)?).map_err(|e|
Error::DataInvalid {
+ message: "Failed to escape Variant string".to_string(),
+ source: Some(Box::new(e)),
+ })?,
+ ),
+ VariantKind::Double => out.push_str(&get_double(value,
pos)?.to_string()),
+ VariantKind::Decimal => out.push_str(&get_decimal(value,
pos)?.to_plain_string()),
+ VariantKind::Float => out.push_str(&get_float(value,
pos)?.to_string()),
+ VariantKind::Binary => {
+ let encoded = general_purpose::STANDARD.encode(get_binary(value,
pos)?);
+ out.push_str(
+ &serde_json::to_string(&encoded).map_err(|e|
Error::DataInvalid {
+ message: "Failed to encode Variant binary".to_string(),
+ source: Some(Box::new(e)),
+ })?,
+ );
+ }
+ VariantKind::Uuid => out.push_str(
+ &serde_json::to_string(&get_uuid(value,
pos)?.to_string()).map_err(|e| {
+ Error::DataInvalid {
+ message: "Failed to stringify Variant UUID".to_string(),
+ source: Some(Box::new(e)),
+ }
+ })?,
+ ),
+ VariantKind::Date | VariantKind::Timestamp | VariantKind::TimestampNtz
=> {
+ out.push_str(
+ &serde_json::to_string(&get_long(value,
pos)?.to_string()).map_err(|e| {
+ Error::DataInvalid {
+ message: "Failed to stringify Variant
value".to_string(),
+ source: Some(Box::new(e)),
+ }
+ })?,
+ );
+ }
+ }
+ Ok(())
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+enum PathSegment {
+ Key(String),
+ Index(usize),
+}
+
+fn parse_path(path: &str) -> Result<Vec<PathSegment>> {
+ let bytes = path.as_bytes();
+ if !bytes.starts_with(b"$") {
+ return data_invalid(format!("Invalid Variant path: {path}"));
+ }
+ let mut pos = 1usize;
+ let mut segments = Vec::new();
+ while pos < bytes.len() {
+ match bytes[pos] {
+ b'.' => {
+ pos += 1;
+ let start = pos;
+ while pos < bytes.len() && bytes[pos] != b'.' && bytes[pos] !=
b'[' {
+ pos += 1;
+ }
+ if start == pos {
+ return data_invalid(format!("Invalid Variant path:
{path}"));
+ }
+ segments.push(PathSegment::Key(path[start..pos].to_string()));
+ }
+ b'[' => {
+ pos += 1;
+ if pos >= bytes.len() {
+ return data_invalid(format!("Invalid Variant path:
{path}"));
+ }
+ if bytes[pos] == b'\'' || bytes[pos] == b'"' {
+ let quote = bytes[pos];
+ pos += 1;
+ let start = pos;
+ while pos < bytes.len() && bytes[pos] != quote {
+ pos += 1;
+ }
+ if pos >= bytes.len() || pos + 1 >= bytes.len() ||
bytes[pos + 1] != b']' {
+ return data_invalid(format!("Invalid Variant path:
{path}"));
+ }
+
segments.push(PathSegment::Key(path[start..pos].to_string()));
+ pos += 2;
+ } else {
+ let start = pos;
+ while pos < bytes.len() && bytes[pos].is_ascii_digit() {
+ pos += 1;
+ }
+ if start == pos || pos >= bytes.len() || bytes[pos] !=
b']' {
+ return data_invalid(format!("Invalid Variant path:
{path}"));
+ }
+ let index =
+ path[start..pos]
+ .parse::<usize>()
+ .map_err(|e| Error::DataInvalid {
+ message: format!("Invalid Variant path index:
{path}"),
+ source: Some(Box::new(e)),
+ })?;
+ segments.push(PathSegment::Index(index));
+ pos += 1;
+ }
+ }
+ _ => return data_invalid(format!("Invalid Variant path: {path}")),
+ }
+ }
+ Ok(segments)
+}
+
+#[derive(Clone)]
+struct FieldEntry {
+ key: String,
+ id: usize,
+ offset: usize,
+}
+
+struct VariantBuilder {
+ value: Vec<u8>,
+ dictionary: HashMap<String, usize>,
+ dictionary_keys: Vec<Vec<u8>>,
+}
+
+impl VariantBuilder {
+ fn new() -> Self {
+ Self {
+ value: Vec::with_capacity(128),
+ dictionary: HashMap::new(),
+ dictionary_keys: Vec::new(),
+ }
+ }
+
+ fn result(self) -> Result<GenericVariant> {
+ let num_keys = self.dictionary_keys.len();
+ let dictionary_string_size =
self.dictionary_keys.iter().map(Vec::len).sum::<usize>();
+ let max_size = dictionary_string_size.max(num_keys);
+ let offset_size = integer_size(max_size)?;
+ let offset_start = 1 + offset_size;
+ let string_start = offset_start + (num_keys + 1) * offset_size;
+ let metadata_size = string_start + dictionary_string_size;
+ if metadata_size > SIZE_LIMIT {
+ return data_invalid("Variant metadata exceeds size limit");
+ }
+
+ let mut metadata = vec![0u8; metadata_size];
+ metadata[0] = VERSION | (((offset_size - 1) as u8) << 6);
+ write_le_at(&mut metadata, 1, num_keys, offset_size);
+ let mut current_offset = 0usize;
+ for (i, key) in self.dictionary_keys.iter().enumerate() {
+ write_le_at(
+ &mut metadata,
+ offset_start + i * offset_size,
+ current_offset,
+ offset_size,
+ );
+ metadata[string_start + current_offset..string_start +
current_offset + key.len()]
+ .copy_from_slice(key);
+ current_offset += key.len();
+ }
+ write_le_at(
+ &mut metadata,
+ offset_start + num_keys * offset_size,
+ current_offset,
+ offset_size,
+ );
+ GenericVariant::from_parts(self.value, metadata)
+ }
+
+ fn add_key(&mut self, key: &str) -> usize {
+ if let Some(id) = self.dictionary.get(key) {
+ *id
+ } else {
+ let id = self.dictionary_keys.len();
+ self.dictionary.insert(key.to_string(), id);
+ self.dictionary_keys.push(key.as_bytes().to_vec());
+ id
+ }
+ }
+
+ fn append_null(&mut self) {
+ self.value.push(primitive_header(NULL));
+ }
+
+ fn append_bool(&mut self, value: bool) {
+ self.value
+ .push(primitive_header(if value { TRUE } else { FALSE }));
+ }
+
+ fn append_string(&mut self, value: &str) {
+ let bytes = value.as_bytes();
+ if bytes.len() > MAX_SHORT_STR_SIZE {
+ self.value.push(primitive_header(LONG_STR));
+ let pos = self.value.len();
+ self.value.resize(pos + U32_SIZE, 0);
+ write_le_at(&mut self.value, pos, bytes.len(), U32_SIZE);
+ } else {
+ self.value.push(short_str_header(bytes.len()));
+ }
+ self.value.extend_from_slice(bytes);
+ }
+
+ fn append_long(&mut self, value: i64) {
+ if value == value as i8 as i64 {
+ self.value.push(primitive_header(INT1));
+ push_signed_le(&mut self.value, value as i128, 1);
+ } else if value == value as i16 as i64 {
+ self.value.push(primitive_header(INT2));
+ push_signed_le(&mut self.value, value as i128, 2);
+ } else if value == value as i32 as i64 {
+ self.value.push(primitive_header(INT4));
+ push_signed_le(&mut self.value, value as i128, 4);
+ } else {
+ self.value.push(primitive_header(INT8));
+ push_signed_le(&mut self.value, value as i128, 8);
+ }
+ }
+
+ fn append_double(&mut self, value: f64) {
+ self.value.push(primitive_header(DOUBLE));
+ self.value.extend_from_slice(&value.to_bits().to_le_bytes());
+ }
+
+ fn append_decimal(&mut self, decimal: VariantDecimal) {
+ if decimal.scale as u8 <= MAX_DECIMAL4_PRECISION
+ && decimal.precision <= MAX_DECIMAL4_PRECISION
+ {
+ self.value.push(primitive_header(DECIMAL4));
+ self.value.push(decimal.scale as u8);
+ push_signed_le(&mut self.value, decimal.unscaled, 4);
+ } else if decimal.scale as u8 <= MAX_DECIMAL8_PRECISION
+ && decimal.precision <= MAX_DECIMAL8_PRECISION
+ {
+ self.value.push(primitive_header(DECIMAL8));
+ self.value.push(decimal.scale as u8);
+ push_signed_le(&mut self.value, decimal.unscaled, 8);
+ } else {
+ self.value.push(primitive_header(DECIMAL16));
+ self.value.push(decimal.scale as u8);
+ self.value
+ .extend_from_slice(&decimal.unscaled.to_le_bytes());
+ }
+ }
+
+ fn finish_object(&mut self, start: usize, mut fields: Vec<FieldEntry>) ->
Result<()> {
+ fields.sort_by(|a, b| java_string_cmp(&a.key, &b.key));
+ for pair in fields.windows(2) {
+ if pair[0].key == pair[1].key {
+ return data_invalid("VARIANT_DUPLICATE_KEY");
+ }
+ }
+
+ let data_size = self.value.len() - start;
+ let size = fields.len();
+ let large_size = size > U8_MAX;
+ let size_bytes = if large_size { U32_SIZE } else { 1 };
+ let max_id = fields.iter().map(|f| f.id).max().unwrap_or(0);
+ let id_size = integer_size(max_id)?;
+ let offset_size = integer_size(data_size)?;
+ let header_size = 1 + size_bytes + size * id_size + (size + 1) *
offset_size;
+ self.value.splice(start..start, vec![0; header_size]);
+ self.value[start] = object_header(large_size, id_size, offset_size);
+ write_le_at(&mut self.value, start + 1, size, size_bytes);
+ let id_start = start + 1 + size_bytes;
+ let offset_start = id_start + size * id_size;
+ for (i, field) in fields.iter().enumerate() {
+ write_le_at(&mut self.value, id_start + i * id_size, field.id,
id_size);
+ write_le_at(
+ &mut self.value,
+ offset_start + i * offset_size,
+ field.offset,
+ offset_size,
+ );
+ }
+ write_le_at(
+ &mut self.value,
+ offset_start + size * offset_size,
+ data_size,
+ offset_size,
+ );
+ Ok(())
+ }
+
+ fn finish_array(&mut self, start: usize, offsets: Vec<usize>) ->
Result<()> {
+ let data_size = self.value.len() - start;
+ let size = offsets.len();
+ let large_size = size > U8_MAX;
+ let size_bytes = if large_size { U32_SIZE } else { 1 };
+ let offset_size = integer_size(data_size)?;
+ let header_size = 1 + size_bytes + (size + 1) * offset_size;
+ self.value.splice(start..start, vec![0; header_size]);
+ self.value[start] = array_header(large_size, offset_size);
+ write_le_at(&mut self.value, start + 1, size, size_bytes);
+ let offset_start = start + 1 + size_bytes;
+ for (i, offset) in offsets.iter().enumerate() {
+ write_le_at(
+ &mut self.value,
+ offset_start + i * offset_size,
+ *offset,
+ offset_size,
+ );
+ }
+ write_le_at(
+ &mut self.value,
+ offset_start + size * offset_size,
+ data_size,
+ offset_size,
+ );
+ Ok(())
+ }
+}
+
+struct JsonVariantParser<'a> {
+ input: &'a str,
+ bytes: &'a [u8],
+ pos: usize,
+ builder: VariantBuilder,
+}
+
+impl<'a> JsonVariantParser<'a> {
+ fn new(input: &'a str) -> Self {
+ Self {
+ input,
+ bytes: input.as_bytes(),
+ pos: 0,
+ builder: VariantBuilder::new(),
+ }
+ }
+
+ fn parse(mut self) -> Result<GenericVariant> {
+ self.skip_ws();
+ self.parse_value()?;
+ self.skip_ws();
+ if self.pos != self.bytes.len() {
+ return self.parse_error("Trailing characters after JSON value");
+ }
+ self.builder.result()
+ }
+
+ fn parse_value(&mut self) -> Result<()> {
+ self.skip_ws();
+ let Some(ch) = self.peek() else {
+ return self.parse_error("Unexpected end of JSON input");
+ };
+ match ch {
+ b'{' => self.parse_object(),
+ b'[' => self.parse_array(),
+ b'"' => {
+ let value = self.parse_string()?;
+ self.builder.append_string(&value);
+ Ok(())
+ }
+ b't' => {
+ self.expect_literal("true")?;
+ self.builder.append_bool(true);
+ Ok(())
+ }
+ b'f' => {
+ self.expect_literal("false")?;
+ self.builder.append_bool(false);
+ Ok(())
+ }
+ b'n' => {
+ self.expect_literal("null")?;
+ self.builder.append_null();
+ Ok(())
+ }
+ b'-' | b'0'..=b'9' => self.parse_number(),
+ _ => self.parse_error("Unexpected JSON token"),
+ }
+ }
+
+ fn parse_object(&mut self) -> Result<()> {
+ self.consume(b'{')?;
+ let start = self.builder.value.len();
+ let mut fields = Vec::new();
+ let mut seen = HashSet::new();
+ self.skip_ws();
+ if self.peek() == Some(b'}') {
+ self.pos += 1;
+ return self.builder.finish_object(start, fields);
+ }
+ loop {
+ self.skip_ws();
+ let key = self.parse_string()?;
+ if !seen.insert(key.clone()) {
+ return self.parse_error("VARIANT_DUPLICATE_KEY");
+ }
+ self.skip_ws();
+ self.consume(b':')?;
+ let id = self.builder.add_key(&key);
+ let offset = self.builder.value.len() - start;
+ fields.push(FieldEntry { key, id, offset });
+ self.parse_value()?;
+ self.skip_ws();
+ match self.peek() {
+ Some(b',') => {
+ self.pos += 1;
+ }
+ Some(b'}') => {
+ self.pos += 1;
+ break;
+ }
+ _ => return self.parse_error("Expected ',' or '}' in JSON
object"),
+ }
+ }
+ self.builder.finish_object(start, fields)
+ }
+
+ fn parse_array(&mut self) -> Result<()> {
+ self.consume(b'[')?;
+ let start = self.builder.value.len();
+ let mut offsets = Vec::new();
+ self.skip_ws();
+ if self.peek() == Some(b']') {
+ self.pos += 1;
+ return self.builder.finish_array(start, offsets);
+ }
+ loop {
+ offsets.push(self.builder.value.len() - start);
+ self.parse_value()?;
+ self.skip_ws();
+ match self.peek() {
+ Some(b',') => {
+ self.pos += 1;
+ }
+ Some(b']') => {
+ self.pos += 1;
+ break;
+ }
+ _ => return self.parse_error("Expected ',' or ']' in JSON
array"),
+ }
+ }
+ self.builder.finish_array(start, offsets)
+ }
+
+ fn parse_string(&mut self) -> Result<String> {
+ self.skip_ws();
+ let start = self.pos;
+ self.consume(b'"')?;
+ let mut escaped = false;
+ while let Some(ch) = self.peek() {
+ self.pos += 1;
+ if escaped {
+ escaped = false;
+ continue;
+ }
+ match ch {
+ b'\\' => escaped = true,
+ b'"' => {
+ return
serde_json::from_str(&self.input[start..self.pos]).map_err(|e| {
+ Error::DataInvalid {
+ message: "Invalid JSON string".to_string(),
+ source: Some(Box::new(e)),
+ }
+ });
+ }
+ _ if ch < 0x20 => return self.parse_error("Invalid control
character in string"),
+ _ => {}
+ }
+ }
+ self.parse_error("Unterminated JSON string")
+ }
+
+ fn parse_number(&mut self) -> Result<()> {
+ let start = self.pos;
+ if self.peek() == Some(b'-') {
+ self.pos += 1;
+ }
+ match self.peek() {
+ Some(b'0') => self.pos += 1,
+ Some(b'1'..=b'9') => {
+ self.pos += 1;
+ while matches!(self.peek(), Some(b'0'..=b'9')) {
+ self.pos += 1;
+ }
+ }
+ _ => return self.parse_error("Invalid JSON number"),
+ }
+ let mut has_fraction = false;
+ if self.peek() == Some(b'.') {
+ has_fraction = true;
+ self.pos += 1;
+ let fraction_start = self.pos;
+ while matches!(self.peek(), Some(b'0'..=b'9')) {
+ self.pos += 1;
+ }
+ if self.pos == fraction_start {
+ return self.parse_error("Invalid JSON number fraction");
+ }
+ }
+ let mut has_exponent = false;
+ if matches!(self.peek(), Some(b'e' | b'E')) {
+ has_exponent = true;
+ self.pos += 1;
+ if matches!(self.peek(), Some(b'+' | b'-')) {
+ self.pos += 1;
+ }
+ let exponent_start = self.pos;
+ while matches!(self.peek(), Some(b'0'..=b'9')) {
+ self.pos += 1;
+ }
+ if self.pos == exponent_start {
+ return self.parse_error("Invalid JSON number exponent");
+ }
+ }
+ let token = &self.input[start..self.pos];
+ if !has_fraction && !has_exponent {
+ if let Ok(value) = token.parse::<i64>() {
+ self.builder.append_long(value);
+ return Ok(());
+ }
+ }
+ if !has_exponent {
+ if let Some(decimal) = parse_decimal_token(token)? {
+ self.builder.append_decimal(decimal);
+ return Ok(());
+ }
+ }
+ let value = token.parse::<f64>().map_err(|e| Error::DataInvalid {
+ message: format!("Invalid JSON number: {token}"),
+ source: Some(Box::new(e)),
+ })?;
+ self.builder.append_double(value);
+ Ok(())
+ }
+
+ fn expect_literal(&mut self, literal: &str) -> Result<()> {
+ if self.input[self.pos..].starts_with(literal) {
+ self.pos += literal.len();
+ Ok(())
+ } else {
+ self.parse_error(format!("Expected JSON literal {literal}"))
+ }
+ }
+
+ fn consume(&mut self, expected: u8) -> Result<()> {
+ self.skip_ws();
+ if self.peek() == Some(expected) {
+ self.pos += 1;
+ Ok(())
+ } else {
+ self.parse_error(format!("Expected '{}'", expected as char))
+ }
+ }
+
+ fn skip_ws(&mut self) {
+ while matches!(self.peek(), Some(b' ' | b'\n' | b'\r' | b'\t')) {
+ self.pos += 1;
+ }
+ }
+
+ fn peek(&self) -> Option<u8> {
+ self.bytes.get(self.pos).copied()
+ }
+
+ fn parse_error<T>(&self, message: impl Into<String>) -> Result<T> {
+ data_invalid(format!("{} at byte {}", message.into(), self.pos))
+ }
+}
+
+fn parse_decimal_token(token: &str) -> Result<Option<VariantDecimal>> {
+ if !token
+ .bytes()
+ .all(|ch| ch == b'-' || ch == b'.' || ch.is_ascii_digit())
+ {
+ return Ok(None);
+ }
+ let negative = token.starts_with('-');
+ let unsigned = token.strip_prefix('-').unwrap_or(token);
+ let scale = unsigned
+ .split_once('.')
+ .map(|(_, fraction)| fraction.len())
+ .unwrap_or(0);
+ if scale > MAX_DECIMAL16_PRECISION as usize {
+ return Ok(None);
+ }
+ let mut digits = String::with_capacity(unsigned.len());
+ for ch in unsigned.bytes() {
+ if ch != b'.' {
+ digits.push(ch as char);
+ }
+ }
+ let significant = digits.trim_start_matches('0');
+ let precision = if significant.is_empty() {
+ 1
+ } else {
+ significant.len()
+ };
+ if precision > MAX_DECIMAL16_PRECISION as usize {
+ return Ok(None);
+ }
+ let mut unscaled = digits.parse::<i128>().map_err(|e| Error::DataInvalid {
+ message: format!("Invalid decimal Variant number: {token}"),
+ source: Some(Box::new(e)),
+ })?;
+ if negative {
+ unscaled = -unscaled;
+ }
+ Ok(Some(VariantDecimal {
+ unscaled,
+ precision: precision as u8,
+ scale: scale as i8,
+ }))
+}
+
+fn decimal_precision(unscaled: i128) -> u8 {
+ let mut value = unscaled.unsigned_abs();
+ if value == 0 {
+ return 1;
+ }
+ let mut precision = 0u8;
+ while value > 0 {
+ precision += 1;
+ value /= 10;
+ }
+ precision
+}
+
+fn decimal_to_plain_string(unscaled: i128, scale: i8, strip_trailing_zeros:
bool) -> String {
+ if scale <= 0 {
+ return unscaled.to_string();
+ }
+ let negative = unscaled < 0;
+ let digits = unscaled.unsigned_abs().to_string();
+ let scale = scale as usize;
+ let mut result = if digits.len() > scale {
+ let split = digits.len() - scale;
+ format!("{}.{}", &digits[..split], &digits[split..])
+ } else {
+ format!("0.{}{}", "0".repeat(scale - digits.len()), digits)
+ };
+ if strip_trailing_zeros && result.contains('.') {
+ while result.ends_with('0') {
+ result.pop();
+ }
+ if result.ends_with('.') {
+ result.pop();
+ }
+ }
+ if negative && result != "0" {
+ result.insert(0, '-');
+ }
+ result
+}
+
+fn java_string_cmp(left: &str, right: &str) -> std::cmp::Ordering {
+ left.encode_utf16().cmp(right.encode_utf16())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn empty_metadata() -> Vec<u8> {
+ vec![VERSION, 0, 0]
+ }
+
+ #[test]
+ fn parse_json_matches_java_basic_object_layout() {
+ let variant =
GenericVariant::parse_json(r#"{"age":27,"city":"Beijing"}"#).unwrap();
+ assert_eq!(
+ variant.value(),
+ &[
+ 0x02, 0x02, 0x00, 0x01, 0x00, 0x02, 0x0a, 0x0c, 0x1b, 0x1d,
b'B', b'e', b'i', b'j',
+ b'i', b'n', b'g'
+ ]
+ );
+ assert_eq!(
+ variant.metadata(),
+ &[0x01, 0x02, 0x00, 0x03, 0x07, b'a', b'g', b'e', b'c', b'i',
b't', b'y']
+ );
+ }
+
+ #[test]
+ fn variant_get_path_reads_objects_and_arrays() {
+ let variant =
+
GenericVariant::parse_json(r#"{"object":{"name":"Alice"},"array":[1,2,null]}"#)
+ .unwrap();
+ assert_eq!(
+ variant
+ .get_path("$.object.name")
+ .unwrap()
+ .unwrap()
+ .get_string()
+ .unwrap(),
+ "Alice"
+ );
+ assert_eq!(
+ variant
+ .get_path("$.array[1]")
+ .unwrap()
+ .unwrap()
+ .get_long()
+ .unwrap(),
+ 2
+ );
+ assert!(variant
+ .get_path("$.array[2]")
+ .unwrap()
+ .unwrap()
+ .is_null()
+ .unwrap());
+ assert!(variant.get_path("$.array[9]").unwrap().is_none());
+ }
+
+ #[test]
+ fn parse_json_rejects_duplicate_object_keys() {
+ let err = GenericVariant::parse_json(r#"{"a":1,"a":2}"#).unwrap_err();
+ assert!(err.to_string().contains("VARIANT_DUPLICATE_KEY"));
+ }
+
+ #[test]
+ fn validate_payload_rejects_malformed_root_value() {
+ let metadata = empty_metadata();
+ assert!(validate_payload(&[], &metadata).is_err());
+
+ let truncated_short_string = [short_str_header(3), b'a'];
+ assert!(validate_payload(&truncated_short_string, &metadata).is_err());
+ }
+
+ #[test]
+ fn validate_payload_rejects_bad_object_metadata_ids() {
+ let variant = GenericVariant::parse_json(r#"{"a":1}"#).unwrap();
+ let mut value = variant.value().to_vec();
+ let layout = object_layout(&value, 0).unwrap();
+ write_le_at(&mut value, layout.id_start, 1, layout.id_size);
+
+ assert!(validate_payload(&value, variant.metadata()).is_err());
+ }
+
+ #[test]
+ fn validate_payload_rejects_bad_object_offsets() {
+ let variant = GenericVariant::parse_json(r#"{"a":1}"#).unwrap();
+ let mut value = variant.value().to_vec();
+ let layout = object_layout(&value, 0).unwrap();
+ let data_size = value.len() - layout.data_start;
+ write_le_at(
+ &mut value,
+ layout.offset_start + layout.size * layout.offset_size,
+ data_size + 1,
+ layout.offset_size,
+ );
+
+ assert!(validate_payload(&value, variant.metadata()).is_err());
+ }
+
+ #[test]
+ fn json_number_encoding_keeps_decimal_and_double_distinct() {
+ let decimal = GenericVariant::parse_json(r#"{"d":123.4500}"#).unwrap();
+ let d = decimal.get_path("$.d").unwrap().unwrap();
+ assert_eq!(d.kind().unwrap(), VariantKind::Decimal);
+ assert_eq!(d.to_json().unwrap(), "123.45");
+
+ let double = GenericVariant::parse_json(r#"{"d":1.23e10}"#).unwrap();
+ assert_eq!(
+ double.get_path("$.d").unwrap().unwrap().kind().unwrap(),
+ VariantKind::Double
+ );
+ }
+}
diff --git a/crates/paimon/tests/fixtures/variant_type.json
b/crates/paimon/tests/fixtures/variant_type.json
new file mode 100644
index 0000000..a4b1f54
--- /dev/null
+++ b/crates/paimon/tests/fixtures/variant_type.json
@@ -0,0 +1 @@
+"VARIANT NOT NULL"
diff --git a/crates/paimon/tests/fixtures/variant_type_nullable.json
b/crates/paimon/tests/fixtures/variant_type_nullable.json
new file mode 100644
index 0000000..e96faaf
--- /dev/null
+++ b/crates/paimon/tests/fixtures/variant_type_nullable.json
@@ -0,0 +1 @@
+"VARIANT"
diff --git a/docs/src/sql.md b/docs/src/sql.md
index b43f84c..283e03f 100644
--- a/docs/src/sql.md
+++ b/docs/src/sql.md
@@ -95,6 +95,7 @@ The following SQL data types are supported in CREATE TABLE
and mapped to their c
| `DOUBLE` / `DOUBLE PRECISION` | DoubleType | |
| `VARCHAR` / `TEXT` / `STRING` / `CHAR` | VarCharType | |
| `BINARY` / `VARBINARY` / `BYTEA` | VarBinaryType | |
+| `VARIANT` | VariantType | Semi-structured value encoded as value + metadata
binary buffers |
| `BLOB` | BlobType | Binary large object |
| `DATE` | DateType | |
| `TIMESTAMP[(p)]` | TimestampType | Precision p: 0/3/6/9, default 3 |
@@ -104,6 +105,123 @@ The following SQL data types are supported in CREATE
TABLE and mapped to their c
| `MAP(key, value)` | MapType | e.g. `MAP(STRING, INT)` |
| `STRUCT<field TYPE, ...>` | RowType | e.g. `STRUCT<city STRING, zip INT>` |
+### Variant Usage
+
+`VARIANT` stores semi-structured data using the same logical value + metadata
binary shape as Paimon Java. Use it for JSON-like fields whose schema may
differ row by row.
+
+Create `VARIANT` columns like ordinary table columns:
+
+```sql
+CREATE TABLE paimon.my_db.user_events (
+ user_id BIGINT NOT NULL,
+ event_time TIMESTAMP,
+ payload VARIANT,
+ attributes VARIANT,
+ dt STRING,
+ PRIMARY KEY (user_id, dt)
+) PARTITIONED BY (dt)
+WITH ('bucket' = '4');
+```
+
+`VARIANT` columns can be nullable or `NOT NULL`:
+
+```sql
+CREATE TABLE paimon.my_db.variant_examples (
+ id INT NOT NULL,
+ payload VARIANT NOT NULL,
+ optional_payload VARIANT
+);
+```
+
+Do not use `VARIANT` as a partition column. Partition values must be scalar
strings, numbers, dates, or timestamps that can be encoded as stable partition
names.
+
+Use `parse_json` when inserting JSON text into a `VARIANT` column:
+
+```sql
+INSERT INTO paimon.my_db.user_events VALUES
+(
+ 1,
+ TIMESTAMP '2024-01-01 10:00:00',
+
parse_json('{"event":"login","device":{"os":"ios","version":17},"score":98.5}'),
+ parse_json('{"city":"Beijing","tags":["new","mobile"],"vip":true}'),
+ '2024-01-01'
+);
+```
+
+`parse_json` rejects invalid JSON and duplicate object keys. Use
`try_parse_json` when malformed JSON should become SQL `NULL` instead of
failing the query:
+
+```sql
+INSERT INTO paimon.my_db.user_events
+SELECT
+ user_id,
+ event_time,
+ try_parse_json(raw_payload),
+ try_parse_json(raw_attributes),
+ dt
+FROM staging_events;
+```
+
+`SQLContext::new` registers Spark-compatible scalar functions for common
`VARIANT` workflows:
+
+```sql
+SELECT
+ user_id,
+ variant_get(payload, '$.event', 'string') AS event_name,
+ variant_get(payload, '$.device.os', 'string') AS os,
+ variant_get(payload, '$.score', 'double') AS score,
+ variant_get(attributes, '$.tags[0]', 'string') AS first_tag
+FROM paimon.my_db.user_events
+WHERE variant_get(attributes, '$.vip', 'boolean') = true;
+```
+
+Supported functions:
+
+| Function | Notes |
+|---|---|
+| `parse_json(json)` | Parses a JSON string into `VARIANT`; invalid JSON
returns an error |
+| `try_parse_json(json)` | Parses a JSON string into `VARIANT`; invalid JSON
returns `NULL` |
+| `variant_get(v, path[, type])` | Extracts a path; missing paths return
`NULL`; invalid casts return an error |
+| `try_variant_get(v, path[, type])` | Extracts a path; missing paths, invalid
paths, and invalid casts return `NULL` |
+| `is_variant_null(v)` | Returns true for JSON `null` inside `VARIANT`; SQL
`NULL` returns false |
+
+Path syntax supports the root path `$`, object access (`$.field`), quoted
object access (`$["field"]` or `$['field']`), array indexes (`$[0]`), and
nested combinations such as `$.items[0].price`.
+
+The optional `type` argument is a string literal. Supported result types are
`variant` (or omitted), `boolean`, `byte` / `tinyint`, `short` / `smallint`,
`int` / `integer`, `long` / `bigint`, `float`, `double`, `decimal(p, s)`, and
`string`.
+
+When `type` is omitted or set to `variant`, `variant_get` returns a nested
`VARIANT` value that can be passed to another `variant_get` call:
+
+```sql
+SELECT
+ variant_get(
+ variant_get(payload, '$.device'),
+ '$.os',
+ 'string'
+ ) AS os
+FROM paimon.my_db.user_events;
+```
+
+Missing paths return SQL `NULL`. JSON `null` is represented as a non-SQL-null
Variant value, so use `is_variant_null` when you need to distinguish it:
+
+```sql
+SELECT
+ is_variant_null(parse_json('null')) AS json_null,
+ is_variant_null(NULL) AS sql_null;
+```
+
+Current limitations:
+
+- `schema_of_variant`, `schema_of_variant_agg`, `to_variant_object`,
`variant_explode`, and `variant_explode_outer` are not implemented yet.
+- `variant_get` currently casts to scalar types and `VARIANT`. It does not yet
cast directly to `ARRAY`, `MAP`, or `STRUCT`.
+- Predicate pushdown is not applied through `variant_get`; DataFusion
evaluates Variant filters after reading rows.
+
+With a raw DataFusion `SessionContext`, register these scalar functions
explicitly:
+
+```rust
+use paimon_datafusion::register_variant_functions;
+
+register_variant_functions(&ctx);
+```
+
## DDL
### CREATE DATABASE / CREATE SCHEMA / DROP SCHEMA
@@ -280,6 +398,19 @@ INSERT INTO paimon.my_db.users VALUES (1, 'alice'), (2,
'bob'), (3, 'carol');
INSERT INTO paimon.my_db.users SELECT * FROM source_table;
```
+For `VARIANT` columns, convert JSON text with `parse_json` or `try_parse_json`:
+
+```sql
+INSERT INTO paimon.my_db.user_events (user_id, event_time, payload,
attributes, dt)
+VALUES (
+ 1,
+ TIMESTAMP '2024-01-01 10:00:00',
+ parse_json('{"event":"login","device":{"os":"ios"}}'),
+ try_parse_json('{"vip":true,"tags":["mobile"]}'),
+ '2024-01-01'
+);
+```
+
For primary-key tables, records with duplicate keys are deduplicated according
to the merge engine (default: Deduplicate engine, where the last written value
wins).
### Mosaic Read Scope
@@ -492,6 +623,29 @@ All DataFusion query capabilities are supported (JOINs,
aggregations, subqueries
SELECT id, name FROM paimon.my_db.users WHERE id > 10 ORDER BY id LIMIT 100;
```
+### Variant Queries
+
+Use `variant_get` to extract fields from `VARIANT` columns. Provide a target
type string when the query needs a scalar result:
+
+```sql
+SELECT
+ user_id,
+ variant_get(payload, '$.event', 'string') AS event_name,
+ variant_get(payload, '$.device.os', 'string') AS device_os,
+ variant_get(attributes, '$.vip', 'boolean') AS is_vip
+FROM paimon.my_db.user_events
+WHERE variant_get(payload, '$.event', 'string') = 'login';
+```
+
+Use `try_variant_get` when incompatible values should return `NULL`:
+
+```sql
+SELECT
+ user_id,
+ try_variant_get(payload, '$.score', 'double') AS score
+FROM paimon.my_db.user_events;
+```
+
### Column Projection
Only the required columns are read, reducing I/O:
