risingwave_common/types/
variant.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::BTreeSet;
16use std::fmt;
17use std::hash::{Hash, Hasher};
18use std::str::FromStr;
19
20use anyhow::{Context, bail};
21use bytes::{Buf, BufMut, Bytes, BytesMut};
22use itertools::Itertools;
23use memcomparable::{Deserializer, Serializer};
24use parquet_variant::{
25    ObjectFieldBuilder, Variant as ParquetVariant, VariantBuilder, VariantBuilderExt,
26    VariantDecimal16,
27};
28use parquet_variant_json::VariantToJson;
29use postgres_types::{FromSql, IsNull, ToSql, Type, accepts, to_sql_checked};
30use risingwave_common_estimate_size::EstimateSize;
31use serde::{Deserialize, Serialize};
32
33use super::jsonb::{JsonbRef, JsonbVal};
34use super::to_binary::ToBinary;
35use super::to_text::ToText;
36use super::{
37    DataType, Decimal, Scalar, ScalarRef, ScalarRefImpl, StructType, scalar_ref_type_match,
38};
39use crate::util::iter_util::ZipEqFast;
40
41const METADATA_LEN_SIZE: usize = size_of::<u32>();
42
43/// Owned value of the `variant` type.
44///
45/// The inner bytes are the little-endian metadata length followed by the Apache Parquet / Iceberg
46/// Variant `metadata` and `value` sections.
47#[derive(Debug, Clone)]
48pub struct VariantVal {
49    data: Box<[u8]>,
50}
51
52/// Borrowed value of the `variant` type.
53#[derive(Debug, Copy, Clone)]
54pub struct VariantRef<'a> {
55    data: &'a [u8],
56}
57
58/// A parsed path for accessing nested fields in a [`VariantRef`].
59///
60/// Parsing a path once and reusing it avoids repeated string parsing and allocations when the same
61/// path is applied to multiple values.
62#[derive(Debug)]
63pub struct VariantPath {
64    tokens: Vec<PathToken>,
65}
66
67impl VariantPath {
68    /// Parses a variant path.
69    pub fn parse(path: &str) -> anyhow::Result<Self> {
70        parse_path(path).map(|tokens| Self { tokens })
71    }
72}
73
74impl EstimateSize for VariantVal {
75    fn estimated_heap_size(&self) -> usize {
76        self.data.len()
77    }
78}
79
80impl fmt::Display for VariantVal {
81    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
82        self.as_scalar_ref().write(f)
83    }
84}
85
86impl fmt::Display for VariantRef<'_> {
87    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
88        self.write(f)
89    }
90}
91
92impl Scalar for VariantVal {
93    type ScalarRefType<'a> = VariantRef<'a>;
94
95    fn as_scalar_ref(&self) -> Self::ScalarRefType<'_> {
96        VariantRef { data: &self.data }
97    }
98}
99
100impl<'a> ScalarRef<'a> for VariantRef<'a> {
101    type ScalarType = VariantVal;
102
103    fn to_owned_scalar(&self) -> Self::ScalarType {
104        VariantVal {
105            data: self.data.into(),
106        }
107    }
108
109    fn hash_scalar<H: std::hash::Hasher>(&self, state: &mut H) {
110        self.hash(state);
111    }
112}
113
114impl PartialEq for VariantVal {
115    fn eq(&self, other: &Self) -> bool {
116        self.as_scalar_ref() == other.as_scalar_ref()
117    }
118}
119
120impl Eq for VariantVal {}
121
122impl Hash for VariantVal {
123    fn hash<H: Hasher>(&self, state: &mut H) {
124        self.as_scalar_ref().hash(state);
125    }
126}
127
128impl PartialEq for VariantRef<'_> {
129    fn eq(&self, other: &Self) -> bool {
130        self.data == other.data
131    }
132}
133
134impl Eq for VariantRef<'_> {}
135
136impl Hash for VariantRef<'_> {
137    fn hash<H: Hasher>(&self, state: &mut H) {
138        self.data.hash(state);
139    }
140}
141
142impl PartialOrd for VariantVal {
143    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
144        Some(self.cmp(other))
145    }
146}
147
148impl Ord for VariantVal {
149    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
150        self.as_scalar_ref().cmp(&other.as_scalar_ref())
151    }
152}
153
154impl PartialOrd for VariantRef<'_> {
155    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
156        Some(self.cmp(other))
157    }
158}
159
160impl Ord for VariantRef<'_> {
161    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
162        // Must agree with `memcmp_serialize`, which encodes the whole serialized buffer.
163        self.data.cmp(other.data)
164    }
165}
166
167impl ToText for VariantRef<'_> {
168    fn write<W: fmt::Write>(&self, f: &mut W) -> fmt::Result {
169        let variant = self.parquet_variant();
170        // `to_json_string` writes non-finite floats bare (e.g. `NaN`), which is not valid
171        // JSON; take the lossy path rendering them as strings instead.
172        let json = if contains_non_finite_float(&variant) {
173            variant_to_json_value_lossy(variant)
174                .map_err(|_| fmt::Error)?
175                .to_string()
176        } else {
177            variant.to_json_string().map_err(|_| fmt::Error)?
178        };
179        f.write_str(&json)
180    }
181
182    fn write_with_type<W: fmt::Write>(&self, _ty: &DataType, f: &mut W) -> fmt::Result {
183        self.write(f)
184    }
185}
186
187impl ToBinary for VariantRef<'_> {
188    fn to_binary_with_type(&self, _ty: &DataType) -> super::to_binary::Result<Bytes> {
189        Ok(Bytes::from(self.value_serialize()))
190    }
191}
192
193impl FromStr for VariantVal {
194    type Err = anyhow::Error;
195
196    fn from_str(s: &str) -> Result<Self, Self::Err> {
197        let json = serde_json::Value::from_str(s)?;
198        Self::from_json_value(&json)
199    }
200}
201
202impl VariantVal {
203    pub fn null() -> Self {
204        Self::from_parquet_variant(ParquetVariant::Null).expect("null variant should encode")
205    }
206
207    pub fn from_parts(metadata: &[u8], value: &[u8]) -> anyhow::Result<Self> {
208        let variant =
209            ParquetVariant::try_new(metadata, value).context("invalid variant encoding")?;
210        Self::from_parquet_variant(variant)
211    }
212
213    fn from_canonical_parts(metadata: &[u8], value: &[u8]) -> Self {
214        // Builder output is trusted; untrusted bytes are validated in `from_parts` and friends.
215        debug_assert!(
216            ParquetVariant::try_new(metadata, value).is_ok(),
217            "canonical variant parts should be valid"
218        );
219        let metadata_len =
220            u32::try_from(metadata.len()).expect("variant metadata exceeds u32::MAX bytes");
221        let mut data = Vec::with_capacity(METADATA_LEN_SIZE + metadata.len() + value.len());
222        data.put_u32_le(metadata_len);
223        data.extend_from_slice(metadata);
224        data.extend_from_slice(value);
225        Self {
226            data: data.into_boxed_slice(),
227        }
228    }
229
230    pub fn from_parquet_variant(variant: ParquetVariant<'_, '_>) -> anyhow::Result<Self> {
231        let field_names = collect_variant_field_names(variant.clone())?;
232        let mut builder = canonical_builder(&field_names);
233        append_variant_value(variant, &mut builder).context("failed to encode variant")?;
234        let (metadata, value) = builder.finish();
235        Ok(Self::from_canonical_parts(&metadata, &value))
236    }
237
238    pub fn from_json_value(json: &serde_json::Value) -> anyhow::Result<Self> {
239        let mut field_names = BTreeSet::new();
240        collect_json_field_names(json, &mut field_names);
241        let mut builder = canonical_builder(&field_names);
242        append_json_value(json, &mut builder)?;
243        let (metadata, value) = builder.finish();
244        Ok(Self::from_canonical_parts(&metadata, &value))
245    }
246
247    pub fn try_from_scalar_ref(
248        value: Option<ScalarRefImpl<'_>>,
249        data_type: &DataType,
250    ) -> anyhow::Result<Self> {
251        // A whole jsonb document needs only one text parse, not one per pass.
252        if let Some(ScalarRefImpl::Jsonb(v)) = value
253            && matches!(data_type, DataType::Jsonb)
254        {
255            return Self::from_jsonb(v);
256        }
257        let mut field_names = BTreeSet::new();
258        collect_datum_field_names(value, data_type, &mut field_names)?;
259        let mut builder = canonical_builder(&field_names);
260        append_datum_value(value, data_type, &mut builder)?;
261        let (metadata, value) = builder.finish();
262        Ok(Self::from_canonical_parts(&metadata, &value))
263    }
264
265    /// Decodes a value produced by [`VariantRef::value_serialize`], checking structure only.
266    /// Use only on trusted bytes; external bytes must go through
267    /// [`Self::from_serialized_untrusted`] to re-establish the canonical invariants.
268    pub fn value_deserialize(buf: &[u8]) -> Option<Self> {
269        VariantRef::from_serialized(buf).map(|v| v.to_owned_scalar())
270    }
271
272    /// Decodes a buffer from an untrusted origin (e.g. pgwire binary parameters),
273    /// re-canonicalizing it so non-canonical inputs (unsorted dictionaries, non-canonical
274    /// float bit patterns) cannot break `Eq`/`Hash`/`Ord`.
275    pub fn from_serialized_untrusted(buf: &[u8]) -> anyhow::Result<Self> {
276        let (metadata, value) = split_serialized_value(buf).context("invalid variant encoding")?;
277        Self::from_parts(metadata, value)
278    }
279
280    pub fn memcmp_deserialize(
281        deserializer: &mut Deserializer<impl Buf>,
282    ) -> memcomparable::Result<Self> {
283        let bytes = <serde_bytes::ByteBuf as Deserialize>::deserialize(deserializer)?;
284        Self::value_deserialize(&bytes)
285            .ok_or_else(|| memcomparable::Error::Message("invalid variant".into()))
286    }
287
288    pub fn from_jsonb(jsonb: JsonbRef<'_>) -> anyhow::Result<Self> {
289        Self::from_json_value(&jsonb_to_json_value(jsonb)?)
290    }
291
292    pub fn metadata(&self) -> &[u8] {
293        expect_serialized_value(&self.data).0
294    }
295
296    pub fn value(&self) -> &[u8] {
297        expect_serialized_value(&self.data).1
298    }
299
300    pub fn parquet_variant(&self) -> ParquetVariant<'_, '_> {
301        let (metadata, value) = expect_serialized_value(&self.data);
302        ParquetVariant::new(metadata, value)
303    }
304
305    pub fn serialized_len(&self) -> usize {
306        self.data.len()
307    }
308}
309
310impl<'a> VariantRef<'a> {
311    /// Same bytes as [`Self::value_serialize`], without the copy.
312    pub fn as_bytes(&self) -> &'a [u8] {
313        debug_assert!(split_serialized_value(self.data).is_some());
314        self.data
315    }
316
317    pub fn value_serialize(&self) -> Vec<u8> {
318        self.as_bytes().to_vec()
319    }
320
321    pub fn memcmp_serialize(
322        &self,
323        serializer: &mut Serializer<impl BufMut>,
324    ) -> memcomparable::Result<()> {
325        Serialize::serialize(&serde_bytes::Bytes::new(self.as_bytes()), serializer)
326    }
327
328    pub fn from_serialized(buf: &'a [u8]) -> Option<Self> {
329        let (metadata, value) = split_serialized_value(buf)?;
330        ParquetVariant::try_new(metadata, value).ok()?;
331        Some(Self { data: buf })
332    }
333
334    /// Wraps bytes already validated by [`Self::from_serialized`], such as a slot of a
335    /// [`VariantArray`](crate::array::VariantArray), which validates on the way in.
336    pub fn from_serialized_unchecked(buf: &'a [u8]) -> Self {
337        debug_assert!(split_serialized_value(buf).is_some());
338        Self { data: buf }
339    }
340
341    pub fn metadata(&self) -> &'a [u8] {
342        expect_serialized_value(self.data).0
343    }
344
345    pub fn value(&self) -> &'a [u8] {
346        expect_serialized_value(self.data).1
347    }
348
349    pub fn parquet_variant(&self) -> ParquetVariant<'a, 'a> {
350        let (metadata, value) = expect_serialized_value(self.data);
351        ParquetVariant::new(metadata, value)
352    }
353
354    pub fn is_variant_null(&self) -> bool {
355        matches!(self.parquet_variant(), ParquetVariant::Null)
356    }
357
358    pub fn is_array(&self) -> bool {
359        matches!(self.parquet_variant(), ParquetVariant::List(_))
360    }
361
362    pub fn is_object(&self) -> bool {
363        matches!(self.parquet_variant(), ParquetVariant::Object(_))
364    }
365
366    pub fn type_name(&self) -> &'static str {
367        match self.parquet_variant() {
368            ParquetVariant::Null => "null",
369            ParquetVariant::BooleanTrue | ParquetVariant::BooleanFalse => "boolean",
370            ParquetVariant::Int8(_) => "int8",
371            ParquetVariant::Int16(_) => "int16",
372            ParquetVariant::Int32(_) => "int32",
373            ParquetVariant::Int64(_) => "int64",
374            ParquetVariant::Float(_) => "float",
375            ParquetVariant::Double(_) => "double",
376            ParquetVariant::Decimal4(_) => "decimal4",
377            ParquetVariant::Decimal8(_) => "decimal8",
378            ParquetVariant::Decimal16(_) => "decimal16",
379            ParquetVariant::Date(_) => "date",
380            ParquetVariant::TimestampMicros(_) => "timestamp_micros",
381            ParquetVariant::TimestampNtzMicros(_) => "timestamp_ntz_micros",
382            ParquetVariant::TimestampNanos(_) => "timestamp_nanos",
383            ParquetVariant::TimestampNtzNanos(_) => "timestamp_ntz_nanos",
384            ParquetVariant::Binary(_) => "binary",
385            ParquetVariant::String(_) | ParquetVariant::ShortString(_) => "string",
386            ParquetVariant::Time(_) => "time",
387            ParquetVariant::Uuid(_) => "uuid",
388            ParquetVariant::Object(_) => "object",
389            ParquetVariant::List(_) => "array",
390        }
391    }
392
393    pub fn access_path(self, path: &str) -> Option<VariantVal> {
394        self.access_path_strict(path).ok().flatten()
395    }
396
397    pub fn access_path_strict(self, path: &str) -> anyhow::Result<Option<VariantVal>> {
398        self.access_path_parsed(&VariantPath::parse(path)?)
399    }
400
401    /// Accesses a nested value using a pre-parsed path.
402    pub fn access_path_parsed(self, path: &VariantPath) -> anyhow::Result<Option<VariantVal>> {
403        // Walk the whole path on borrowed variants sharing the same metadata, and canonicalize
404        // (re-encode) only the final leaf.
405        let mut variant = self.parquet_variant();
406        for token in &path.tokens {
407            let next = match token {
408                PathToken::Field(field) => variant.get_object_field(field),
409                // Pattern-match instead of `as_list`, whose `&'m self` receiver would keep
410                // `variant` borrowed and forbid the reassignment below.
411                PathToken::Index(index) => match &variant {
412                    ParquetVariant::List(list) => {
413                        let index = if *index >= 0 {
414                            Some(*index as usize)
415                        } else {
416                            list.len().checked_sub(index.unsigned_abs() as usize)
417                        };
418                        index.and_then(|index| list.get(index))
419                    }
420                    _ => None,
421                },
422            };
423            match next {
424                Some(next) => variant = next,
425                None => return Ok(None),
426            }
427        }
428        VariantVal::from_parquet_variant(variant).map(Some)
429    }
430
431    pub fn to_jsonb(self) -> anyhow::Result<JsonbVal> {
432        let json = variant_to_json_value_lossy(self.parquet_variant())
433            .context("failed to convert variant to jsonb")?;
434        Ok(json.into())
435    }
436}
437
438impl From<VariantRef<'_>> for VariantVal {
439    fn from(value: VariantRef<'_>) -> Self {
440        value.to_owned_scalar()
441    }
442}
443
444impl<'a> FromSql<'a> for VariantVal {
445    accepts!(JSON, JSONB);
446
447    fn from_sql(
448        ty: &Type,
449        raw: &'a [u8],
450    ) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
451        Ok(match *ty {
452            Type::JSON => Self::from_str(std::str::from_utf8(raw)?)?,
453            Type::JSONB => {
454                let mut raw = raw;
455                if raw.is_empty() || raw.get_u8() != 1 {
456                    return Err("invalid postgres jsonb encoding".into());
457                }
458                Self::from_str(std::str::from_utf8(raw)?)?
459            }
460            _ => {
461                bail_not_implemented!("the VariantVal's postgres decoding for {ty} is unsupported")
462            }
463        })
464    }
465}
466
467impl ToSql for VariantRef<'_> {
468    accepts!(JSON, JSONB);
469
470    to_sql_checked!();
471
472    fn to_sql(
473        &self,
474        ty: &Type,
475        out: &mut BytesMut,
476    ) -> Result<IsNull, Box<dyn std::error::Error + Sync + Send>>
477    where
478        Self: Sized,
479    {
480        if matches!(*ty, Type::JSONB) {
481            out.put_u8(1);
482        }
483        out.extend_from_slice(self.to_text().as_bytes());
484        Ok(IsNull::No)
485    }
486}
487
488fn split_serialized_value(buf: &[u8]) -> Option<(&[u8], &[u8])> {
489    if buf.len() < METADATA_LEN_SIZE {
490        return None;
491    }
492    let metadata_len = u32::from_le_bytes(buf[..METADATA_LEN_SIZE].try_into().unwrap()) as usize;
493    let metadata_end = METADATA_LEN_SIZE.checked_add(metadata_len)?;
494    if metadata_end > buf.len() {
495        return None;
496    }
497    Some((&buf[METADATA_LEN_SIZE..metadata_end], &buf[metadata_end..]))
498}
499
500fn expect_serialized_value(buf: &[u8]) -> (&[u8], &[u8]) {
501    split_serialized_value(buf).expect("variant should use a valid serialized format")
502}
503
504fn canonical_builder(field_names: &BTreeSet<String>) -> VariantBuilder {
505    VariantBuilder::new().with_field_names(field_names.iter().map(String::as_str))
506}
507
508/// Parses a jsonb value through its text form. The round trip is deliberate: `serde_json`'s parser
509/// enforces a nesting limit, while walking `jsonbb` directly would recurse without a bound.
510fn jsonb_to_json_value(jsonb: JsonbRef<'_>) -> anyhow::Result<serde_json::Value> {
511    Ok(serde_json::Value::from_str(&jsonb.to_string())?)
512}
513
514fn collect_json_field_names(json: &serde_json::Value, field_names: &mut BTreeSet<String>) {
515    match json {
516        serde_json::Value::Array(values) => {
517            for value in values {
518                collect_json_field_names(value, field_names);
519            }
520        }
521        serde_json::Value::Object(fields) => {
522            for (field, value) in fields {
523                field_names.insert(field.clone());
524                collect_json_field_names(value, field_names);
525            }
526        }
527        _ => {}
528    }
529}
530
531fn collect_variant_field_names(
532    variant: ParquetVariant<'_, '_>,
533) -> anyhow::Result<BTreeSet<String>> {
534    let mut field_names = BTreeSet::new();
535    collect_variant_field_names_inner(variant, &mut field_names)?;
536    Ok(field_names)
537}
538
539/// Collects object field names for the canonical dictionary.
540fn collect_variant_field_names_inner(
541    variant: ParquetVariant<'_, '_>,
542    field_names: &mut BTreeSet<String>,
543) -> anyhow::Result<()> {
544    match variant {
545        ParquetVariant::Object(object) => {
546            for field in object.iter_try() {
547                let (field_name, value) = field.context("failed to read variant object")?;
548                field_names.insert(field_name.to_owned());
549                collect_variant_field_names_inner(value, field_names)?;
550            }
551        }
552        ParquetVariant::List(list) => {
553            for value in list.iter_try() {
554                collect_variant_field_names_inner(
555                    value.context("failed to read variant list")?,
556                    field_names,
557                )?;
558            }
559        }
560        _ => {}
561    }
562    Ok(())
563}
564
565fn collect_datum_field_names(
566    value: Option<ScalarRefImpl<'_>>,
567    data_type: &DataType,
568    field_names: &mut BTreeSet<String>,
569) -> anyhow::Result<()> {
570    let Some(value) = value else {
571        return Ok(());
572    };
573
574    match (value, data_type) {
575        (ScalarRefImpl::Jsonb(v), _) => {
576            collect_json_field_names(&jsonb_to_json_value(v)?, field_names);
577        }
578        (ScalarRefImpl::Variant(v), _) => {
579            collect_variant_field_names_inner(v.parquet_variant(), field_names)?;
580        }
581        (ScalarRefImpl::List(v), DataType::List(list_type)) => {
582            for value in v.iter() {
583                collect_datum_field_names(value, list_type.elem(), field_names)?;
584            }
585        }
586        (ScalarRefImpl::Struct(v), DataType::Struct(struct_type)) => {
587            for (value, (field_name, field_type)) in
588                v.iter_fields_ref().zip_eq_fast(struct_type.iter())
589            {
590                field_names.insert(field_name.to_owned());
591                collect_datum_field_names(value, field_type, field_names)?;
592            }
593        }
594        (ScalarRefImpl::Map(v), DataType::Map(map_type)) => {
595            for (key, value) in v.iter() {
596                let field = key.to_text_with_type(map_type.key());
597                field_names.insert(field);
598                collect_datum_field_names(value, map_type.value(), field_names)?;
599            }
600        }
601        // Listed rather than `_`: a field name this pass misses is absent from the dictionary,
602        // which silently yields a non-canonical encoding.
603        (
604            ScalarRefImpl::Bool(_)
605            | ScalarRefImpl::Int16(_)
606            | ScalarRefImpl::Int32(_)
607            | ScalarRefImpl::Int64(_)
608            | ScalarRefImpl::Int256(_)
609            | ScalarRefImpl::Serial(_)
610            | ScalarRefImpl::Float32(_)
611            | ScalarRefImpl::Float64(_)
612            | ScalarRefImpl::Decimal(_)
613            | ScalarRefImpl::Utf8(_)
614            | ScalarRefImpl::Bytea(_)
615            | ScalarRefImpl::Date(_)
616            | ScalarRefImpl::Time(_)
617            | ScalarRefImpl::Timestamp(_)
618            | ScalarRefImpl::Timestamptz(_)
619            | ScalarRefImpl::Interval(_)
620            | ScalarRefImpl::Vector(_)
621            | ScalarRefImpl::List(_)
622            | ScalarRefImpl::Struct(_)
623            | ScalarRefImpl::Map(_),
624            _,
625        ) => {}
626    }
627    Ok(())
628}
629
630fn append_json_value(
631    json: &serde_json::Value,
632    builder: &mut impl VariantBuilderExt,
633) -> anyhow::Result<()> {
634    match json {
635        serde_json::Value::Null => builder.append_value(ParquetVariant::Null),
636        serde_json::Value::Bool(v) => builder.append_value(*v),
637        serde_json::Value::Number(v) => {
638            if let Some(v) = v.as_i64() {
639                builder.append_value(v);
640            } else if let Some(v) = v.as_u64() {
641                builder.append_value(v);
642            } else if let Some(v) = v.as_f64() {
643                append_float64(v, builder);
644            } else {
645                bail!("unsupported JSON number: {v}");
646            }
647        }
648        serde_json::Value::String(v) => builder.append_value(v.as_str()),
649        serde_json::Value::Array(values) => {
650            let mut list = builder
651                .try_new_list()
652                .context("failed to create variant list")?;
653            for value in values {
654                append_json_value(value, &mut list)?;
655            }
656            list.finish();
657        }
658        serde_json::Value::Object(fields) => {
659            let mut object = builder
660                .try_new_object()
661                .context("failed to create variant object")?;
662            for (field, value) in fields.iter().sorted_by(|a, b| a.0.cmp(b.0)) {
663                let mut field_builder = ObjectFieldBuilder::new(field.as_str(), &mut object);
664                append_json_value(value, &mut field_builder)?;
665            }
666            object.finish();
667        }
668    }
669    Ok(())
670}
671
672fn append_datum_value(
673    value: Option<ScalarRefImpl<'_>>,
674    data_type: &DataType,
675    builder: &mut impl VariantBuilderExt,
676) -> anyhow::Result<()> {
677    let Some(value) = value else {
678        builder.append_value(ParquetVariant::Null);
679        return Ok(());
680    };
681    assert!(
682        scalar_ref_type_match(data_type, value),
683        "variant conversion input {} does not match {data_type}",
684        value.get_ident()
685    );
686
687    // Preserve the SQL type identity when Parquet Variant V1 can represent it.
688    // Untyped JSON construction still follows its own default Variant primitive types.
689    match (value, data_type) {
690        (ScalarRefImpl::Bool(v), _) => builder.append_value(v),
691        (ScalarRefImpl::Int16(v), _) => builder.append_value(v),
692        (ScalarRefImpl::Int32(v), _) => builder.append_value(v),
693        (ScalarRefImpl::Int64(v), _) => builder.append_value(v),
694        (ScalarRefImpl::Serial(v), _) => builder.append_value(v.into_inner()),
695        (ScalarRefImpl::Float32(v), _) => append_float64(f64::from(v.into_inner()), builder),
696        (ScalarRefImpl::Float64(v), _) => append_float64(v.into_inner(), builder),
697        (ScalarRefImpl::Decimal(v), _) => append_decimal(v, builder)?,
698        (ScalarRefImpl::Utf8(v), _) => builder.append_value(v),
699        (ScalarRefImpl::Bytea(v), _) => builder.append_value(v),
700        (ScalarRefImpl::Date(v), _) => builder.append_value(v.0),
701        (ScalarRefImpl::Time(v), _) => builder.append_value(v.0),
702        (ScalarRefImpl::Timestamp(v), _) => builder.append_value(v.0),
703        (ScalarRefImpl::Timestamptz(v), _) => builder.append_value(v.to_datetime_utc()),
704        (ScalarRefImpl::Jsonb(v), _) => append_json_value(&jsonb_to_json_value(v)?, builder)?,
705        (ScalarRefImpl::Variant(v), _) => builder.append_value(v.parquet_variant()),
706        // Variant V1 has no primitive for these, so store the text form, as `to_jsonb` does.
707        (ScalarRefImpl::Int256(v), _) => builder.append_value(v.to_text().as_str()),
708        (ScalarRefImpl::Interval(v), _) => builder.append_value(v.to_text().as_str()),
709        (ScalarRefImpl::Vector(v), _) => builder.append_value(v.to_text().as_str()),
710        (ScalarRefImpl::List(v), DataType::List(list_type)) => {
711            let mut list = builder
712                .try_new_list()
713                .context("failed to create variant list")?;
714            for value in v.iter() {
715                append_datum_value(value, list_type.elem(), &mut list)?;
716            }
717            list.finish();
718        }
719        (ScalarRefImpl::Struct(v), DataType::Struct(struct_type)) => {
720            append_struct(v, struct_type, builder)?;
721        }
722        (ScalarRefImpl::Map(v), DataType::Map(map_type)) => {
723            let mut object = builder
724                .try_new_object()
725                .context("failed to create variant map object")?;
726            let entries = v
727                .iter()
728                .map(|(key, value)| {
729                    let field = key.to_text_with_type(map_type.key());
730                    (field, value)
731                })
732                .sorted_by(|a, b| a.0.cmp(&b.0))
733                .collect_vec();
734            for (field, value) in entries {
735                let mut field_builder = ObjectFieldBuilder::new(field.as_str(), &mut object);
736                append_datum_value(value, map_type.value(), &mut field_builder)?;
737            }
738            object.finish();
739        }
740        (value, ty) => bail!("cannot convert {} as {ty} to variant", value.get_ident()),
741    }
742    Ok(())
743}
744
745/// Variant objects require unique field names, but RisingWave struct types allow duplicates.
746/// Callers iterate fields in name order, so comparing against the previous name suffices.
747fn reject_duplicate_field(previous: Option<&str>, field_name: &str) -> anyhow::Result<()> {
748    if previous == Some(field_name) {
749        bail!("variant object cannot have duplicate field name `{field_name}`");
750    }
751    Ok(())
752}
753
754fn append_struct(
755    value: super::StructRef<'_>,
756    struct_type: &StructType,
757    builder: &mut impl VariantBuilderExt,
758) -> anyhow::Result<()> {
759    let mut object = builder
760        .try_new_object()
761        .context("failed to create variant struct object")?;
762    let fields = value
763        .iter_fields_ref()
764        .zip_eq_fast(struct_type.iter())
765        .sorted_by(|(_, (field_a, _)), (_, (field_b, _))| field_a.cmp(field_b));
766    let mut previous_field_name = None;
767    for (value, (field_name, field_type)) in fields {
768        reject_duplicate_field(previous_field_name, field_name)?;
769        previous_field_name = Some(field_name);
770        let mut field_builder = ObjectFieldBuilder::new(field_name, &mut object);
771        append_datum_value(value, field_type, &mut field_builder)?;
772    }
773    object.finish();
774    Ok(())
775}
776
777// Positive quiet NaN with zero payload. All NaN bit patterns collapse to this one so
778// that byte-wise `Eq`/`Hash`/`Ord` treat NaN as a single value, like `memcmp_encoding`
779// does for float columns.
780const CANONICAL_NAN_F32: u32 = 0x7fc0_0000;
781const CANONICAL_NAN_F64: u64 = 0x7ff8_0000_0000_0000;
782
783/// Normalizes a float to its canonical bit pattern: a single `NaN` representation and
784/// `-0.0` collapsed to `+0.0`.
785fn normalize_float(value: f32) -> f32 {
786    if value.is_nan() {
787        f32::from_bits(CANONICAL_NAN_F32)
788    } else if value == 0.0 {
789        0.0
790    } else {
791        value
792    }
793}
794
795/// See [`normalize_float`].
796fn normalize_double(value: f64) -> f64 {
797    if value.is_nan() {
798        f64::from_bits(CANONICAL_NAN_F64)
799    } else if value == 0.0 {
800        0.0
801    } else {
802        value
803    }
804}
805
806fn append_float64(value: f64, builder: &mut impl VariantBuilderExt) {
807    builder.append_value(normalize_double(value));
808}
809
810/// Re-encodes a variant value with the canonical builder, normalizing floats to their
811/// canonical bit patterns.
812fn append_variant_value(
813    variant: ParquetVariant<'_, '_>,
814    builder: &mut impl VariantBuilderExt,
815) -> anyhow::Result<()> {
816    match variant {
817        ParquetVariant::Object(object) => {
818            let mut object_builder = builder
819                .try_new_object()
820                .context("failed to create variant object")?;
821            let mut previous_field_name = None;
822            for field in object.iter_try() {
823                let (field_name, value) = field.context("failed to read variant object")?;
824                reject_duplicate_field(previous_field_name, field_name)?;
825                previous_field_name = Some(field_name);
826                let mut field_builder = ObjectFieldBuilder::new(field_name, &mut object_builder);
827                append_variant_value(value, &mut field_builder)?;
828            }
829            object_builder.finish();
830        }
831        ParquetVariant::List(list) => {
832            let mut list_builder = builder
833                .try_new_list()
834                .context("failed to create variant list")?;
835            for value in list.iter_try() {
836                append_variant_value(
837                    value.context("failed to read variant list")?,
838                    &mut list_builder,
839                )?;
840            }
841            list_builder.finish();
842        }
843        ParquetVariant::Float(v) => builder.append_value(normalize_float(v)),
844        ParquetVariant::Double(v) => append_float64(v, builder),
845        other => builder.append_value(other),
846    }
847    Ok(())
848}
849
850/// Returns whether the variant contains a non-finite float. Read errors also count so
851/// that the lossy path, which reports them, is taken.
852fn contains_non_finite_float(variant: &ParquetVariant<'_, '_>) -> bool {
853    match variant {
854        ParquetVariant::Object(object) => object.iter_try().any(|field| match field {
855            Ok((_, value)) => contains_non_finite_float(&value),
856            Err(_) => true,
857        }),
858        ParquetVariant::List(list) => list.iter_try().any(|value| match value {
859            Ok(value) => contains_non_finite_float(&value),
860            Err(_) => true,
861        }),
862        ParquetVariant::Float(v) => !v.is_finite(),
863        ParquetVariant::Double(v) => !v.is_finite(),
864        _ => false,
865    }
866}
867
868/// Converts a variant to a `serde_json::Value`, rendering non-finite floats as the
869/// strings `"NaN"` / `"Infinity"` / `"-Infinity"`, following `to_jsonb` on float types.
870fn variant_to_json_value_lossy(
871    variant: ParquetVariant<'_, '_>,
872) -> anyhow::Result<serde_json::Value> {
873    Ok(match variant {
874        ParquetVariant::Object(object) => {
875            let mut fields = serde_json::Map::new();
876            for field in object.iter_try() {
877                let (field_name, value) = field.context("failed to read variant object")?;
878                fields.insert(field_name.to_owned(), variant_to_json_value_lossy(value)?);
879            }
880            serde_json::Value::Object(fields)
881        }
882        ParquetVariant::List(list) => serde_json::Value::Array(
883            list.iter_try()
884                .map(|value| {
885                    variant_to_json_value_lossy(value.context("failed to read variant list")?)
886                })
887                .collect::<anyhow::Result<Vec<_>>>()?,
888        ),
889        ParquetVariant::Float(v) if !v.is_finite() => non_finite_to_json(f64::from(v)),
890        ParquetVariant::Double(v) if !v.is_finite() => non_finite_to_json(v),
891        other => other
892            .to_json_value()
893            .context("failed to convert variant to json")?,
894    })
895}
896
897fn non_finite_to_json(value: f64) -> serde_json::Value {
898    let s = if value.is_nan() {
899        "NaN"
900    } else if value == f64::INFINITY {
901        "Infinity"
902    } else {
903        "-Infinity"
904    };
905    serde_json::Value::String(s.to_owned())
906}
907
908fn append_decimal(value: Decimal, builder: &mut impl VariantBuilderExt) -> anyhow::Result<()> {
909    match value {
910        Decimal::Normalized(value) => {
911            let value = value.normalize();
912            let decimal = VariantDecimal16::try_new(value.mantissa(), value.scale() as u8)
913                .context("failed to encode decimal as variant")?;
914            builder.append_value(decimal);
915        }
916        Decimal::NaN | Decimal::PositiveInf | Decimal::NegativeInf => {
917            builder.append_value(value.to_text().as_str());
918        }
919    }
920    Ok(())
921}
922
923#[derive(Debug)]
924enum PathToken {
925    Field(String),
926    Index(i32),
927}
928
929fn parse_path(path: &str) -> anyhow::Result<Vec<PathToken>> {
930    let original_path = path;
931    let path = path.strip_prefix('$').unwrap_or(path);
932    let mut chars = path.chars().peekable();
933    let mut tokens = vec![];
934    while let Some(ch) = chars.next() {
935        match ch {
936            '.' => {
937                let mut field = String::new();
938                while let Some(&c) = chars.peek() {
939                    if c == '.' || c == '[' {
940                        break;
941                    }
942                    field.push(c);
943                    chars.next();
944                }
945                if field.is_empty() {
946                    bail!("invalid variant path `{original_path}`");
947                }
948                tokens.push(PathToken::Field(field));
949            }
950            '[' => {
951                if matches!(chars.peek(), Some('\'') | Some('"')) {
952                    let quote = chars.next().unwrap();
953                    let mut field = String::new();
954                    let mut closed = false;
955                    for c in chars.by_ref() {
956                        if c == quote {
957                            closed = true;
958                            break;
959                        }
960                        field.push(c);
961                    }
962                    if !closed || chars.next() != Some(']') {
963                        bail!("invalid variant path `{original_path}`");
964                    }
965                    tokens.push(PathToken::Field(field));
966                } else {
967                    let mut index = String::new();
968                    while let Some(&c) = chars.peek() {
969                        if c == ']' {
970                            break;
971                        }
972                        index.push(c);
973                        chars.next();
974                    }
975                    if chars.next() != Some(']') {
976                        bail!("invalid variant path `{original_path}`");
977                    }
978                    tokens.push(PathToken::Index(index.parse().with_context(|| {
979                        format!("invalid variant path `{original_path}`")
980                    })?));
981                }
982            }
983            _ if tokens.is_empty() => {
984                let mut field = String::from(ch);
985                while let Some(&c) = chars.peek() {
986                    if c == '.' || c == '[' {
987                        break;
988                    }
989                    field.push(c);
990                    chars.next();
991                }
992                tokens.push(PathToken::Field(field));
993            }
994            _ => bail!("invalid variant path `{original_path}`"),
995        }
996    }
997    Ok(tokens)
998}
999
1000#[cfg(test)]
1001mod tests {
1002    use super::*;
1003    use crate::array::StructValue;
1004    use crate::types::{Date, F32, F64, Int256, Interval, Serial, Time, Timestamptz};
1005
1006    fn scalar_variant(value: ScalarRefImpl<'_>, data_type: &DataType) -> VariantVal {
1007        VariantVal::try_from_scalar_ref(Some(value), data_type).unwrap()
1008    }
1009
1010    fn assert_same_variant(lhs: &VariantVal, rhs: &VariantVal) {
1011        assert_eq!(lhs, rhs);
1012        assert_eq!(
1013            lhs.as_scalar_ref().value_serialize(),
1014            rhs.as_scalar_ref().value_serialize()
1015        );
1016    }
1017
1018    fn assert_variant_parts(
1019        name: &str,
1020        variant: &VariantVal,
1021        expected_metadata_hex: &str,
1022        expected_value_hex: &str,
1023    ) {
1024        assert_eq!(
1025            hex::encode(variant.metadata()),
1026            expected_metadata_hex,
1027            "{name} metadata bytes changed"
1028        );
1029        assert_eq!(
1030            hex::encode(variant.value()),
1031            expected_value_hex,
1032            "{name} value bytes changed"
1033        );
1034    }
1035
1036    #[test]
1037    fn access_path_with_preparsed_path() {
1038        let v: VariantVal = r#"{"a":[{"b":7}]}"#.parse().unwrap();
1039        let path = VariantPath::parse("$.a[0].b").unwrap();
1040        assert_eq!(
1041            v.as_scalar_ref()
1042                .access_path_parsed(&path)
1043                .unwrap()
1044                .unwrap()
1045                .to_string(),
1046            "7"
1047        );
1048    }
1049
1050    #[test]
1051    fn path_access_supports_dot_and_bracket() {
1052        let v: VariantVal = r#"{"a":[{"b":7}]}"#.parse().unwrap();
1053        assert_eq!(
1054            v.as_scalar_ref()
1055                .access_path("$.a[0].b")
1056                .unwrap()
1057                .to_string(),
1058            "7"
1059        );
1060        assert_eq!(
1061            v.as_scalar_ref()
1062                .access_path("a[0]['b']")
1063                .unwrap()
1064                .to_string(),
1065            "7"
1066        );
1067        assert_eq!(
1068            v.as_scalar_ref()
1069                .access_path("$.a[-1].b")
1070                .unwrap()
1071                .to_string(),
1072            "7"
1073        );
1074        assert!(v.as_scalar_ref().access_path("$.a[-2]").is_none());
1075        assert!(v.as_scalar_ref().access_path("$.a[0].b[0]").is_none());
1076        assert!(
1077            v.as_scalar_ref()
1078                .access_path_strict("$.missing")
1079                .unwrap()
1080                .is_none()
1081        );
1082        assert!(v.as_scalar_ref().access_path_strict("$.").is_err());
1083        assert!(v.as_scalar_ref().access_path("$.").is_none());
1084    }
1085
1086    #[test]
1087    fn ord_matches_memcmp_encoding_order() {
1088        fn memcmp_encode(v: &VariantVal) -> Vec<u8> {
1089            let mut serializer = Serializer::new(vec![]);
1090            v.as_scalar_ref().memcmp_serialize(&mut serializer).unwrap();
1091            serializer.into_inner()
1092        }
1093
1094        // Include values with different metadata lengths.
1095        let mut values: Vec<VariantVal> = [
1096            "1",
1097            r#""short""#,
1098            r#"{"b":1,"c":1}"#,
1099            r#"{"aaaaaaaaaa":1}"#,
1100            r#"{"a":1}"#,
1101            "[1,2,3]",
1102        ]
1103        .iter()
1104        .map(|s| s.parse().unwrap())
1105        .collect();
1106        for f in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
1107            values.push(
1108                VariantVal::try_from_scalar_ref(
1109                    Some(ScalarRefImpl::Float64(F64::from(f))),
1110                    &DataType::Float64,
1111                )
1112                .unwrap(),
1113            );
1114        }
1115        for a in &values {
1116            for b in &values {
1117                assert_eq!(
1118                    a.cmp(b),
1119                    memcmp_encode(a).cmp(&memcmp_encode(b)),
1120                    "Ord and memcmp encoding disagree for {a} vs {b}"
1121                );
1122            }
1123        }
1124    }
1125
1126    #[test]
1127    fn deep_jsonb_to_variant_returns_error() {
1128        let shallow = JsonbVal::from(serde_json::Value::from(1));
1129        assert!(VariantVal::from_jsonb(shallow.as_scalar_ref()).is_ok());
1130
1131        // Deeper than serde_json's 128-level parser limit; must error, not panic.
1132        let mut json = serde_json::Value::from(1);
1133        for _ in 0..200 {
1134            json = serde_json::Value::Array(vec![json]);
1135        }
1136        let deep = JsonbVal::from(json);
1137        assert!(VariantVal::from_jsonb(deep.as_scalar_ref()).is_err());
1138    }
1139
1140    #[test]
1141    fn normalizes_non_finite_floats_to_canonical_bit_patterns() {
1142        let from_double_bits = |bits: u64| {
1143            let mut builder = VariantBuilder::new();
1144            builder.append_value(f64::from_bits(bits));
1145            let (metadata, value) = builder.finish();
1146            VariantVal::from_parts(&metadata, &value).unwrap()
1147        };
1148
1149        // Every NaN bit pattern (sign, signaling, payload) collapses to the canonical one.
1150        let canonical_nan = from_double_bits(CANONICAL_NAN_F64);
1151        assert_same_variant(&canonical_nan, &from_double_bits(0xfff8_0000_0000_0001));
1152        assert_same_variant(&canonical_nan, &from_double_bits(0x7ff0_0000_0000_0001));
1153
1154        let nan = canonical_nan.as_scalar_ref();
1155        assert_eq!(nan.type_name(), "double");
1156        assert_eq!(nan.to_text(), r#""NaN""#);
1157        assert_eq!(nan.to_jsonb().unwrap().to_string(), r#""NaN""#);
1158
1159        // Infinities have unique bit patterns and render as strings, like `to_jsonb(float8)`.
1160        let inf = from_double_bits(f64::INFINITY.to_bits());
1161        assert_eq!(inf.as_scalar_ref().type_name(), "double");
1162        assert_eq!(inf.as_scalar_ref().to_text(), r#""Infinity""#);
1163        assert_eq!(
1164            from_double_bits(f64::NEG_INFINITY.to_bits())
1165                .as_scalar_ref()
1166                .to_text(),
1167            r#""-Infinity""#
1168        );
1169
1170        // `-0.0` collapses to `+0.0` on this path too, matching `to_variant`.
1171        assert_same_variant(
1172            &from_double_bits((-0.0_f64).to_bits()),
1173            &from_double_bits(0.0_f64.to_bits()),
1174        );
1175
1176        // The same normalization applies to the f32 variant type.
1177        let from_float_bits = |bits: u32| {
1178            let mut builder = VariantBuilder::new();
1179            builder.append_value(f32::from_bits(bits));
1180            let (metadata, value) = builder.finish();
1181            VariantVal::from_parts(&metadata, &value).unwrap()
1182        };
1183        let float_nan = from_float_bits(0xffc0_0001);
1184        assert_same_variant(&float_nan, &from_float_bits(CANONICAL_NAN_F32));
1185        assert_eq!(float_nan.as_scalar_ref().type_name(), "float");
1186        assert_eq!(float_nan.as_scalar_ref().to_text(), r#""NaN""#);
1187
1188        // Non-finite floats nested in objects/lists render as strings as well.
1189        let mut builder = VariantBuilder::new();
1190        builder.add_field_name("x");
1191        let mut object = builder.new_object();
1192        object.insert("x", f64::NAN);
1193        object.finish();
1194        let (metadata, value) = builder.finish();
1195        let nested = VariantVal::from_parts(&metadata, &value).unwrap();
1196        assert_eq!(nested.as_scalar_ref().to_text(), r#"{"x":"NaN"}"#);
1197    }
1198
1199    #[test]
1200    fn untrusted_serialized_canonicalizes_non_finite_double() {
1201        let mut builder = VariantBuilder::new();
1202        builder.append_value(f64::from_bits(0xfff8_0000_0000_0001));
1203        let (metadata, value) = builder.finish();
1204        let bytes = VariantVal::from_canonical_parts(&metadata, &value)
1205            .as_scalar_ref()
1206            .value_serialize();
1207
1208        let canonicalized = VariantVal::from_serialized_untrusted(&bytes).unwrap();
1209        assert_eq!(canonicalized.as_scalar_ref().to_text(), r#""NaN""#);
1210        // The non-canonical NaN bit pattern was rewritten.
1211        assert_ne!(bytes, canonicalized.as_scalar_ref().value_serialize());
1212        // The trusted decode path only checks structure.
1213        assert!(VariantVal::value_deserialize(&bytes).is_some());
1214    }
1215
1216    #[test]
1217    fn untrusted_serialized_canonicalizes_unsorted_dictionary() {
1218        // Structurally valid but non-canonical: unsorted metadata dictionary `["b", "a"]`.
1219        let mut builder = VariantBuilder::new();
1220        builder.add_field_name("b");
1221        builder.add_field_name("a");
1222        let mut object = builder.new_object();
1223        object.insert("a", 1i64);
1224        object.insert("b", 2i64);
1225        object.finish();
1226        let (metadata, value) = builder.finish();
1227        let non_canonical = VariantVal::from_canonical_parts(&metadata, &value)
1228            .as_scalar_ref()
1229            .value_serialize();
1230
1231        let canonical: VariantVal = r#"{"a":1,"b":2}"#.parse().unwrap();
1232        let canonical_bytes = canonical.as_scalar_ref().value_serialize();
1233
1234        assert_ne!(non_canonical, canonical_bytes);
1235        let recanonicalized = VariantVal::from_serialized_untrusted(&non_canonical).unwrap();
1236        assert_same_variant(&recanonicalized, &canonical);
1237    }
1238
1239    #[test]
1240    fn maps_numeric_scalars_to_variant_types() {
1241        let json_int: VariantVal = "1".parse().unwrap();
1242        let int16 =
1243            VariantVal::try_from_scalar_ref(Some(ScalarRefImpl::Int16(1)), &DataType::Int16)
1244                .unwrap();
1245        let int32 =
1246            VariantVal::try_from_scalar_ref(Some(ScalarRefImpl::Int32(1)), &DataType::Int32)
1247                .unwrap();
1248        let int64 =
1249            VariantVal::try_from_scalar_ref(Some(ScalarRefImpl::Int64(1)), &DataType::Int64)
1250                .unwrap();
1251        let serial = VariantVal::try_from_scalar_ref(
1252            Some(ScalarRefImpl::Serial(Serial::from(1))),
1253            &DataType::Serial,
1254        )
1255        .unwrap();
1256
1257        assert_eq!(json_int.as_scalar_ref().type_name(), "int64");
1258        assert_eq!(int16.as_scalar_ref().type_name(), "int16");
1259        assert_eq!(int32.as_scalar_ref().type_name(), "int32");
1260        assert_eq!(int64.as_scalar_ref().type_name(), "int64");
1261        assert_ne!(json_int, int16);
1262        assert_ne!(json_int, int32);
1263        assert_same_variant(&json_int, &int64);
1264        assert_same_variant(&json_int, &serial);
1265
1266        let json_float: VariantVal = "1.5".parse().unwrap();
1267        let float32 = VariantVal::try_from_scalar_ref(
1268            Some(ScalarRefImpl::Float32(F32::from(1.5))),
1269            &DataType::Float32,
1270        )
1271        .unwrap();
1272        let float64 = VariantVal::try_from_scalar_ref(
1273            Some(ScalarRefImpl::Float64(F64::from(1.5))),
1274            &DataType::Float64,
1275        )
1276        .unwrap();
1277
1278        assert_eq!(json_float.as_scalar_ref().type_name(), "double");
1279        assert_same_variant(&json_float, &float32);
1280        assert_same_variant(&json_float, &float64);
1281
1282        let positive_zero: VariantVal = "0.0".parse().unwrap();
1283        let negative_zero = VariantVal::try_from_scalar_ref(
1284            Some(ScalarRefImpl::Float64(F64::from(-0.0))),
1285            &DataType::Float64,
1286        )
1287        .unwrap();
1288        assert_same_variant(&positive_zero, &negative_zero);
1289
1290        // Non-finite floats keep their type identity and render as strings.
1291        let sql_nan = VariantVal::try_from_scalar_ref(
1292            Some(ScalarRefImpl::Float64(F64::from(f64::NAN))),
1293            &DataType::Float64,
1294        )
1295        .unwrap();
1296        assert_eq!(sql_nan.as_scalar_ref().type_name(), "double");
1297        assert_eq!(sql_nan.as_scalar_ref().to_text(), r#""NaN""#);
1298    }
1299
1300    #[test]
1301    fn stores_sql_only_scalars_as_text() {
1302        let interval = Interval::from_month_day_usec(1, 2, 3);
1303        let text = interval.to_text();
1304        assert_same_variant(
1305            &scalar_variant(ScalarRefImpl::Interval(interval), &DataType::Interval),
1306            &scalar_variant(ScalarRefImpl::Utf8(&text), &DataType::Varchar),
1307        );
1308
1309        let int256 = Int256::from(1);
1310        let text = int256.as_scalar_ref().to_text();
1311        assert_same_variant(
1312            &scalar_variant(int256.as_scalar_ref().into(), &DataType::Int256),
1313            &scalar_variant(ScalarRefImpl::Utf8(&text), &DataType::Varchar),
1314        );
1315    }
1316
1317    #[test]
1318    fn maps_temporal_scalars_to_variant_temporal_types() {
1319        let date = VariantVal::try_from_scalar_ref(
1320            Some(ScalarRefImpl::Date(Date::from_ymd_uncheck(2024, 1, 2))),
1321            &DataType::Date,
1322        )
1323        .unwrap();
1324        let time = VariantVal::try_from_scalar_ref(
1325            Some(ScalarRefImpl::Time(Time::from_hms_micro_uncheck(
1326                3, 4, 5, 6000,
1327            ))),
1328            &DataType::Time,
1329        )
1330        .unwrap();
1331        let timestamp = VariantVal::try_from_scalar_ref(
1332            Some(ScalarRefImpl::Timestamp(
1333                Date::from_ymd_uncheck(2024, 1, 2).and_hms_micro_uncheck(3, 4, 5, 6000),
1334            )),
1335            &DataType::Timestamp,
1336        )
1337        .unwrap();
1338        let timestamptz = VariantVal::try_from_scalar_ref(
1339            Some(ScalarRefImpl::Timestamptz(
1340                Timestamptz::from_micros(1).unwrap(),
1341            )),
1342            &DataType::Timestamptz,
1343        )
1344        .unwrap();
1345
1346        assert_eq!(date.as_scalar_ref().type_name(), "date");
1347        assert_eq!(time.as_scalar_ref().type_name(), "time");
1348        assert_eq!(
1349            timestamp.as_scalar_ref().type_name(),
1350            "timestamp_ntz_micros"
1351        );
1352        assert_eq!(timestamptz.as_scalar_ref().type_name(), "timestamp_micros");
1353    }
1354
1355    #[test]
1356    fn canonicalizes_object_fields_across_construction_paths() {
1357        let json: VariantVal = r#"{"a":1,"c":2}"#.parse().unwrap();
1358        let struct_type = DataType::Struct(StructType::new(vec![
1359            ("c", DataType::Int64),
1360            ("a", DataType::Int64),
1361        ]));
1362        let struct_value = StructValue::new(vec![
1363            Some(ScalarRefImpl::Int64(2).into()),
1364            Some(ScalarRefImpl::Int64(1).into()),
1365        ]);
1366        let variant = VariantVal::try_from_scalar_ref(
1367            Some(ScalarRefImpl::Struct(struct_value.as_scalar_ref())),
1368            &struct_type,
1369        )
1370        .unwrap();
1371
1372        assert_eq!(json, variant);
1373        assert_eq!(
1374            json.as_scalar_ref().value_serialize(),
1375            variant.as_scalar_ref().value_serialize()
1376        );
1377    }
1378
1379    #[test]
1380    fn rejects_duplicate_struct_field_names() {
1381        let struct_type = DataType::Struct(StructType::new(vec![
1382            ("a", DataType::Int32),
1383            ("a", DataType::Varchar),
1384        ]));
1385        let struct_value = StructValue::new(vec![
1386            Some(ScalarRefImpl::Int32(1).into()),
1387            Some(ScalarRefImpl::Utf8("x").into()),
1388        ]);
1389        let err = VariantVal::try_from_scalar_ref(
1390            Some(ScalarRefImpl::Struct(struct_value.as_scalar_ref())),
1391            &struct_type,
1392        )
1393        .unwrap_err();
1394        assert!(
1395            err.to_string().contains("duplicate field name `a`"),
1396            "unexpected error: {err}"
1397        );
1398
1399        let outer_type = DataType::Struct(StructType::new(vec![("s", struct_type)]));
1400        let outer_value = StructValue::new(vec![Some(struct_value.into())]);
1401        assert!(
1402            VariantVal::try_from_scalar_ref(
1403                Some(ScalarRefImpl::Struct(outer_value.as_scalar_ref())),
1404                &outer_type,
1405            )
1406            .is_err()
1407        );
1408    }
1409
1410    #[test]
1411    fn canonicalizes_extracted_subvalues() {
1412        let root: VariantVal = r#"{"a":[{"b":1}]}"#.parse().unwrap();
1413        let extracted = root.as_scalar_ref().access_path("$.a[0]").unwrap();
1414        let parsed: VariantVal = r#"{"b":1}"#.parse().unwrap();
1415
1416        assert_same_variant(&extracted, &parsed);
1417    }
1418
1419    #[test]
1420    fn serializes_canonical_snapshot() {
1421        let v: VariantVal = r#"{"a":1,"c":[true,null]}"#.parse().unwrap();
1422        let bytes = v.as_scalar_ref().value_serialize();
1423        assert_eq!(
1424            hex::encode(&bytes),
1425            "07000000110200010261630202000100091018010000000000000003020001020400"
1426        );
1427        assert_eq!(
1428            u32::from_le_bytes(bytes[..METADATA_LEN_SIZE].try_into().unwrap()),
1429            7
1430        );
1431        assert_eq!(VariantVal::value_deserialize(&bytes).unwrap(), v);
1432        assert!(VariantVal::value_deserialize(&bytes[1..]).is_none());
1433        assert!(VariantRef::from_serialized(&bytes[1..]).is_none());
1434    }
1435
1436    #[test]
1437    fn golden_bytes_for_object_field_order_and_nested_values() {
1438        let ordered_object: VariantVal = r#"{"a":1,"c":2}"#.parse().unwrap();
1439        let reordered_object: VariantVal = r#"{"c":2,"a":1}"#.parse().unwrap();
1440        assert_same_variant(&ordered_object, &reordered_object);
1441        assert_variant_parts(
1442            "object",
1443            &reordered_object,
1444            "11020001026163",
1445            "02020001000912180100000000000000180200000000000000",
1446        );
1447
1448        let nested: VariantVal = r#"{"z":[{"b":true,"a":1},["short",null]],"a":{"c":"longish"}}"#
1449            .parse()
1450            .unwrap();
1451        assert_variant_parts(
1452            "nested",
1453            &nested,
1454            "110400010203046162637a",
1455            "02020003000d2f02010200081d6c6f6e67697368030200111d0202000100090a1801000000000000000403020006071573686f727400",
1456        );
1457    }
1458
1459    #[test]
1460    fn golden_bytes_for_scalar_variant_values() {
1461        assert_variant_parts(
1462            "short_string",
1463            &scalar_variant(ScalarRefImpl::Utf8("short"), &DataType::Varchar),
1464            "010000",
1465            "1573686f7274",
1466        );
1467        assert_variant_parts(
1468            "long_string",
1469            &scalar_variant(
1470                ScalarRefImpl::Utf8(
1471                    "0123456789012345678901234567890123456789012345678901234567890123",
1472                ),
1473                &DataType::Varchar,
1474            ),
1475            "010000",
1476            "404000000030313233343536373839303132333435363738393031323334353637383930313233343536373839303132333435363738393031323334353637383930313233",
1477        );
1478        assert_variant_parts(
1479            "int16",
1480            &scalar_variant(ScalarRefImpl::Int16(1), &DataType::Int16),
1481            "010000",
1482            "100100",
1483        );
1484        assert_variant_parts(
1485            "int32",
1486            &scalar_variant(ScalarRefImpl::Int32(1), &DataType::Int32),
1487            "010000",
1488            "1401000000",
1489        );
1490        assert_variant_parts(
1491            "int64",
1492            &scalar_variant(ScalarRefImpl::Int64(1), &DataType::Int64),
1493            "010000",
1494            "180100000000000000",
1495        );
1496        assert_variant_parts(
1497            "decimal",
1498            &scalar_variant(
1499                ScalarRefImpl::Decimal("123.45".parse().unwrap()),
1500                &DataType::Decimal,
1501            ),
1502            "010000",
1503            "280239300000000000000000000000000000",
1504        );
1505        assert_variant_parts(
1506            "date",
1507            &scalar_variant(
1508                ScalarRefImpl::Date(Date::from_ymd_uncheck(2024, 1, 2)),
1509                &DataType::Date,
1510            ),
1511            "010000",
1512            "2c0c4d0000",
1513        );
1514        assert_variant_parts(
1515            "time",
1516            &scalar_variant(
1517                ScalarRefImpl::Time(Time::from_hms_micro_uncheck(3, 4, 5, 6000)),
1518                &DataType::Time,
1519            ),
1520            "010000",
1521            "44b06a559202000000",
1522        );
1523        assert_variant_parts(
1524            "timestamp",
1525            &scalar_variant(
1526                ScalarRefImpl::Timestamp(
1527                    Date::from_ymd_uncheck(2024, 1, 2).and_hms_micro_uncheck(3, 4, 5, 6000),
1528                ),
1529                &DataType::Timestamp,
1530            ),
1531            "010000",
1532            "34b0ea4dc0ed0d0600",
1533        );
1534        assert_variant_parts(
1535            "timestamptz",
1536            &scalar_variant(
1537                ScalarRefImpl::Timestamptz(Timestamptz::from_micros(1).unwrap()),
1538                &DataType::Timestamptz,
1539            ),
1540            "010000",
1541            "300100000000000000",
1542        );
1543        assert_variant_parts(
1544            "binary",
1545            &scalar_variant(ScalarRefImpl::Bytea(&[0x12, 0x34, 0xff]), &DataType::Bytea),
1546            "010000",
1547            "3c030000001234ff",
1548        );
1549    }
1550}
risingwave_common/types/variant.rs

risingwave_common/types/
variant.rs
