risingwave_common/types/
mod.rs

1// Copyright 2022 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
15//! Data types in RisingWave.
16
17// NOTE: When adding or modifying data types, remember to update the type matrix in
18// src/expr/macro/src/types.rs
19
20use std::fmt::Debug;
21use std::hash::Hash;
22use std::str::FromStr;
23
24use bytes::{Buf, BufMut, Bytes};
25use chrono::{Datelike, Timelike};
26use itertools::Itertools;
27use parse_display::{Display, FromStr};
28use paste::paste;
29use postgres_types::{FromSql, IsNull, ToSql, Type};
30use risingwave_common_estimate_size::{EstimateSize, ZeroHeapSize};
31use risingwave_pb::data::PbDataType;
32use risingwave_pb::data::data_type::PbTypeName;
33use rw_iter_util::ZipEqFast as _;
34use serde::{Deserialize, Serialize, Serializer};
35use strum_macros::EnumDiscriminants;
36use thiserror_ext::AsReport;
37
38use crate::array::{
39    ArrayBuilderImpl, ArrayError, ArrayResult, NULL_VAL_FOR_HASH, PrimitiveArrayItemType,
40};
41// Complex type's value is based on the array
42pub use crate::array::{
43    ListRef, ListValue, MapRef, MapValue, StructRef, StructValue, VectorRef, VectorVal,
44};
45use crate::cast::{str_to_bool, str_to_bytea};
46use crate::catalog::ColumnId;
47use crate::error::BoxedError;
48use crate::{
49    dispatch_data_types, dispatch_scalar_ref_variants, dispatch_scalar_variants, for_all_variants,
50};
51
52mod cow;
53mod datetime;
54mod decimal;
55mod fields;
56mod from_sql;
57mod interval;
58mod jsonb;
59mod list_type;
60mod macros;
61mod map_type;
62mod native_type;
63mod num256;
64mod ops;
65mod ordered;
66mod ordered_float;
67pub mod postgres_type;
68mod scalar_impl;
69mod sentinel;
70mod serial;
71mod struct_type;
72mod successor;
73mod timestamptz;
74mod to_binary;
75mod to_sql;
76mod to_text;
77mod variant;
78mod with_data_type;
79
80pub use fields::Fields;
81pub use risingwave_fields_derive::Fields;
82use risingwave_pb::id::TypedId;
83
84pub use self::cow::DatumCow;
85pub use self::datetime::{Date, Time, Timestamp};
86pub use self::decimal::{Decimal, PowError as DecimalPowError};
87pub use self::interval::{DateTimeField, Interval, IntervalDisplay, test_utils};
88pub use self::jsonb::{JsonbRef, JsonbVal};
89pub use self::list_type::ListType;
90pub use self::map_type::MapType;
91pub use self::native_type::*;
92pub use self::num256::{Int256, Int256Ref};
93pub use self::ops::{CheckedAdd, IsNegative};
94pub use self::ordered::*;
95pub use self::ordered_float::{FloatExt, IntoOrdered};
96pub use self::scalar_impl::*;
97pub use self::sentinel::Sentinelled;
98pub use self::serial::Serial;
99pub use self::struct_type::StructType;
100pub use self::successor::Successor;
101pub use self::timestamptz::*;
102pub use self::to_text::ToText;
103pub use self::variant::{VariantPath, VariantRef, VariantVal};
104pub use self::with_data_type::WithDataType;
105
106/// A 32-bit floating point type with total order.
107pub type F32 = ordered_float::OrderedFloat<f32>;
108
109/// A 64-bit floating point type with total order.
110pub type F64 = ordered_float::OrderedFloat<f64>;
111
112pub const DEBEZIUM_UNAVAILABLE_VALUE: &str = "__debezium_unavailable_value";
113
114// Pre-built JSON value for Debezium unavailable value to avoid rebuilding it every time
115pub static DEBEZIUM_UNAVAILABLE_JSON: std::sync::LazyLock<JsonbVal> =
116    std::sync::LazyLock::new(|| {
117        let mut builder = jsonbb::Builder::default();
118        builder.add_string(DEBEZIUM_UNAVAILABLE_VALUE);
119        JsonbVal(builder.finish())
120    });
121
122/// Magic per-element value used to build the Debezium unchanged-TOAST sentinel for
123/// pgvector columns. Picked because normal embeddings sit in a normalised range and
124/// having all elements simultaneously equal to `f32::MAX` is effectively impossible.
125pub const DEBEZIUM_UNAVAILABLE_VECTOR_ELEM: f32 = f32::MAX;
126
127/// Build a sentinel `VectorVal` of the given dimension to represent Debezium's
128/// unchanged-TOAST placeholder. The dimension must match the column's declared
129/// `vector(N)` size so it passes `check_datum_type` on the way through the
130/// `SourceStreamChunkBuilder`; the materialize executor recognises this sentinel
131/// by checking that every element equals `DEBEZIUM_UNAVAILABLE_VECTOR_ELEM`.
132pub fn debezium_unavailable_vector(size: usize) -> VectorVal {
133    VectorVal::from(
134        (0..size)
135            .map(|_| {
136                crate::array::Finite32::try_from(DEBEZIUM_UNAVAILABLE_VECTOR_ELEM)
137                    .expect("f32::MAX is finite")
138            })
139            .collect::<Vec<_>>(),
140    )
141}
142
143/// The set of datatypes that are supported in RisingWave.
144///
145/// # Trait implementations
146///
147/// - `EnumDiscriminants` generates [`DataTypeName`] enum with the same variants,
148///   but without data fields.
149/// - `FromStr` is only used internally for tests.
150///   The generated implementation isn't efficient, and doesn't handle whitespaces, etc.
151#[derive(Debug, Display, Clone, PartialEq, Eq, Hash, EnumDiscriminants, FromStr)]
152#[strum_discriminants(derive(Hash, Ord, PartialOrd))]
153#[strum_discriminants(name(DataTypeName))]
154#[strum_discriminants(vis(pub))]
155#[cfg_attr(test, strum_discriminants(derive(strum_macros::EnumIter)))]
156pub enum DataType {
157    #[display("boolean")]
158    #[from_str(regex = "(?i)^bool$|^boolean$")]
159    Boolean,
160    #[display("smallint")]
161    #[from_str(regex = "(?i)^smallint$|^int2$")]
162    Int16,
163    #[display("integer")]
164    #[from_str(regex = "(?i)^integer$|^int$|^int4$")]
165    Int32,
166    #[display("bigint")]
167    #[from_str(regex = "(?i)^bigint$|^int8$")]
168    Int64,
169    #[display("real")]
170    #[from_str(regex = "(?i)^real$|^float4$")]
171    Float32,
172    #[display("double precision")]
173    #[from_str(regex = "(?i)^double precision$|^float8$")]
174    Float64,
175    #[display("numeric")]
176    #[from_str(regex = "(?i)^numeric$|^decimal$")]
177    Decimal,
178    #[display("date")]
179    #[from_str(regex = "(?i)^date$")]
180    Date,
181    #[display("character varying")]
182    #[from_str(regex = "(?i)^character varying$|^varchar$")]
183    Varchar,
184    #[display("time without time zone")]
185    #[from_str(regex = "(?i)^time$|^time without time zone$")]
186    Time,
187    #[display("timestamp without time zone")]
188    #[from_str(regex = "(?i)^timestamp$|^timestamp without time zone$")]
189    Timestamp,
190    #[display("timestamp with time zone")]
191    #[from_str(regex = "(?i)^timestamptz$|^timestamp with time zone$")]
192    Timestamptz,
193    #[display("interval")]
194    #[from_str(regex = "(?i)^interval$")]
195    Interval,
196    #[display("{0}")]
197    #[from_str(regex = "(?i)^(?P<0>.+)$")]
198    Struct(StructType),
199    #[display("{0}")]
200    #[from_str(regex = "(?i)^(?P<0>.+)$")]
201    List(ListType),
202    #[display("bytea")]
203    #[from_str(regex = "(?i)^bytea$")]
204    Bytea,
205    #[display("jsonb")]
206    #[from_str(regex = "(?i)^jsonb$")]
207    Jsonb,
208    #[display("serial")]
209    #[from_str(regex = "(?i)^serial$")]
210    Serial,
211    #[display("rw_int256")]
212    #[from_str(regex = "(?i)^rw_int256$")]
213    Int256,
214    #[display("{0}")]
215    #[from_str(regex = "(?i)^(?P<0>.+)$")]
216    Map(MapType),
217    #[display("vector({0})")]
218    #[from_str(regex = "(?i)^vector\\((?P<0>.+)\\)$")]
219    Vector(usize),
220    #[display("variant")]
221    #[from_str(regex = "(?i)^variant$")]
222    Variant,
223}
224
225impl !PartialOrd for DataType {}
226
227impl ZeroHeapSize for DataType {}
228
229impl TryFrom<DataTypeName> for DataType {
230    type Error = &'static str;
231
232    fn try_from(type_name: DataTypeName) -> Result<Self, Self::Error> {
233        match type_name {
234            DataTypeName::Boolean => Ok(DataType::Boolean),
235            DataTypeName::Int16 => Ok(DataType::Int16),
236            DataTypeName::Int32 => Ok(DataType::Int32),
237            DataTypeName::Int64 => Ok(DataType::Int64),
238            DataTypeName::Int256 => Ok(DataType::Int256),
239            DataTypeName::Serial => Ok(DataType::Serial),
240            DataTypeName::Decimal => Ok(DataType::Decimal),
241            DataTypeName::Float32 => Ok(DataType::Float32),
242            DataTypeName::Float64 => Ok(DataType::Float64),
243            DataTypeName::Varchar => Ok(DataType::Varchar),
244            DataTypeName::Bytea => Ok(DataType::Bytea),
245            DataTypeName::Date => Ok(DataType::Date),
246            DataTypeName::Timestamp => Ok(DataType::Timestamp),
247            DataTypeName::Timestamptz => Ok(DataType::Timestamptz),
248            DataTypeName::Time => Ok(DataType::Time),
249            DataTypeName::Interval => Ok(DataType::Interval),
250            DataTypeName::Jsonb => Ok(DataType::Jsonb),
251            DataTypeName::Variant => Ok(DataType::Variant),
252            DataTypeName::Struct
253            | DataTypeName::List
254            | DataTypeName::Map
255            | DataTypeName::Vector => Err(
256                "Functions returning parameterized types cannot be inferred. Please use `FunctionCall::new_unchecked`.",
257            ),
258        }
259    }
260}
261
262impl From<&PbDataType> for DataType {
263    fn from(proto: &PbDataType) -> DataType {
264        match proto.get_type_name().expect("missing type field") {
265            PbTypeName::TypeUnspecified => unreachable!(),
266            PbTypeName::Int16 => DataType::Int16,
267            PbTypeName::Int32 => DataType::Int32,
268            PbTypeName::Int64 => DataType::Int64,
269            PbTypeName::Serial => DataType::Serial,
270            PbTypeName::Float => DataType::Float32,
271            PbTypeName::Double => DataType::Float64,
272            PbTypeName::Boolean => DataType::Boolean,
273            PbTypeName::Varchar => DataType::Varchar,
274            PbTypeName::Date => DataType::Date,
275            PbTypeName::Time => DataType::Time,
276            PbTypeName::Timestamp => DataType::Timestamp,
277            PbTypeName::Timestamptz => DataType::Timestamptz,
278            PbTypeName::Decimal => DataType::Decimal,
279            PbTypeName::Interval => DataType::Interval,
280            PbTypeName::Bytea => DataType::Bytea,
281            PbTypeName::Jsonb => DataType::Jsonb,
282            PbTypeName::Variant => DataType::Variant,
283            PbTypeName::Struct => {
284                let fields: Vec<DataType> = proto.field_type.iter().map(|f| f.into()).collect_vec();
285                let field_names: Vec<String> = proto.field_names.iter().cloned().collect_vec();
286                let field_ids = (proto.field_ids.iter().copied())
287                    .map(ColumnId::new)
288                    .collect_vec();
289
290                let mut struct_type = if proto.field_names.is_empty() {
291                    StructType::unnamed(fields)
292                } else {
293                    StructType::new(field_names.into_iter().zip_eq_fast(fields))
294                };
295                // `field_ids` is used for nested-schema evolution. Cases when `field_ids` is empty:
296                //
297                // 1. The data type is not associated with a table column, so we don't need to set it.
298                // 2. The column is created before nested-schema evolution is supported, thus is using
299                //    the old serialization format and does not have field ids.
300                // 3. This is an empty struct, which is always considered alterable, and setting ids
301                //    is a no-op.
302                if !field_ids.is_empty() {
303                    struct_type = struct_type.with_ids(field_ids);
304                }
305                struct_type.into()
306            }
307            PbTypeName::List => DataType::list(
308                // The first (and only) item is the list element type.
309                proto.field_type[0].clone().into(),
310            ),
311            PbTypeName::Map => {
312                // Map is physically the same as a list.
313                // So the first (and only) item is the list element type.
314                let list_entries_type: DataType = (&proto.field_type[0]).into();
315                DataType::Map(MapType::from_entries(list_entries_type))
316            }
317            PbTypeName::Vector => DataType::Vector(proto.precision as _),
318            PbTypeName::Int256 => DataType::Int256,
319        }
320    }
321}
322
323impl From<PbDataType> for DataType {
324    fn from(proto: PbDataType) -> DataType {
325        DataType::from(&proto)
326    }
327}
328
329impl From<DataTypeName> for PbTypeName {
330    fn from(type_name: DataTypeName) -> Self {
331        match type_name {
332            DataTypeName::Boolean => PbTypeName::Boolean,
333            DataTypeName::Int16 => PbTypeName::Int16,
334            DataTypeName::Int32 => PbTypeName::Int32,
335            DataTypeName::Int64 => PbTypeName::Int64,
336            DataTypeName::Serial => PbTypeName::Serial,
337            DataTypeName::Float32 => PbTypeName::Float,
338            DataTypeName::Float64 => PbTypeName::Double,
339            DataTypeName::Varchar => PbTypeName::Varchar,
340            DataTypeName::Date => PbTypeName::Date,
341            DataTypeName::Timestamp => PbTypeName::Timestamp,
342            DataTypeName::Timestamptz => PbTypeName::Timestamptz,
343            DataTypeName::Time => PbTypeName::Time,
344            DataTypeName::Interval => PbTypeName::Interval,
345            DataTypeName::Decimal => PbTypeName::Decimal,
346            DataTypeName::Bytea => PbTypeName::Bytea,
347            DataTypeName::Jsonb => PbTypeName::Jsonb,
348            DataTypeName::Variant => PbTypeName::Variant,
349            DataTypeName::Struct => PbTypeName::Struct,
350            DataTypeName::List => PbTypeName::List,
351            DataTypeName::Int256 => PbTypeName::Int256,
352            DataTypeName::Map => PbTypeName::Map,
353            DataTypeName::Vector => PbTypeName::Vector,
354        }
355    }
356}
357
358/// Convenient macros to generate match arms for [`DataType`].
359pub mod data_types {
360    use super::DataType;
361
362    /// Numeric [`DataType`]s supported to be `offset` of `RANGE` frame.
363    #[macro_export]
364    macro_rules! _range_frame_numeric_data_types {
365        () => {
366            DataType::Int16
367                | DataType::Int32
368                | DataType::Int64
369                | DataType::Float32
370                | DataType::Float64
371                | DataType::Decimal
372        };
373    }
374    pub use _range_frame_numeric_data_types as range_frame_numeric;
375
376    /// Date/time [`DataType`]s supported to be `offset` of `RANGE` frame.
377    #[macro_export]
378    macro_rules! _range_frame_datetime_data_types {
379        () => {
380            DataType::Date
381                | DataType::Time
382                | DataType::Timestamp
383                | DataType::Timestamptz
384                | DataType::Interval
385        };
386    }
387    pub use _range_frame_datetime_data_types as range_frame_datetime;
388
389    /// Data types that do not have inner fields.
390    #[macro_export]
391    macro_rules! _simple_data_types {
392        () => {
393            DataType::Boolean
394                | DataType::Int16
395                | DataType::Int32
396                | DataType::Int64
397                | DataType::Float32
398                | DataType::Float64
399                | DataType::Decimal
400                | DataType::Date
401                | DataType::Varchar
402                | DataType::Time
403                | DataType::Timestamp
404                | DataType::Timestamptz
405                | DataType::Interval
406                | DataType::Bytea
407                | DataType::Jsonb
408                | DataType::Variant
409                | DataType::Serial
410                | DataType::Int256
411                | DataType::Vector(_)
412        };
413    }
414    pub use _simple_data_types as simple;
415
416    /// Data types that have inner fields.
417    #[macro_export]
418    macro_rules! _composite_data_types {
419        () => {
420            DataType::Struct { .. } | DataType::List { .. } | DataType::Map { .. }
421        };
422    }
423    pub use _composite_data_types as composite;
424
425    /// Test that all data types are covered either by `simple!()` or `composite!()`.
426    fn _simple_composite_data_types_exhausted(dt: DataType) {
427        match dt {
428            simple!() => {}
429            composite!() => {}
430        }
431    }
432}
433
434impl DataType {
435    /// Same as pgvector; unsure how it was chosen there
436    /// <https://github.com/pgvector/pgvector/blob/v0.8.0/README.md#vector-type>
437    pub const VEC_MAX_SIZE: usize = 16000;
438
439    pub fn create_array_builder(&self, capacity: usize) -> ArrayBuilderImpl {
440        use crate::array::*;
441
442        dispatch_data_types!(self, [B = ArrayBuilder], {
443            B::with_type(capacity, self.clone()).into()
444        })
445    }
446
447    pub fn type_name(&self) -> DataTypeName {
448        DataTypeName::from(self)
449    }
450
451    pub fn prost_type_name(&self) -> PbTypeName {
452        self.type_name().into()
453    }
454
455    pub fn to_protobuf(&self) -> PbDataType {
456        let mut pb = PbDataType {
457            type_name: self.prost_type_name() as i32,
458            ..Default::default()
459        };
460        match self {
461            DataType::Struct(t) => {
462                if !t.is_unnamed() {
463                    // To be consistent with `From<&PbDataType>`,
464                    // we only set field names when it's a named struct.
465                    pb.field_names = t.names().map(|s| s.into()).collect();
466                }
467                pb.field_type = t.types().map(|f| f.to_protobuf()).collect();
468                if let Some(ids) = t.ids() {
469                    pb.field_ids = ids.map(|id| id.get_id()).collect();
470                }
471            }
472            DataType::List(list) => {
473                pb.field_type = vec![list.elem().to_protobuf()];
474            }
475            DataType::Map(map) => {
476                // Same as List<Struct<K,V>>
477                pb.field_type = vec![map.clone().into_struct().to_protobuf()];
478            }
479            DataType::Vector(size) => {
480                pb.precision = *size as _;
481            }
482            DataType::Boolean
483            | DataType::Int16
484            | DataType::Int32
485            | DataType::Int64
486            | DataType::Float32
487            | DataType::Float64
488            | DataType::Decimal
489            | DataType::Date
490            | DataType::Varchar
491            | DataType::Time
492            | DataType::Timestamp
493            | DataType::Timestamptz
494            | DataType::Interval
495            | DataType::Bytea
496            | DataType::Jsonb
497            | DataType::Variant
498            | DataType::Serial
499            | DataType::Int256 => (),
500        }
501        pb
502    }
503
504    pub fn is_numeric(&self) -> bool {
505        matches!(
506            self,
507            DataType::Int16
508                | DataType::Int32
509                | DataType::Int64
510                | DataType::Serial
511                | DataType::Float32
512                | DataType::Float64
513                | DataType::Decimal
514        )
515    }
516
517    /// Returns whether the data type does not have inner fields.
518    pub fn is_simple(&self) -> bool {
519        matches!(self, data_types::simple!())
520    }
521
522    /// Returns whether the data type has inner fields.
523    pub fn is_composite(&self) -> bool {
524        matches!(self, data_types::composite!())
525    }
526
527    pub fn is_array(&self) -> bool {
528        matches!(self, DataType::List(_))
529    }
530
531    pub fn is_struct(&self) -> bool {
532        matches!(self, DataType::Struct(_))
533    }
534
535    pub fn is_map(&self) -> bool {
536        matches!(self, DataType::Map(_))
537    }
538
539    pub fn is_int(&self) -> bool {
540        matches!(self, DataType::Int16 | DataType::Int32 | DataType::Int64)
541    }
542
543    /// Returns the output type of time window function on a given input type.
544    pub fn window_of(input: &DataType) -> Option<DataType> {
545        match input {
546            DataType::Timestamptz => Some(DataType::Timestamptz),
547            DataType::Timestamp | DataType::Date => Some(DataType::Timestamp),
548            _ => None,
549        }
550    }
551
552    pub fn as_struct(&self) -> &StructType {
553        match self {
554            DataType::Struct(t) => t,
555            t => panic!("expect struct type, got {t}"),
556        }
557    }
558
559    pub fn into_struct(self) -> StructType {
560        match self {
561            DataType::Struct(t) => t,
562            t => panic!("expect struct type, got {t}"),
563        }
564    }
565
566    pub fn as_map(&self) -> &MapType {
567        match self {
568            DataType::Map(t) => t,
569            t => panic!("expect map type, got {t}"),
570        }
571    }
572
573    pub fn into_map(self) -> MapType {
574        match self {
575            DataType::Map(t) => t,
576            t => panic!("expect map type, got {t}"),
577        }
578    }
579
580    pub fn as_list(&self) -> &ListType {
581        match self {
582            DataType::List(t) => t,
583            t => panic!("expect list type, got {t}"),
584        }
585    }
586
587    pub fn into_list(self) -> ListType {
588        match self {
589            DataType::List(t) => t,
590            t => panic!("expect list type, got {t}"),
591        }
592    }
593
594    /// Returns the inner element's type if `self` is a list type.
595    /// Equivalent to `self.as_list().elem()`.
596    pub fn as_list_elem(&self) -> &DataType {
597        self.as_list().elem()
598    }
599
600    /// Returns the inner element's type if `self` is a list type.
601    /// Equivalent to `self.into_list().into_elem()`.
602    pub fn into_list_elem(self) -> DataType {
603        self.into_list().into_elem()
604    }
605
606    /// Return a new type that removes the outer list, and get the innermost element type.
607    ///
608    /// Use [`DataType::as_list_elem`] if you only want the element type of a list.
609    ///
610    /// ```
611    /// use risingwave_common::types::DataType::*;
612    /// assert_eq!(Int32.list().unnest_list(), &Int32);
613    /// assert_eq!(Int32.list().list().unnest_list(), &Int32);
614    /// ```
615    pub fn unnest_list(&self) -> &Self {
616        match self {
617            DataType::List(list) => list.elem().unnest_list(),
618            _ => self,
619        }
620    }
621
622    /// Return the number of dimensions of this array/list type. Return `0` when this type is not an
623    /// array/list.
624    pub fn array_ndims(&self) -> usize {
625        let mut d = 0;
626        let mut t = self;
627        while let Self::List(list) = t {
628            d += 1;
629            t = list.elem();
630        }
631        d
632    }
633
634    /// Compares the datatype with another, ignoring nested field names and ids.
635    pub fn equals_datatype(&self, other: &DataType) -> bool {
636        match (self, other) {
637            (Self::Struct(s1), Self::Struct(s2)) => s1.equals_datatype(s2),
638            (Self::List(d1), Self::List(d2)) => d1.elem().equals_datatype(d2.elem()),
639            (Self::Map(m1), Self::Map(m2)) => {
640                m1.key().equals_datatype(m2.key()) && m1.value().equals_datatype(m2.value())
641            }
642            _ => self == other,
643        }
644    }
645
646    /// Whether a column with this data type can be altered to a new data type. This determines
647    /// the encoding of the column data.
648    ///
649    /// Returns...
650    /// - `None`, if the data type is simple or does not contain a struct type.
651    /// - `Some(true)`, if the data type contains a struct type with field ids ([`StructType::has_ids`]).
652    /// - `Some(false)`, if the data type contains a struct type without field ids.
653    pub fn can_alter(&self) -> Option<bool> {
654        match self {
655            data_types::simple!() => None,
656            DataType::Struct(struct_type) => {
657                // As long as we meet a struct type, we can check its `ids` field to determine if
658                // it can be altered.
659                let struct_can_alter = struct_type.has_ids();
660                // In debug build, we assert that once a struct type does (or does not) have ids,
661                // all its composite fields should have the same property.
662                if cfg!(debug_assertions) {
663                    for field in struct_type.types() {
664                        if let Some(field_can_alter) = field.can_alter() {
665                            assert_eq!(struct_can_alter, field_can_alter);
666                        }
667                    }
668                }
669                Some(struct_can_alter)
670            }
671
672            DataType::List(list_type) => list_type.elem().can_alter(),
673            DataType::Map(map_type) => {
674                debug_assert!(
675                    map_type.key().is_simple(),
676                    "unexpected key type of map {map_type:?}"
677                );
678                map_type.value().can_alter()
679            }
680        }
681    }
682
683    /// Whether this type is `VARIANT` or contains a nested `VARIANT`.
684    pub fn contains_variant(&self) -> bool {
685        matches!(self, DataType::Variant)
686            || match self {
687                DataType::List(list_type) => list_type.elem().contains_variant(),
688                DataType::Struct(struct_type) => {
689                    struct_type.types().any(DataType::contains_variant)
690                }
691                DataType::Map(map_type) => {
692                    map_type.key().contains_variant() || map_type.value().contains_variant()
693                }
694                // Listed rather than `_`: a new composite type answering `false` here would slip
695                // past every VARIANT-as-key gate.
696                data_types::simple!() => false,
697            }
698    }
699}
700
701impl From<StructType> for DataType {
702    fn from(value: StructType) -> Self {
703        Self::Struct(value)
704    }
705}
706
707impl From<DataType> for PbDataType {
708    fn from(data_type: DataType) -> Self {
709        data_type.to_protobuf()
710    }
711}
712
713mod private {
714    use super::*;
715
716    // Note: put pub trait inside a private mod just makes the name private,
717    // The trait methods will still be publicly available...
718    // a.k.a. ["Voldemort type"](https://rust-lang.github.io/rfcs/2145-type-privacy.html#lint-3-voldemort-types-its-reachable-but-i-cant-name-it)
719
720    /// Common trait bounds of scalar and scalar reference types.
721    ///
722    /// NOTE(rc): `Hash` is not in the trait bound list, it's implemented as [`ScalarRef::hash_scalar`].
723    pub trait ScalarBounds<Impl> = Debug
724        + Send
725        + Sync
726        + Clone
727        + PartialEq
728        + Eq
729        // in default ascending order
730        + PartialOrd
731        + Ord
732        + TryFrom<Impl, Error = ArrayError>
733        // `ScalarImpl`/`ScalarRefImpl`
734        + Into<Impl>;
735}
736
737/// `Scalar` is a trait over all possible owned types in the evaluation
738/// framework.
739///
740/// `Scalar` is reciprocal to `ScalarRef`. Use `as_scalar_ref` to get a
741/// reference which has the same lifetime as `self`.
742pub trait Scalar: private::ScalarBounds<ScalarImpl> + 'static {
743    /// Type for reference of `Scalar`
744    type ScalarRefType<'a>: ScalarRef<'a, ScalarType = Self> + 'a
745    where
746        Self: 'a;
747
748    /// Get a reference to current scalar.
749    fn as_scalar_ref(&self) -> Self::ScalarRefType<'_>;
750
751    fn to_scalar_value(self) -> ScalarImpl {
752        self.into()
753    }
754}
755
756/// `ScalarRef` is a trait over all possible references in the evaluation
757/// framework.
758///
759/// `ScalarRef` is reciprocal to `Scalar`. Use `to_owned_scalar` to get an
760/// owned scalar.
761pub trait ScalarRef<'a>: private::ScalarBounds<ScalarRefImpl<'a>> + 'a + Copy {
762    /// `ScalarType` is the owned type of current `ScalarRef`.
763    type ScalarType: Scalar<ScalarRefType<'a> = Self>;
764
765    /// Convert `ScalarRef` to an owned scalar.
766    fn to_owned_scalar(&self) -> Self::ScalarType;
767
768    /// A wrapped hash function to get the hash value for this scaler.
769    fn hash_scalar<H: std::hash::Hasher>(&self, state: &mut H);
770}
771
772/// Define `ScalarImpl` and `ScalarRefImpl` with macro.
773macro_rules! scalar_impl_enum {
774    ($( { $data_type:ident, $variant_name:ident, $suffix_name:ident, $scalar:ty, $scalar_ref:ty, $array:ty, $builder:ty } ),*) => {
775        /// `ScalarImpl` embeds all possible scalars in the evaluation framework.
776        ///
777        /// Note: `ScalarImpl` doesn't contain all information of its `DataType`,
778        /// so sometimes they need to be used together.
779        /// e.g., for `Struct`, we don't have the field names in the value.
780        ///
781        /// See `for_all_variants` for the definition.
782        #[derive(Debug, Clone, PartialEq, Eq, EstimateSize)]
783        pub enum ScalarImpl {
784            $( $variant_name($scalar) ),*
785        }
786
787        /// `ScalarRefImpl` embeds all possible scalar references in the evaluation
788        /// framework.
789        ///
790        /// Note: `ScalarRefImpl` doesn't contain all information of its `DataType`,
791        /// so sometimes they need to be used together.
792        /// e.g., for `Struct`, we don't have the field names in the value.
793        ///
794        /// See `for_all_variants` for the definition.
795        #[derive(Debug, Copy, Clone, PartialEq, Eq)]
796        pub enum ScalarRefImpl<'scalar> {
797            $( $variant_name($scalar_ref) ),*
798        }
799    };
800}
801
802for_all_variants! { scalar_impl_enum }
803
804// We MUST NOT implement `Ord` for `ScalarImpl` because that will make `Datum` derive an incorrect
805// default `Ord`. To get a default-ordered `ScalarImpl`/`ScalarRefImpl`/`Datum`/`DatumRef`, you can
806// use `DefaultOrdered<T>`. If non-default order is needed, please refer to `sort_util`.
807impl !PartialOrd for ScalarImpl {}
808impl !PartialOrd for ScalarRefImpl<'_> {}
809
810pub type Datum = Option<ScalarImpl>;
811pub type DatumRef<'a> = Option<ScalarRefImpl<'a>>;
812
813/// This trait is to implement `to_owned_datum` for `Option<ScalarImpl>`
814pub trait ToOwnedDatum {
815    /// Convert the datum to an owned [`Datum`].
816    fn to_owned_datum(self) -> Datum;
817}
818
819impl ToOwnedDatum for &Datum {
820    #[inline(always)]
821    fn to_owned_datum(self) -> Datum {
822        self.clone()
823    }
824}
825
826impl<T: Into<ScalarImpl>> ToOwnedDatum for T {
827    #[inline(always)]
828    fn to_owned_datum(self) -> Datum {
829        Some(self.into())
830    }
831}
832
833impl<T: Into<ScalarImpl>> ToOwnedDatum for Option<T> {
834    #[inline(always)]
835    fn to_owned_datum(self) -> Datum {
836        self.map(Into::into)
837    }
838}
839
840impl<const N: usize> From<TypedId<N, u32>> for ScalarImpl {
841    fn from(value: TypedId<N, u32>) -> Self {
842        value.as_i32_id().into()
843    }
844}
845
846impl<const N: usize> From<TypedId<N, u64>> for ScalarImpl {
847    fn from(value: TypedId<N, u64>) -> Self {
848        value.as_i64_id().into()
849    }
850}
851
852#[auto_impl::auto_impl(&)]
853pub trait ToDatumRef: PartialEq + Eq + Debug + Send + Sync {
854    /// Convert the datum to [`DatumRef`].
855    fn to_datum_ref(&self) -> DatumRef<'_>;
856}
857
858impl ToDatumRef for Datum {
859    #[inline(always)]
860    fn to_datum_ref(&self) -> DatumRef<'_> {
861        self.as_ref().map(|d| d.as_scalar_ref_impl())
862    }
863}
864impl ToDatumRef for Option<&ScalarImpl> {
865    #[inline(always)]
866    fn to_datum_ref(&self) -> DatumRef<'_> {
867        self.map(|d| d.as_scalar_ref_impl())
868    }
869}
870impl ToDatumRef for DatumRef<'_> {
871    #[inline(always)]
872    fn to_datum_ref(&self) -> DatumRef<'_> {
873        *self
874    }
875}
876
877/// To make sure there is `as_scalar_ref` for all scalar ref types.
878/// See <https://github.com/risingwavelabs/risingwave/pull/9977/files#r1208972881>
879///
880/// This is used by the expr macro.
881pub trait SelfAsScalarRef {
882    fn as_scalar_ref(&self) -> Self;
883}
884macro_rules! impl_self_as_scalar_ref {
885    ($($t:ty),*) => {
886        $(
887            impl SelfAsScalarRef for $t {
888                fn as_scalar_ref(&self) -> Self {
889                    *self
890                }
891            }
892        )*
893    };
894}
895impl_self_as_scalar_ref! { &str, &[u8], Int256Ref<'_>, JsonbRef<'_>, VariantRef<'_>, ListRef<'_>, StructRef<'_>, ScalarRefImpl<'_>, MapRef<'_> }
896
897/// `for_all_native_types` includes all native variants of our scalar types.
898///
899/// Specifically, it doesn't support u8/u16/u32/u64.
900#[macro_export]
901macro_rules! for_all_native_types {
902    ($macro:ident) => {
903        $macro! {
904            { i16, Int16, read_i16 },
905            { i32, Int32, read_i32 },
906            { i64, Int64, read_i64 },
907            { Serial, Serial, read_i64 },
908            { $crate::types::F32, Float32, read_f32 },
909            { $crate::types::F64, Float64, read_f64 }
910        }
911    };
912}
913
914/// `impl_convert` implements several conversions for `Scalar`.
915/// * `Scalar <-> ScalarImpl` with `From` and `TryFrom` trait.
916/// * `ScalarRef <-> ScalarRefImpl` with `From` and `TryFrom` trait.
917/// * `&ScalarImpl -> &Scalar` with `impl.as_int16()`.
918/// * `ScalarImpl -> Scalar` with `impl.into_int16()`.
919macro_rules! impl_convert {
920    ($( { $data_type:ident, $variant_name:ident, $suffix_name:ident, $scalar:ty, $scalar_ref:ty, $array:ty, $builder:ty } ),*) => {
921        $(
922            impl From<$scalar> for ScalarImpl {
923                fn from(val: $scalar) -> Self {
924                    ScalarImpl::$variant_name(val)
925                }
926            }
927
928            impl TryFrom<ScalarImpl> for $scalar {
929                type Error = ArrayError;
930
931                fn try_from(val: ScalarImpl) -> ArrayResult<Self> {
932                    match val {
933                        ScalarImpl::$variant_name(scalar) => Ok(scalar),
934                        other_scalar => bail!("cannot convert ScalarImpl::{} to concrete type", other_scalar.get_ident()),
935                    }
936                }
937            }
938
939            impl <'scalar> From<$scalar_ref> for ScalarRefImpl<'scalar> {
940                fn from(val: $scalar_ref) -> Self {
941                    ScalarRefImpl::$variant_name(val)
942                }
943            }
944
945            impl <'scalar> TryFrom<ScalarRefImpl<'scalar>> for $scalar_ref {
946                type Error = ArrayError;
947
948                fn try_from(val: ScalarRefImpl<'scalar>) -> ArrayResult<Self> {
949                    match val {
950                        ScalarRefImpl::$variant_name(scalar_ref) => Ok(scalar_ref),
951                        other_scalar => bail!("cannot convert ScalarRefImpl::{} to concrete type {}", other_scalar.get_ident(), stringify!($variant_name)),
952                    }
953                }
954            }
955
956            paste! {
957                impl ScalarImpl {
958                    /// # Panics
959                    /// If the scalar is not of the expected type.
960                    pub fn [<as_ $suffix_name>](&self) -> &$scalar {
961                        match self {
962                            Self::$variant_name(scalar) => scalar,
963                            other_scalar => panic!("cannot convert ScalarImpl::{} to concrete type {}", other_scalar.get_ident(), stringify!($variant_name))
964                        }
965                    }
966
967                    /// # Panics
968                    /// If the scalar is not of the expected type.
969                    pub fn [<into_ $suffix_name>](self) -> $scalar {
970                        match self {
971                            Self::$variant_name(scalar) => scalar,
972                            other_scalar =>  panic!("cannot convert ScalarImpl::{} to concrete type {}", other_scalar.get_ident(), stringify!($variant_name))
973                        }
974                    }
975                }
976
977                impl <'scalar> ScalarRefImpl<'scalar> {
978                    /// # Panics
979                    /// If the scalar is not of the expected type.
980                    pub fn [<into_ $suffix_name>](self) -> $scalar_ref {
981                        match self {
982                            Self::$variant_name(inner) => inner,
983                            other_scalar => panic!("cannot convert ScalarRefImpl::{} to concrete type {}", other_scalar.get_ident(), stringify!($variant_name))
984                        }
985                    }
986                }
987            }
988        )*
989    };
990}
991
992for_all_variants! { impl_convert }
993
994// Implement `From<raw float>` for `ScalarImpl::Float` as a sugar.
995impl From<f32> for ScalarImpl {
996    fn from(f: f32) -> Self {
997        Self::Float32(f.into())
998    }
999}
1000impl From<f64> for ScalarImpl {
1001    fn from(f: f64) -> Self {
1002        Self::Float64(f.into())
1003    }
1004}
1005
1006// Implement `From<string like>` for `ScalarImpl::Utf8` as a sugar.
1007impl From<String> for ScalarImpl {
1008    fn from(s: String) -> Self {
1009        Self::Utf8(s.into_boxed_str())
1010    }
1011}
1012impl From<&str> for ScalarImpl {
1013    fn from(s: &str) -> Self {
1014        Self::Utf8(s.into())
1015    }
1016}
1017impl From<&String> for ScalarImpl {
1018    fn from(s: &String) -> Self {
1019        Self::Utf8(s.as_str().into())
1020    }
1021}
1022impl TryFrom<ScalarImpl> for String {
1023    type Error = ArrayError;
1024
1025    fn try_from(val: ScalarImpl) -> ArrayResult<Self> {
1026        match val {
1027            ScalarImpl::Utf8(s) => Ok(s.into()),
1028            other_scalar => bail!(
1029                "cannot convert ScalarImpl::{} to concrete type",
1030                other_scalar.get_ident()
1031            ),
1032        }
1033    }
1034}
1035
1036impl From<char> for ScalarImpl {
1037    fn from(c: char) -> Self {
1038        Self::Utf8(c.to_string().into())
1039    }
1040}
1041
1042impl From<&[u8]> for ScalarImpl {
1043    fn from(s: &[u8]) -> Self {
1044        Self::Bytea(s.into())
1045    }
1046}
1047
1048impl From<JsonbRef<'_>> for ScalarImpl {
1049    fn from(jsonb: JsonbRef<'_>) -> Self {
1050        Self::Jsonb(jsonb.to_owned_scalar())
1051    }
1052}
1053
1054impl From<VariantRef<'_>> for ScalarImpl {
1055    fn from(variant: VariantRef<'_>) -> Self {
1056        Self::Variant(variant.to_owned_scalar())
1057    }
1058}
1059
1060impl<T: PrimitiveArrayItemType> From<Vec<T>> for ScalarImpl {
1061    fn from(v: Vec<T>) -> Self {
1062        Self::List(v.into_iter().collect())
1063    }
1064}
1065
1066impl<T: PrimitiveArrayItemType> From<Vec<Option<T>>> for ScalarImpl {
1067    fn from(v: Vec<Option<T>>) -> Self {
1068        Self::List(v.into_iter().collect())
1069    }
1070}
1071
1072impl From<Vec<String>> for ScalarImpl {
1073    fn from(v: Vec<String>) -> Self {
1074        Self::List(v.iter().map(|s| s.as_str()).collect())
1075    }
1076}
1077
1078impl From<Vec<u8>> for ScalarImpl {
1079    fn from(v: Vec<u8>) -> Self {
1080        Self::Bytea(v.into())
1081    }
1082}
1083
1084impl From<Bytes> for ScalarImpl {
1085    fn from(v: Bytes) -> Self {
1086        Self::Bytea(v.as_ref().into())
1087    }
1088}
1089
1090impl From<ListRef<'_>> for ScalarImpl {
1091    fn from(list: ListRef<'_>) -> Self {
1092        Self::List(list.to_owned_scalar())
1093    }
1094}
1095
1096impl ScalarImpl {
1097    /// Creates a scalar from pgwire "BINARY" format.
1098    ///
1099    /// The counterpart of [`to_binary::ToBinary`].
1100    pub fn from_binary(bytes: &Bytes, data_type: &DataType) -> Result<Self, BoxedError> {
1101        let res = match data_type {
1102            DataType::Varchar => Self::Utf8(String::from_sql(&Type::VARCHAR, bytes)?.into()),
1103            DataType::Bytea => Self::Bytea(Vec::<u8>::from_sql(&Type::BYTEA, bytes)?.into()),
1104            DataType::Boolean => Self::Bool(bool::from_sql(&Type::BOOL, bytes)?),
1105            DataType::Int16 => Self::Int16(i16::from_sql(&Type::INT2, bytes)?),
1106            DataType::Int32 => Self::Int32(i32::from_sql(&Type::INT4, bytes)?),
1107            DataType::Int64 => Self::Int64(i64::from_sql(&Type::INT8, bytes)?),
1108            DataType::Serial => Self::Serial(Serial::from(i64::from_sql(&Type::INT8, bytes)?)),
1109            DataType::Float32 => Self::Float32(f32::from_sql(&Type::FLOAT4, bytes)?.into()),
1110            DataType::Float64 => Self::Float64(f64::from_sql(&Type::FLOAT8, bytes)?.into()),
1111            DataType::Decimal => {
1112                Self::Decimal(rust_decimal::Decimal::from_sql(&Type::NUMERIC, bytes)?.into())
1113            }
1114            DataType::Date => Self::Date(chrono::NaiveDate::from_sql(&Type::DATE, bytes)?.into()),
1115            DataType::Time => Self::Time(chrono::NaiveTime::from_sql(&Type::TIME, bytes)?.into()),
1116            DataType::Timestamp => {
1117                Self::Timestamp(chrono::NaiveDateTime::from_sql(&Type::TIMESTAMP, bytes)?.into())
1118            }
1119            DataType::Timestamptz => Self::Timestamptz(
1120                chrono::DateTime::<chrono::Utc>::from_sql(&Type::TIMESTAMPTZ, bytes)?.into(),
1121            ),
1122            DataType::Interval => Self::Interval(Interval::from_sql(&Type::INTERVAL, bytes)?),
1123            DataType::Jsonb => Self::Jsonb(
1124                JsonbVal::value_deserialize(bytes)
1125                    .ok_or_else(|| "invalid value of Jsonb".to_owned())?,
1126            ),
1127            // pgwire binary parameters are untrusted and must be re-canonicalized.
1128            DataType::Variant => Self::Variant(VariantVal::from_serialized_untrusted(bytes)?),
1129            DataType::Int256 => Self::Int256(Int256::from_binary(bytes)?),
1130            DataType::Vector(_) | DataType::Struct(_) | DataType::List(_) | DataType::Map(_) => {
1131                return Err(format!("unsupported data type: {}", data_type).into());
1132            }
1133        };
1134        Ok(res)
1135    }
1136
1137    /// Creates a scalar from pgwire "TEXT" format.
1138    ///
1139    /// The counterpart of [`ToText`].
1140    pub fn from_text(s: &str, data_type: &DataType) -> Result<Self, BoxedError> {
1141        Ok(match data_type {
1142            DataType::Boolean => str_to_bool(s)?.into(),
1143            DataType::Int16 => i16::from_str(s)?.into(),
1144            DataType::Int32 => i32::from_str(s)?.into(),
1145            DataType::Int64 => i64::from_str(s)?.into(),
1146            DataType::Int256 => Int256::from_str(s)?.into(),
1147            DataType::Serial => Serial::from(i64::from_str(s)?).into(),
1148            DataType::Decimal => Decimal::from_str(s)?.into(),
1149            DataType::Float32 => F32::from_str(s)?.into(),
1150            DataType::Float64 => F64::from_str(s)?.into(),
1151            DataType::Varchar => s.into(),
1152            DataType::Date => Date::from_str(s)?.into(),
1153            DataType::Timestamp => Timestamp::from_str(s)?.into(),
1154            // We only handle the case with timezone here, and leave the implicit session timezone case
1155            // for later phase.
1156            DataType::Timestamptz => Timestamptz::from_str(s)?.into(),
1157            DataType::Time => Time::from_str(s)?.into(),
1158            DataType::Interval => Interval::from_str(s)?.into(),
1159            DataType::List(_) => ListValue::from_str(s, data_type)?.into(),
1160            DataType::Struct(st) => StructValue::from_str(s, st)?.into(),
1161            DataType::Jsonb => JsonbVal::from_str(s)?.into(),
1162            DataType::Variant => VariantVal::from_str(s)?.into(),
1163            DataType::Bytea => {
1164                let mut buf = Vec::new();
1165                str_to_bytea(s, &mut buf)?;
1166                buf.into()
1167            }
1168            DataType::Vector(size) => VectorVal::from_text(s, *size)?.into(),
1169            DataType::Map(_m) => return Err("map from text is not supported".into()),
1170        })
1171    }
1172
1173    pub fn from_text_for_test(s: &str, data_type: &DataType) -> Result<Self, BoxedError> {
1174        Ok(match data_type {
1175            DataType::Map(map_type) => MapValue::from_str_for_test(s, map_type)?.into(),
1176            _ => ScalarImpl::from_text(s, data_type)?,
1177        })
1178    }
1179}
1180
1181impl From<ScalarRefImpl<'_>> for ScalarImpl {
1182    fn from(scalar_ref: ScalarRefImpl<'_>) -> Self {
1183        scalar_ref.into_scalar_impl()
1184    }
1185}
1186
1187impl<'a> From<&'a ScalarImpl> for ScalarRefImpl<'a> {
1188    fn from(scalar: &'a ScalarImpl) -> Self {
1189        scalar.as_scalar_ref_impl()
1190    }
1191}
1192
1193impl ScalarImpl {
1194    /// Converts [`ScalarImpl`] to [`ScalarRefImpl`]
1195    pub fn as_scalar_ref_impl(&self) -> ScalarRefImpl<'_> {
1196        dispatch_scalar_variants!(self, inner, { inner.as_scalar_ref().into() })
1197    }
1198}
1199
1200impl ScalarRefImpl<'_> {
1201    /// Converts [`ScalarRefImpl`] to [`ScalarImpl`]
1202    pub fn into_scalar_impl(self) -> ScalarImpl {
1203        dispatch_scalar_ref_variants!(self, inner, { inner.to_owned_scalar().into() })
1204    }
1205}
1206
1207impl Hash for ScalarImpl {
1208    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
1209        dispatch_scalar_variants!(self, inner, { inner.as_scalar_ref().hash_scalar(state) })
1210    }
1211}
1212
1213impl Hash for ScalarRefImpl<'_> {
1214    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
1215        dispatch_scalar_ref_variants!(self, inner, { inner.hash_scalar(state) })
1216    }
1217}
1218
1219/// Feeds the raw scalar reference of `datum` to the given `state`, which should behave the same
1220/// as [`crate::array::Array::hash_at`], where NULL value will be carefully handled.
1221///
1222/// **FIXME**: the result of this function might be different from [`std::hash::Hash`] due to the
1223/// type alias of `DatumRef = Option<_>`, we should manually implement [`std::hash::Hash`] for
1224/// [`DatumRef`] in the future when it becomes a newtype. (#477)
1225#[inline(always)]
1226pub fn hash_datum(datum: impl ToDatumRef, state: &mut impl std::hash::Hasher) {
1227    match datum.to_datum_ref() {
1228        Some(scalar_ref) => scalar_ref.hash(state),
1229        None => NULL_VAL_FOR_HASH.hash(state),
1230    }
1231}
1232
1233impl ScalarRefImpl<'_> {
1234    pub fn binary_format(&self, data_type: &DataType) -> to_binary::Result<Bytes> {
1235        use self::to_binary::ToBinary;
1236        self.to_binary_with_type(data_type)
1237    }
1238
1239    pub fn text_format(&self, data_type: &DataType) -> String {
1240        self.to_text_with_type(data_type)
1241    }
1242
1243    /// Serialize the scalar into the `memcomparable` format.
1244    pub fn serialize(
1245        &self,
1246        ser: &mut memcomparable::Serializer<impl BufMut>,
1247    ) -> memcomparable::Result<()> {
1248        match self {
1249            Self::Int16(v) => v.serialize(ser)?,
1250            Self::Int32(v) => v.serialize(ser)?,
1251            Self::Int64(v) => v.serialize(ser)?,
1252            Self::Serial(v) => v.serialize(ser)?,
1253            Self::Float32(v) => v.serialize(ser)?,
1254            Self::Float64(v) => v.serialize(ser)?,
1255            Self::Utf8(v) => v.serialize(ser)?,
1256            Self::Bytea(v) => ser.serialize_bytes(v)?,
1257            Self::Bool(v) => v.serialize(ser)?,
1258            Self::Decimal(v) => ser.serialize_decimal((*v).into())?,
1259            Self::Interval(v) => v.serialize(ser)?,
1260            Self::Date(v) => v.0.num_days_from_ce().serialize(ser)?,
1261            Self::Timestamp(v) => {
1262                v.0.and_utc().timestamp().serialize(&mut *ser)?;
1263                v.0.and_utc().timestamp_subsec_nanos().serialize(ser)?;
1264            }
1265            Self::Timestamptz(v) => v.serialize(ser)?,
1266            Self::Time(v) => {
1267                v.0.num_seconds_from_midnight().serialize(&mut *ser)?;
1268                v.0.nanosecond().serialize(ser)?;
1269            }
1270            Self::Int256(v) => v.memcmp_serialize(ser)?,
1271            Self::Jsonb(v) => v.memcmp_serialize(ser)?,
1272            Self::Variant(v) => v.memcmp_serialize(ser)?,
1273            Self::Struct(v) => v.memcmp_serialize(ser)?,
1274            Self::List(v) => v.memcmp_serialize(ser)?,
1275            Self::Map(v) => v.memcmp_serialize(ser)?,
1276            Self::Vector(v) => v.memcmp_serialize(ser)?,
1277        };
1278        Ok(())
1279    }
1280}
1281
1282impl ScalarImpl {
1283    /// Serialize the scalar into the `memcomparable` format.
1284    pub fn serialize(
1285        &self,
1286        ser: &mut memcomparable::Serializer<impl BufMut>,
1287    ) -> memcomparable::Result<()> {
1288        self.as_scalar_ref_impl().serialize(ser)
1289    }
1290
1291    /// Deserialize the scalar from the `memcomparable` format.
1292    pub fn deserialize(
1293        ty: &DataType,
1294        de: &mut memcomparable::Deserializer<impl Buf>,
1295    ) -> memcomparable::Result<Self> {
1296        use DataType as Ty;
1297        Ok(match ty {
1298            Ty::Int16 => Self::Int16(i16::deserialize(de)?),
1299            Ty::Int32 => Self::Int32(i32::deserialize(de)?),
1300            Ty::Int64 => Self::Int64(i64::deserialize(de)?),
1301            Ty::Int256 => Self::Int256(Int256::memcmp_deserialize(de)?),
1302            Ty::Serial => Self::Serial(Serial::from(i64::deserialize(de)?)),
1303            Ty::Float32 => Self::Float32(f32::deserialize(de)?.into()),
1304            Ty::Float64 => Self::Float64(f64::deserialize(de)?.into()),
1305            Ty::Varchar => Self::Utf8(Box::<str>::deserialize(de)?),
1306            Ty::Bytea => Self::Bytea(serde_bytes::ByteBuf::deserialize(de)?.into_vec().into()),
1307            Ty::Boolean => Self::Bool(bool::deserialize(de)?),
1308            Ty::Decimal => Self::Decimal(de.deserialize_decimal()?.into()),
1309            Ty::Interval => Self::Interval(Interval::deserialize(de)?),
1310            Ty::Time => Self::Time({
1311                let secs = u32::deserialize(&mut *de)?;
1312                let nano = u32::deserialize(de)?;
1313                Time::with_secs_nano(secs, nano)
1314                    .map_err(|e| memcomparable::Error::Message(e.to_report_string()))?
1315            }),
1316            Ty::Timestamp => Self::Timestamp({
1317                let secs = i64::deserialize(&mut *de)?;
1318                let nsecs = u32::deserialize(de)?;
1319                Timestamp::with_secs_nsecs(secs, nsecs)
1320                    .map_err(|e| memcomparable::Error::Message(e.to_report_string()))?
1321            }),
1322            Ty::Timestamptz => Self::Timestamptz(Timestamptz::deserialize(de)?),
1323            Ty::Date => Self::Date({
1324                let days = i32::deserialize(de)?;
1325                Date::with_days_since_ce(days)
1326                    .map_err(|e| memcomparable::Error::Message(e.to_report_string()))?
1327            }),
1328            Ty::Jsonb => Self::Jsonb(JsonbVal::memcmp_deserialize(de)?),
1329            Ty::Variant => Self::Variant(VariantVal::memcmp_deserialize(de)?),
1330            Ty::Struct(t) => StructValue::memcmp_deserialize(t.types(), de)?.to_scalar_value(),
1331            Ty::List(t) => ListValue::memcmp_deserialize(t, de)?.to_scalar_value(),
1332            Ty::Map(t) => MapValue::memcmp_deserialize(t, de)?.to_scalar_value(),
1333            Ty::Vector(dimension) => {
1334                VectorVal::memcmp_deserialize(*dimension, de)?.to_scalar_value()
1335            }
1336        })
1337    }
1338
1339    pub fn as_integral(&self) -> i64 {
1340        match self {
1341            Self::Int16(v) => *v as i64,
1342            Self::Int32(v) => *v as i64,
1343            Self::Int64(v) => *v,
1344            _ => panic!(
1345                "Can't convert ScalarImpl::{} to a integral",
1346                self.get_ident()
1347            ),
1348        }
1349    }
1350}
1351
1352/// Returns whether the `literal` matches the `data_type`.
1353pub fn literal_type_match(data_type: &DataType, literal: Option<&ScalarImpl>) -> bool {
1354    match literal {
1355        None => true,
1356        Some(scalar) => scalar_ref_type_match(data_type, scalar.as_scalar_ref_impl()),
1357    }
1358}
1359
1360/// Returns whether the scalar ref matches the `data_type`.
1361///
1362/// This is a lightweight "shape check" intended for callers that need to avoid panics on
1363/// malformed input. For nested types, it checks element/field types recursively.
1364pub fn scalar_ref_type_match(data_type: &DataType, scalar: ScalarRefImpl<'_>) -> bool {
1365    match (data_type, scalar) {
1366        (DataType::List(list_type), ScalarRefImpl::List(v)) => {
1367            v.elem_type().equals_datatype(list_type.elem())
1368        }
1369        (DataType::Map(map_type), ScalarRefImpl::Map(v)) => v
1370            .inner()
1371            .elem_type()
1372            .equals_datatype(&map_type.clone().into_struct()),
1373        (DataType::Vector(size), ScalarRefImpl::Vector(v)) => v.dimension() == *size,
1374        (DataType::Struct(struct_type), ScalarRefImpl::Struct(v)) => {
1375            struct_ref_type_match(struct_type, v)
1376        }
1377
1378        _ => {
1379            macro_rules! matches {
1380                ($( { $data_type:ident, $variant_name:ident, $suffix_name:ident, $scalar:ty, $scalar_ref:ty, $array:ty, $builder:ty }),*) => {
1381                    match (data_type, scalar) {
1382                        $(
1383                            (DataType::$data_type { .. }, ScalarRefImpl::$variant_name(_)) => true,
1384                            (DataType::$data_type { .. }, _) => false, // keep exhaustive over DataType variants
1385                        )*
1386                    }
1387                }
1388            }
1389            for_all_variants! { matches }
1390        }
1391    }
1392}
1393
1394/// Returns whether the `datum` matches the `data_type`.
1395#[inline(always)]
1396pub fn datum_ref_type_match(data_type: &DataType, datum: DatumRef<'_>) -> bool {
1397    match datum {
1398        None => true,
1399        Some(scalar) => scalar_ref_type_match(data_type, scalar),
1400    }
1401}
1402
1403fn struct_ref_type_match(expected: &StructType, value: StructRef<'_>) -> bool {
1404    match value {
1405        StructRef::Indexed { arr, .. } => {
1406            // `StructRef::Indexed` comes with a `StructArray`, whose type can be compared directly.
1407            crate::array::Array::data_type(arr).equals_datatype(&DataType::Struct(expected.clone()))
1408        }
1409        StructRef::ValueRef { val } => {
1410            let fields = val.fields();
1411            if fields.len() != expected.len() {
1412                return false;
1413            }
1414            expected
1415                .types()
1416                .zip_eq_fast(fields.iter())
1417                .all(|(ty, datum)| datum_ref_type_match(ty, datum.to_datum_ref()))
1418        }
1419    }
1420}
1421
1422#[cfg(test)]
1423mod tests {
1424    use std::hash::{BuildHasher, Hasher};
1425
1426    use strum::IntoEnumIterator;
1427
1428    use super::*;
1429    use crate::util::hash_util::Crc32FastBuilder;
1430
1431    #[test]
1432    fn test_size() {
1433        use static_assertions::const_assert_eq;
1434
1435        use crate::array::*;
1436
1437        macro_rules! assert_item_size_eq {
1438            ($array:ty, $size:literal) => {
1439                const_assert_eq!(std::mem::size_of::<<$array as Array>::OwnedItem>(), $size);
1440            };
1441        }
1442
1443        assert_item_size_eq!(StructArray, 16); // Box<[Datum]>
1444        assert_item_size_eq!(ListArray, 8); // Box<ArrayImpl>
1445        assert_item_size_eq!(Utf8Array, 16); // Box<str>
1446        assert_item_size_eq!(IntervalArray, 16);
1447        assert_item_size_eq!(TimestampArray, 12);
1448
1449        // TODO: try to reduce the memory usage of `Decimal`, `ScalarImpl` and `Datum`.
1450        assert_item_size_eq!(DecimalArray, 20);
1451
1452        const_assert_eq!(std::mem::size_of::<ScalarImpl>(), 24);
1453        const_assert_eq!(std::mem::size_of::<ScalarRefImpl<'_>>(), 24);
1454        const_assert_eq!(std::mem::size_of::<Datum>(), 24);
1455        const_assert_eq!(std::mem::size_of::<StructType>(), 8);
1456        const_assert_eq!(std::mem::size_of::<DataType>(), 16);
1457    }
1458
1459    #[test]
1460    fn test_data_type_display() {
1461        let d: DataType =
1462            StructType::new(vec![("i", DataType::Int32), ("j", DataType::Varchar)]).into();
1463        assert_eq!(
1464            format!("{}", d),
1465            "struct<i integer, j character varying>".to_owned()
1466        );
1467    }
1468
1469    #[test]
1470    fn test_hash_implementation() {
1471        fn test(datum: Datum, data_type: DataType) {
1472            assert!(literal_type_match(&data_type, datum.as_ref()));
1473
1474            let mut builder = data_type.create_array_builder(6);
1475            for _ in 0..3 {
1476                builder.append_null();
1477                builder.append(&datum);
1478            }
1479            let array = builder.finish();
1480
1481            let hash_from_array = {
1482                let mut state = Crc32FastBuilder.build_hasher();
1483                array.hash_at(3, &mut state);
1484                state.finish()
1485            };
1486
1487            let hash_from_datum = {
1488                let mut state = Crc32FastBuilder.build_hasher();
1489                hash_datum(&datum, &mut state);
1490                state.finish()
1491            };
1492
1493            let hash_from_datum_ref = {
1494                let mut state = Crc32FastBuilder.build_hasher();
1495                hash_datum(datum.to_datum_ref(), &mut state);
1496                state.finish()
1497            };
1498
1499            assert_eq!(hash_from_array, hash_from_datum);
1500            assert_eq!(hash_from_datum, hash_from_datum_ref);
1501        }
1502
1503        for name in DataTypeName::iter() {
1504            let (scalar, data_type) = match name {
1505                DataTypeName::Boolean => (ScalarImpl::Bool(true), DataType::Boolean),
1506                DataTypeName::Int16 => (ScalarImpl::Int16(233), DataType::Int16),
1507                DataTypeName::Int32 => (ScalarImpl::Int32(233333), DataType::Int32),
1508                DataTypeName::Int64 => (ScalarImpl::Int64(233333333333), DataType::Int64),
1509                DataTypeName::Int256 => (
1510                    ScalarImpl::Int256(233333333333_i64.into()),
1511                    DataType::Int256,
1512                ),
1513                DataTypeName::Serial => (ScalarImpl::Serial(233333333333.into()), DataType::Serial),
1514                DataTypeName::Float32 => (ScalarImpl::Float32(23.33.into()), DataType::Float32),
1515                DataTypeName::Float64 => (
1516                    ScalarImpl::Float64(23.333333333333.into()),
1517                    DataType::Float64,
1518                ),
1519                DataTypeName::Decimal => (
1520                    ScalarImpl::Decimal("233.33".parse().unwrap()),
1521                    DataType::Decimal,
1522                ),
1523                DataTypeName::Date => (
1524                    ScalarImpl::Date(Date::from_ymd_uncheck(2333, 3, 3)),
1525                    DataType::Date,
1526                ),
1527                DataTypeName::Varchar => (ScalarImpl::Utf8("233".into()), DataType::Varchar),
1528                DataTypeName::Bytea => (
1529                    ScalarImpl::Bytea("\\x233".as_bytes().into()),
1530                    DataType::Bytea,
1531                ),
1532                DataTypeName::Time => (
1533                    ScalarImpl::Time(Time::from_hms_uncheck(2, 3, 3)),
1534                    DataType::Time,
1535                ),
1536                DataTypeName::Timestamp => (
1537                    ScalarImpl::Timestamp(Timestamp::from_timestamp_uncheck(23333333, 2333)),
1538                    DataType::Timestamp,
1539                ),
1540                DataTypeName::Timestamptz => (
1541                    ScalarImpl::Timestamptz(Timestamptz::from_micros(233333333).unwrap()),
1542                    DataType::Timestamptz,
1543                ),
1544                DataTypeName::Interval => (
1545                    ScalarImpl::Interval(Interval::from_month_day_usec(2, 3, 3333)),
1546                    DataType::Interval,
1547                ),
1548                DataTypeName::Jsonb => (ScalarImpl::Jsonb(JsonbVal::null()), DataType::Jsonb),
1549                DataTypeName::Variant => {
1550                    (ScalarImpl::Variant(VariantVal::null()), DataType::Variant)
1551                }
1552                DataTypeName::Struct => (
1553                    ScalarImpl::Struct(StructValue::new(vec![
1554                        ScalarImpl::Int64(233).into(),
1555                        ScalarImpl::Float64(23.33.into()).into(),
1556                    ])),
1557                    DataType::Struct(StructType::new(vec![
1558                        ("a", DataType::Int64),
1559                        ("b", DataType::Float64),
1560                    ])),
1561                ),
1562                DataTypeName::List => (
1563                    ScalarImpl::List(ListValue::from_iter([233i64, 2333])),
1564                    DataType::Int64.list(),
1565                ),
1566                DataTypeName::Vector => (
1567                    ScalarImpl::Vector(VectorVal::from_iter(
1568                        (0..VectorVal::TEST_VECTOR_DIMENSION)
1569                            .map(|i| ((i + 1) as f32).try_into().unwrap()),
1570                    )),
1571                    DataType::Vector(VectorVal::TEST_VECTOR_DIMENSION),
1572                ),
1573                DataTypeName::Map => {
1574                    // map is not hashable
1575                    continue;
1576                }
1577            };
1578
1579            test(Some(scalar), data_type.clone());
1580            test(None, data_type);
1581        }
1582    }
1583
1584    #[test]
1585    fn test_data_type_from_str() {
1586        assert_eq!(DataType::from_str("bool").unwrap(), DataType::Boolean);
1587        assert_eq!(DataType::from_str("boolean").unwrap(), DataType::Boolean);
1588        assert_eq!(DataType::from_str("BOOL").unwrap(), DataType::Boolean);
1589        assert_eq!(DataType::from_str("BOOLEAN").unwrap(), DataType::Boolean);
1590
1591        assert_eq!(DataType::from_str("int2").unwrap(), DataType::Int16);
1592        assert_eq!(DataType::from_str("smallint").unwrap(), DataType::Int16);
1593        assert_eq!(DataType::from_str("INT2").unwrap(), DataType::Int16);
1594        assert_eq!(DataType::from_str("SMALLINT").unwrap(), DataType::Int16);
1595
1596        assert_eq!(DataType::from_str("int4").unwrap(), DataType::Int32);
1597        assert_eq!(DataType::from_str("integer").unwrap(), DataType::Int32);
1598        assert_eq!(DataType::from_str("int4").unwrap(), DataType::Int32);
1599        assert_eq!(DataType::from_str("INT4").unwrap(), DataType::Int32);
1600        assert_eq!(DataType::from_str("INTEGER").unwrap(), DataType::Int32);
1601        assert_eq!(DataType::from_str("INT").unwrap(), DataType::Int32);
1602
1603        assert_eq!(DataType::from_str("int8").unwrap(), DataType::Int64);
1604        assert_eq!(DataType::from_str("bigint").unwrap(), DataType::Int64);
1605        assert_eq!(DataType::from_str("INT8").unwrap(), DataType::Int64);
1606        assert_eq!(DataType::from_str("BIGINT").unwrap(), DataType::Int64);
1607
1608        assert_eq!(DataType::from_str("rw_int256").unwrap(), DataType::Int256);
1609        assert_eq!(DataType::from_str("RW_INT256").unwrap(), DataType::Int256);
1610
1611        assert_eq!(DataType::from_str("float4").unwrap(), DataType::Float32);
1612        assert_eq!(DataType::from_str("real").unwrap(), DataType::Float32);
1613        assert_eq!(DataType::from_str("FLOAT4").unwrap(), DataType::Float32);
1614        assert_eq!(DataType::from_str("REAL").unwrap(), DataType::Float32);
1615
1616        assert_eq!(DataType::from_str("float8").unwrap(), DataType::Float64);
1617        assert_eq!(
1618            DataType::from_str("double precision").unwrap(),
1619            DataType::Float64
1620        );
1621        assert_eq!(DataType::from_str("FLOAT8").unwrap(), DataType::Float64);
1622        assert_eq!(
1623            DataType::from_str("DOUBLE PRECISION").unwrap(),
1624            DataType::Float64
1625        );
1626
1627        assert_eq!(DataType::from_str("decimal").unwrap(), DataType::Decimal);
1628        assert_eq!(DataType::from_str("DECIMAL").unwrap(), DataType::Decimal);
1629        assert_eq!(DataType::from_str("numeric").unwrap(), DataType::Decimal);
1630        assert_eq!(DataType::from_str("NUMERIC").unwrap(), DataType::Decimal);
1631
1632        assert_eq!(DataType::from_str("date").unwrap(), DataType::Date);
1633        assert_eq!(DataType::from_str("DATE").unwrap(), DataType::Date);
1634
1635        assert_eq!(DataType::from_str("varchar").unwrap(), DataType::Varchar);
1636        assert_eq!(DataType::from_str("VARCHAR").unwrap(), DataType::Varchar);
1637
1638        assert_eq!(DataType::from_str("time").unwrap(), DataType::Time);
1639        assert_eq!(
1640            DataType::from_str("time without time zone").unwrap(),
1641            DataType::Time
1642        );
1643        assert_eq!(DataType::from_str("TIME").unwrap(), DataType::Time);
1644        assert_eq!(
1645            DataType::from_str("TIME WITHOUT TIME ZONE").unwrap(),
1646            DataType::Time
1647        );
1648
1649        assert_eq!(
1650            DataType::from_str("timestamp").unwrap(),
1651            DataType::Timestamp
1652        );
1653        assert_eq!(
1654            DataType::from_str("timestamp without time zone").unwrap(),
1655            DataType::Timestamp
1656        );
1657        assert_eq!(
1658            DataType::from_str("TIMESTAMP").unwrap(),
1659            DataType::Timestamp
1660        );
1661        assert_eq!(
1662            DataType::from_str("TIMESTAMP WITHOUT TIME ZONE").unwrap(),
1663            DataType::Timestamp
1664        );
1665
1666        assert_eq!(
1667            DataType::from_str("timestamptz").unwrap(),
1668            DataType::Timestamptz
1669        );
1670        assert_eq!(
1671            DataType::from_str("timestamp with time zone").unwrap(),
1672            DataType::Timestamptz
1673        );
1674        assert_eq!(
1675            DataType::from_str("TIMESTAMPTZ").unwrap(),
1676            DataType::Timestamptz
1677        );
1678        assert_eq!(
1679            DataType::from_str("TIMESTAMP WITH TIME ZONE").unwrap(),
1680            DataType::Timestamptz
1681        );
1682
1683        assert_eq!(DataType::from_str("interval").unwrap(), DataType::Interval);
1684        assert_eq!(DataType::from_str("INTERVAL").unwrap(), DataType::Interval);
1685
1686        assert_eq!(
1687            DataType::from_str("int2[]").unwrap(),
1688            DataType::Int16.list()
1689        );
1690        assert_eq!(DataType::from_str("int[]").unwrap(), DataType::Int32.list());
1691        assert_eq!(
1692            DataType::from_str("int8[]").unwrap(),
1693            DataType::Int64.list()
1694        );
1695        assert_eq!(
1696            DataType::from_str("float4[]").unwrap(),
1697            DataType::Float32.list()
1698        );
1699        assert_eq!(
1700            DataType::from_str("float8[]").unwrap(),
1701            DataType::Float64.list()
1702        );
1703        assert_eq!(
1704            DataType::from_str("decimal[]").unwrap(),
1705            DataType::Decimal.list()
1706        );
1707        assert_eq!(
1708            DataType::from_str("varchar[]").unwrap(),
1709            DataType::Varchar.list()
1710        );
1711        assert_eq!(DataType::from_str("variant").unwrap(), DataType::Variant);
1712        assert_eq!(DataType::from_str("VARIANT").unwrap(), DataType::Variant);
1713        assert_eq!(
1714            DataType::from_str("variant[]").unwrap(),
1715            DataType::Variant.list()
1716        );
1717
1718        assert_eq!(DataType::from_str("date[]").unwrap(), DataType::Date.list());
1719        assert_eq!(DataType::from_str("time[]").unwrap(), DataType::Time.list());
1720        assert_eq!(
1721            DataType::from_str("timestamp[]").unwrap(),
1722            DataType::Timestamp.list()
1723        );
1724        assert_eq!(
1725            DataType::from_str("timestamptz[]").unwrap(),
1726            DataType::Timestamptz.list()
1727        );
1728        assert_eq!(
1729            DataType::from_str("interval[]").unwrap(),
1730            DataType::Interval.list()
1731        );
1732
1733        assert_eq!(
1734            DataType::from_str("record").unwrap(),
1735            DataType::Struct(StructType::unnamed(vec![]))
1736        );
1737        assert_eq!(
1738            DataType::from_str("struct<a int4, b varchar>").unwrap(),
1739            DataType::Struct(StructType::new(vec![
1740                ("a", DataType::Int32),
1741                ("b", DataType::Varchar)
1742            ]))
1743        );
1744    }
1745
1746    #[test]
1747    fn test_can_alter() {
1748        let cannots = [
1749            (DataType::Int32, None),
1750            (DataType::Int32.list(), None),
1751            (
1752                MapType::from_kv(DataType::Varchar, DataType::Int32.list()).into(),
1753                None,
1754            ),
1755            (
1756                StructType::new([("a", DataType::Int32)]).into(),
1757                Some(false),
1758            ),
1759            (
1760                MapType::from_kv(
1761                    DataType::Varchar,
1762                    StructType::new([("a", DataType::Int32)]).into(),
1763                )
1764                .into(),
1765                Some(false),
1766            ),
1767        ];
1768        for (cannot, why) in cannots {
1769            assert_eq!(cannot.can_alter(), why, "{cannot:?}");
1770        }
1771
1772        let cans = [
1773            StructType::new([("a", DataType::Int32), ("b", DataType::Int32.list())])
1774                .with_ids([ColumnId::new(1), ColumnId::new(2)])
1775                .into(),
1776            DataType::list(DataType::Struct(
1777                StructType::new([("a", DataType::Int32)]).with_ids([ColumnId::new(1)]),
1778            )),
1779            MapType::from_kv(
1780                DataType::Varchar,
1781                StructType::new([("a", DataType::Int32)])
1782                    .with_ids([ColumnId::new(1)])
1783                    .into(),
1784            )
1785            .into(),
1786        ];
1787        for can in cans {
1788            assert_eq!(can.can_alter(), Some(true), "{can:?}");
1789        }
1790    }
1791}
risingwave_common/types/mod.rs

risingwave_common/types/
mod.rs
