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

risingwave_common/types/
variant.rs
