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::{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 can not 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            is_nullable: true,
459            ..Default::default()
460        };
461        match self {
462            DataType::Struct(t) => {
463                if !t.is_unnamed() {
464                    // To be consistent with `From<&PbDataType>`,
465                    // we only set field names when it's a named struct.
466                    pb.field_names = t.names().map(|s| s.into()).collect();
467                }
468                pb.field_type = t.types().map(|f| f.to_protobuf()).collect();
469                if let Some(ids) = t.ids() {
470                    pb.field_ids = ids.map(|id| id.get_id()).collect();
471                }
472            }
473            DataType::List(list) => {
474                pb.field_type = vec![list.elem().to_protobuf()];
475            }
476            DataType::Map(map) => {
477                // Same as List<Struct<K,V>>
478                pb.field_type = vec![map.clone().into_struct().to_protobuf()];
479            }
480            DataType::Vector(size) => {
481                pb.precision = *size as _;
482            }
483            DataType::Boolean
484            | DataType::Int16
485            | DataType::Int32
486            | DataType::Int64
487            | DataType::Float32
488            | DataType::Float64
489            | DataType::Decimal
490            | DataType::Date
491            | DataType::Varchar
492            | DataType::Time
493            | DataType::Timestamp
494            | DataType::Timestamptz
495            | DataType::Interval
496            | DataType::Bytea
497            | DataType::Jsonb
498            | DataType::Variant
499            | DataType::Serial
500            | DataType::Int256 => (),
501        }
502        pb
503    }
504
505    pub fn is_numeric(&self) -> bool {
506        matches!(
507            self,
508            DataType::Int16
509                | DataType::Int32
510                | DataType::Int64
511                | DataType::Serial
512                | DataType::Float32
513                | DataType::Float64
514                | DataType::Decimal
515        )
516    }
517
518    /// Returns whether the data type does not have inner fields.
519    pub fn is_simple(&self) -> bool {
520        matches!(self, data_types::simple!())
521    }
522
523    /// Returns whether the data type has inner fields.
524    pub fn is_composite(&self) -> bool {
525        matches!(self, data_types::composite!())
526    }
527
528    pub fn is_array(&self) -> bool {
529        matches!(self, DataType::List(_))
530    }
531
532    pub fn is_struct(&self) -> bool {
533        matches!(self, DataType::Struct(_))
534    }
535
536    pub fn is_map(&self) -> bool {
537        matches!(self, DataType::Map(_))
538    }
539
540    pub fn is_int(&self) -> bool {
541        matches!(self, DataType::Int16 | DataType::Int32 | DataType::Int64)
542    }
543
544    /// Returns the output type of time window function on a given input type.
545    pub fn window_of(input: &DataType) -> Option<DataType> {
546        match input {
547            DataType::Timestamptz => Some(DataType::Timestamptz),
548            DataType::Timestamp | DataType::Date => Some(DataType::Timestamp),
549            _ => None,
550        }
551    }
552
553    pub fn as_struct(&self) -> &StructType {
554        match self {
555            DataType::Struct(t) => t,
556            t => panic!("expect struct type, got {t}"),
557        }
558    }
559
560    pub fn into_struct(self) -> StructType {
561        match self {
562            DataType::Struct(t) => t,
563            t => panic!("expect struct type, got {t}"),
564        }
565    }
566
567    pub fn as_map(&self) -> &MapType {
568        match self {
569            DataType::Map(t) => t,
570            t => panic!("expect map type, got {t}"),
571        }
572    }
573
574    pub fn into_map(self) -> MapType {
575        match self {
576            DataType::Map(t) => t,
577            t => panic!("expect map type, got {t}"),
578        }
579    }
580
581    pub fn as_list(&self) -> &ListType {
582        match self {
583            DataType::List(t) => t,
584            t => panic!("expect list type, got {t}"),
585        }
586    }
587
588    pub fn into_list(self) -> ListType {
589        match self {
590            DataType::List(t) => t,
591            t => panic!("expect list type, got {t}"),
592        }
593    }
594
595    /// Returns the inner element's type if `self` is a list type.
596    /// Equivalent to `self.as_list().elem()`.
597    pub fn as_list_elem(&self) -> &DataType {
598        self.as_list().elem()
599    }
600
601    /// Returns the inner element's type if `self` is a list type.
602    /// Equivalent to `self.into_list().into_elem()`.
603    pub fn into_list_elem(self) -> DataType {
604        self.into_list().into_elem()
605    }
606
607    /// Return a new type that removes the outer list, and get the innermost element type.
608    ///
609    /// Use [`DataType::as_list_elem`] if you only want the element type of a list.
610    ///
611    /// ```
612    /// use risingwave_common::types::DataType::*;
613    /// assert_eq!(Int32.list().unnest_list(), &Int32);
614    /// assert_eq!(Int32.list().list().unnest_list(), &Int32);
615    /// ```
616    pub fn unnest_list(&self) -> &Self {
617        match self {
618            DataType::List(list) => list.elem().unnest_list(),
619            _ => self,
620        }
621    }
622
623    /// Return the number of dimensions of this array/list type. Return `0` when this type is not an
624    /// array/list.
625    pub fn array_ndims(&self) -> usize {
626        let mut d = 0;
627        let mut t = self;
628        while let Self::List(list) = t {
629            d += 1;
630            t = list.elem();
631        }
632        d
633    }
634
635    /// Compares the datatype with another, ignoring nested field names and ids.
636    pub fn equals_datatype(&self, other: &DataType) -> bool {
637        match (self, other) {
638            (Self::Struct(s1), Self::Struct(s2)) => s1.equals_datatype(s2),
639            (Self::List(d1), Self::List(d2)) => d1.elem().equals_datatype(d2.elem()),
640            (Self::Map(m1), Self::Map(m2)) => {
641                m1.key().equals_datatype(m2.key()) && m1.value().equals_datatype(m2.value())
642            }
643            _ => self == other,
644        }
645    }
646
647    /// Whether a column with this data type can be altered to a new data type. This determines
648    /// the encoding of the column data.
649    ///
650    /// Returns...
651    /// - `None`, if the data type is simple or does not contain a struct type.
652    /// - `Some(true)`, if the data type contains a struct type with field ids ([`StructType::has_ids`]).
653    /// - `Some(false)`, if the data type contains a struct type without field ids.
654    pub fn can_alter(&self) -> Option<bool> {
655        match self {
656            data_types::simple!() => None,
657            DataType::Struct(struct_type) => {
658                // As long as we meet a struct type, we can check its `ids` field to determine if
659                // it can be altered.
660                let struct_can_alter = struct_type.has_ids();
661                // In debug build, we assert that once a struct type does (or does not) have ids,
662                // all its composite fields should have the same property.
663                if cfg!(debug_assertions) {
664                    for field in struct_type.types() {
665                        if let Some(field_can_alter) = field.can_alter() {
666                            assert_eq!(struct_can_alter, field_can_alter);
667                        }
668                    }
669                }
670                Some(struct_can_alter)
671            }
672
673            DataType::List(list_type) => list_type.elem().can_alter(),
674            DataType::Map(map_type) => {
675                debug_assert!(
676                    map_type.key().is_simple(),
677                    "unexpected key type of map {map_type:?}"
678                );
679                map_type.value().can_alter()
680            }
681        }
682    }
683
684    /// Whether this type is `VARIANT` or contains a nested `VARIANT`.
685    pub fn contains_variant(&self) -> bool {
686        matches!(self, DataType::Variant)
687            || match self {
688                DataType::List(list_type) => list_type.elem().contains_variant(),
689                DataType::Struct(struct_type) => {
690                    struct_type.types().any(DataType::contains_variant)
691                }
692                DataType::Map(map_type) => {
693                    map_type.key().contains_variant() || map_type.value().contains_variant()
694                }
695                // Listed rather than `_`: a new composite type answering `false` here would slip
696                // past every VARIANT-as-key gate.
697                data_types::simple!() => false,
698            }
699    }
700}
701
702impl From<StructType> for DataType {
703    fn from(value: StructType) -> Self {
704        Self::Struct(value)
705    }
706}
707
708impl From<DataType> for PbDataType {
709    fn from(data_type: DataType) -> Self {
710        data_type.to_protobuf()
711    }
712}
713
714mod private {
715    use super::*;
716
717    // Note: put pub trait inside a private mod just makes the name private,
718    // The trait methods will still be publicly available...
719    // 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)
720
721    /// Common trait bounds of scalar and scalar reference types.
722    ///
723    /// NOTE(rc): `Hash` is not in the trait bound list, it's implemented as [`ScalarRef::hash_scalar`].
724    pub trait ScalarBounds<Impl> = Debug
725        + Send
726        + Sync
727        + Clone
728        + PartialEq
729        + Eq
730        // in default ascending order
731        + PartialOrd
732        + Ord
733        + TryFrom<Impl, Error = ArrayError>
734        // `ScalarImpl`/`ScalarRefImpl`
735        + Into<Impl>;
736}
737
738/// `Scalar` is a trait over all possible owned types in the evaluation
739/// framework.
740///
741/// `Scalar` is reciprocal to `ScalarRef`. Use `as_scalar_ref` to get a
742/// reference which has the same lifetime as `self`.
743pub trait Scalar: private::ScalarBounds<ScalarImpl> + 'static {
744    /// Type for reference of `Scalar`
745    type ScalarRefType<'a>: ScalarRef<'a, ScalarType = Self> + 'a
746    where
747        Self: 'a;
748
749    /// Get a reference to current scalar.
750    fn as_scalar_ref(&self) -> Self::ScalarRefType<'_>;
751
752    fn to_scalar_value(self) -> ScalarImpl {
753        self.into()
754    }
755}
756
757/// `ScalarRef` is a trait over all possible references in the evaluation
758/// framework.
759///
760/// `ScalarRef` is reciprocal to `Scalar`. Use `to_owned_scalar` to get an
761/// owned scalar.
762pub trait ScalarRef<'a>: private::ScalarBounds<ScalarRefImpl<'a>> + 'a + Copy {
763    /// `ScalarType` is the owned type of current `ScalarRef`.
764    type ScalarType: Scalar<ScalarRefType<'a> = Self>;
765
766    /// Convert `ScalarRef` to an owned scalar.
767    fn to_owned_scalar(&self) -> Self::ScalarType;
768
769    /// A wrapped hash function to get the hash value for this scaler.
770    fn hash_scalar<H: std::hash::Hasher>(&self, state: &mut H);
771}
772
773/// Define `ScalarImpl` and `ScalarRefImpl` with macro.
774macro_rules! scalar_impl_enum {
775    ($( { $data_type:ident, $variant_name:ident, $suffix_name:ident, $scalar:ty, $scalar_ref:ty, $array:ty, $builder:ty } ),*) => {
776        /// `ScalarImpl` embeds all possible scalars in the evaluation framework.
777        ///
778        /// Note: `ScalarImpl` doesn't contain all information of its `DataType`,
779        /// so sometimes they need to be used together.
780        /// e.g., for `Struct`, we don't have the field names in the value.
781        ///
782        /// See `for_all_variants` for the definition.
783        #[derive(Debug, Clone, PartialEq, Eq, EstimateSize)]
784        pub enum ScalarImpl {
785            $( $variant_name($scalar) ),*
786        }
787
788        /// `ScalarRefImpl` embeds all possible scalar references in the evaluation
789        /// framework.
790        ///
791        /// Note: `ScalarRefImpl` doesn't contain all information of its `DataType`,
792        /// so sometimes they need to be used together.
793        /// e.g., for `Struct`, we don't have the field names in the value.
794        ///
795        /// See `for_all_variants` for the definition.
796        #[derive(Debug, Copy, Clone, PartialEq, Eq)]
797        pub enum ScalarRefImpl<'scalar> {
798            $( $variant_name($scalar_ref) ),*
799        }
800    };
801}
802
803for_all_variants! { scalar_impl_enum }
804
805// We MUST NOT implement `Ord` for `ScalarImpl` because that will make `Datum` derive an incorrect
806// default `Ord`. To get a default-ordered `ScalarImpl`/`ScalarRefImpl`/`Datum`/`DatumRef`, you can
807// use `DefaultOrdered<T>`. If non-default order is needed, please refer to `sort_util`.
808impl !PartialOrd for ScalarImpl {}
809impl !PartialOrd for ScalarRefImpl<'_> {}
810
811pub type Datum = Option<ScalarImpl>;
812pub type DatumRef<'a> = Option<ScalarRefImpl<'a>>;
813
814/// This trait is to implement `to_owned_datum` for `Option<ScalarImpl>`
815pub trait ToOwnedDatum {
816    /// Convert the datum to an owned [`Datum`].
817    fn to_owned_datum(self) -> Datum;
818}
819
820impl ToOwnedDatum for &Datum {
821    #[inline(always)]
822    fn to_owned_datum(self) -> Datum {
823        self.clone()
824    }
825}
826
827impl<T: Into<ScalarImpl>> ToOwnedDatum for T {
828    #[inline(always)]
829    fn to_owned_datum(self) -> Datum {
830        Some(self.into())
831    }
832}
833
834impl<T: Into<ScalarImpl>> ToOwnedDatum for Option<T> {
835    #[inline(always)]
836    fn to_owned_datum(self) -> Datum {
837        self.map(Into::into)
838    }
839}
840
841impl<const N: usize> From<TypedId<N, u32>> for ScalarImpl {
842    fn from(value: TypedId<N, u32>) -> Self {
843        value.as_i32_id().into()
844    }
845}
846
847impl<const N: usize> From<TypedId<N, u64>> for ScalarImpl {
848    fn from(value: TypedId<N, u64>) -> Self {
849        value.as_i64_id().into()
850    }
851}
852
853#[auto_impl::auto_impl(&)]
854pub trait ToDatumRef: PartialEq + Eq + Debug + Send + Sync {
855    /// Convert the datum to [`DatumRef`].
856    fn to_datum_ref(&self) -> DatumRef<'_>;
857}
858
859impl ToDatumRef for Datum {
860    #[inline(always)]
861    fn to_datum_ref(&self) -> DatumRef<'_> {
862        self.as_ref().map(|d| d.as_scalar_ref_impl())
863    }
864}
865impl ToDatumRef for Option<&ScalarImpl> {
866    #[inline(always)]
867    fn to_datum_ref(&self) -> DatumRef<'_> {
868        self.map(|d| d.as_scalar_ref_impl())
869    }
870}
871impl ToDatumRef for DatumRef<'_> {
872    #[inline(always)]
873    fn to_datum_ref(&self) -> DatumRef<'_> {
874        *self
875    }
876}
877
878/// To make sure there is `as_scalar_ref` for all scalar ref types.
879/// See <https://github.com/risingwavelabs/risingwave/pull/9977/files#r1208972881>
880///
881/// This is used by the expr macro.
882pub trait SelfAsScalarRef {
883    fn as_scalar_ref(&self) -> Self;
884}
885macro_rules! impl_self_as_scalar_ref {
886    ($($t:ty),*) => {
887        $(
888            impl SelfAsScalarRef for $t {
889                fn as_scalar_ref(&self) -> Self {
890                    *self
891                }
892            }
893        )*
894    };
895}
896impl_self_as_scalar_ref! { &str, &[u8], Int256Ref<'_>, JsonbRef<'_>, VariantRef<'_>, ListRef<'_>, StructRef<'_>, ScalarRefImpl<'_>, MapRef<'_> }
897
898/// `for_all_native_types` includes all native variants of our scalar types.
899///
900/// Specifically, it doesn't support u8/u16/u32/u64.
901#[macro_export]
902macro_rules! for_all_native_types {
903    ($macro:ident) => {
904        $macro! {
905            { i16, Int16, read_i16 },
906            { i32, Int32, read_i32 },
907            { i64, Int64, read_i64 },
908            { Serial, Serial, read_i64 },
909            { $crate::types::F32, Float32, read_f32 },
910            { $crate::types::F64, Float64, read_f64 }
911        }
912    };
913}
914
915/// `impl_convert` implements several conversions for `Scalar`.
916/// * `Scalar <-> ScalarImpl` with `From` and `TryFrom` trait.
917/// * `ScalarRef <-> ScalarRefImpl` with `From` and `TryFrom` trait.
918/// * `&ScalarImpl -> &Scalar` with `impl.as_int16()`.
919/// * `ScalarImpl -> Scalar` with `impl.into_int16()`.
920macro_rules! impl_convert {
921    ($( { $data_type:ident, $variant_name:ident, $suffix_name:ident, $scalar:ty, $scalar_ref:ty, $array:ty, $builder:ty } ),*) => {
922        $(
923            impl From<$scalar> for ScalarImpl {
924                fn from(val: $scalar) -> Self {
925                    ScalarImpl::$variant_name(val)
926                }
927            }
928
929            impl TryFrom<ScalarImpl> for $scalar {
930                type Error = ArrayError;
931
932                fn try_from(val: ScalarImpl) -> ArrayResult<Self> {
933                    match val {
934                        ScalarImpl::$variant_name(scalar) => Ok(scalar),
935                        other_scalar => bail!("cannot convert ScalarImpl::{} to concrete type", other_scalar.get_ident()),
936                    }
937                }
938            }
939
940            impl <'scalar> From<$scalar_ref> for ScalarRefImpl<'scalar> {
941                fn from(val: $scalar_ref) -> Self {
942                    ScalarRefImpl::$variant_name(val)
943                }
944            }
945
946            impl <'scalar> TryFrom<ScalarRefImpl<'scalar>> for $scalar_ref {
947                type Error = ArrayError;
948
949                fn try_from(val: ScalarRefImpl<'scalar>) -> ArrayResult<Self> {
950                    match val {
951                        ScalarRefImpl::$variant_name(scalar_ref) => Ok(scalar_ref),
952                        other_scalar => bail!("cannot convert ScalarRefImpl::{} to concrete type {}", other_scalar.get_ident(), stringify!($variant_name)),
953                    }
954                }
955            }
956
957            paste! {
958                impl ScalarImpl {
959                    /// # Panics
960                    /// If the scalar is not of the expected type.
961                    pub fn [<as_ $suffix_name>](&self) -> &$scalar {
962                        match self {
963                            Self::$variant_name(scalar) => scalar,
964                            other_scalar => panic!("cannot convert ScalarImpl::{} to concrete type {}", other_scalar.get_ident(), stringify!($variant_name))
965                        }
966                    }
967
968                    /// # Panics
969                    /// If the scalar is not of the expected type.
970                    pub fn [<into_ $suffix_name>](self) -> $scalar {
971                        match self {
972                            Self::$variant_name(scalar) => scalar,
973                            other_scalar =>  panic!("cannot convert ScalarImpl::{} to concrete type {}", other_scalar.get_ident(), stringify!($variant_name))
974                        }
975                    }
976                }
977
978                impl <'scalar> ScalarRefImpl<'scalar> {
979                    /// # Panics
980                    /// If the scalar is not of the expected type.
981                    pub fn [<into_ $suffix_name>](self) -> $scalar_ref {
982                        match self {
983                            Self::$variant_name(inner) => inner,
984                            other_scalar => panic!("cannot convert ScalarRefImpl::{} to concrete type {}", other_scalar.get_ident(), stringify!($variant_name))
985                        }
986                    }
987                }
988            }
989        )*
990    };
991}
992
993for_all_variants! { impl_convert }
994
995// Implement `From<raw float>` for `ScalarImpl::Float` as a sugar.
996impl From<f32> for ScalarImpl {
997    fn from(f: f32) -> Self {
998        Self::Float32(f.into())
999    }
1000}
1001impl From<f64> for ScalarImpl {
1002    fn from(f: f64) -> Self {
1003        Self::Float64(f.into())
1004    }
1005}
1006
1007// Implement `From<string like>` for `ScalarImpl::Utf8` as a sugar.
1008impl From<String> for ScalarImpl {
1009    fn from(s: String) -> Self {
1010        Self::Utf8(s.into_boxed_str())
1011    }
1012}
1013impl From<&str> for ScalarImpl {
1014    fn from(s: &str) -> Self {
1015        Self::Utf8(s.into())
1016    }
1017}
1018impl From<&String> for ScalarImpl {
1019    fn from(s: &String) -> Self {
1020        Self::Utf8(s.as_str().into())
1021    }
1022}
1023impl TryFrom<ScalarImpl> for String {
1024    type Error = ArrayError;
1025
1026    fn try_from(val: ScalarImpl) -> ArrayResult<Self> {
1027        match val {
1028            ScalarImpl::Utf8(s) => Ok(s.into()),
1029            other_scalar => bail!(
1030                "cannot convert ScalarImpl::{} to concrete type",
1031                other_scalar.get_ident()
1032            ),
1033        }
1034    }
1035}
1036
1037impl From<char> for ScalarImpl {
1038    fn from(c: char) -> Self {
1039        Self::Utf8(c.to_string().into())
1040    }
1041}
1042
1043impl From<&[u8]> for ScalarImpl {
1044    fn from(s: &[u8]) -> Self {
1045        Self::Bytea(s.into())
1046    }
1047}
1048
1049impl From<JsonbRef<'_>> for ScalarImpl {
1050    fn from(jsonb: JsonbRef<'_>) -> Self {
1051        Self::Jsonb(jsonb.to_owned_scalar())
1052    }
1053}
1054
1055impl From<VariantRef<'_>> for ScalarImpl {
1056    fn from(variant: VariantRef<'_>) -> Self {
1057        Self::Variant(variant.to_owned_scalar())
1058    }
1059}
1060
1061impl<T: PrimitiveArrayItemType> From<Vec<T>> for ScalarImpl {
1062    fn from(v: Vec<T>) -> Self {
1063        Self::List(v.into_iter().collect())
1064    }
1065}
1066
1067impl<T: PrimitiveArrayItemType> From<Vec<Option<T>>> for ScalarImpl {
1068    fn from(v: Vec<Option<T>>) -> Self {
1069        Self::List(v.into_iter().collect())
1070    }
1071}
1072
1073impl From<Vec<String>> for ScalarImpl {
1074    fn from(v: Vec<String>) -> Self {
1075        Self::List(v.iter().map(|s| s.as_str()).collect())
1076    }
1077}
1078
1079impl From<Vec<u8>> for ScalarImpl {
1080    fn from(v: Vec<u8>) -> Self {
1081        Self::Bytea(v.into())
1082    }
1083}
1084
1085impl From<Bytes> for ScalarImpl {
1086    fn from(v: Bytes) -> Self {
1087        Self::Bytea(v.as_ref().into())
1088    }
1089}
1090
1091impl From<ListRef<'_>> for ScalarImpl {
1092    fn from(list: ListRef<'_>) -> Self {
1093        Self::List(list.to_owned_scalar())
1094    }
1095}
1096
1097impl ScalarImpl {
1098    /// Creates a scalar from pgwire "BINARY" format.
1099    ///
1100    /// The counterpart of [`to_binary::ToBinary`].
1101    pub fn from_binary(bytes: &Bytes, data_type: &DataType) -> Result<Self, BoxedError> {
1102        let res = match data_type {
1103            DataType::Varchar => Self::Utf8(String::from_sql(&Type::VARCHAR, bytes)?.into()),
1104            DataType::Bytea => Self::Bytea(Vec::<u8>::from_sql(&Type::BYTEA, bytes)?.into()),
1105            DataType::Boolean => Self::Bool(bool::from_sql(&Type::BOOL, bytes)?),
1106            DataType::Int16 => Self::Int16(i16::from_sql(&Type::INT2, bytes)?),
1107            DataType::Int32 => Self::Int32(i32::from_sql(&Type::INT4, bytes)?),
1108            DataType::Int64 => Self::Int64(i64::from_sql(&Type::INT8, bytes)?),
1109            DataType::Serial => Self::Serial(Serial::from(i64::from_sql(&Type::INT8, bytes)?)),
1110            DataType::Float32 => Self::Float32(f32::from_sql(&Type::FLOAT4, bytes)?.into()),
1111            DataType::Float64 => Self::Float64(f64::from_sql(&Type::FLOAT8, bytes)?.into()),
1112            DataType::Decimal => {
1113                Self::Decimal(rust_decimal::Decimal::from_sql(&Type::NUMERIC, bytes)?.into())
1114            }
1115            DataType::Date => Self::Date(chrono::NaiveDate::from_sql(&Type::DATE, bytes)?.into()),
1116            DataType::Time => Self::Time(chrono::NaiveTime::from_sql(&Type::TIME, bytes)?.into()),
1117            DataType::Timestamp => {
1118                Self::Timestamp(chrono::NaiveDateTime::from_sql(&Type::TIMESTAMP, bytes)?.into())
1119            }
1120            DataType::Timestamptz => Self::Timestamptz(
1121                chrono::DateTime::<chrono::Utc>::from_sql(&Type::TIMESTAMPTZ, bytes)?.into(),
1122            ),
1123            DataType::Interval => Self::Interval(Interval::from_sql(&Type::INTERVAL, bytes)?),
1124            DataType::Jsonb => Self::Jsonb(
1125                JsonbVal::value_deserialize(bytes)
1126                    .ok_or_else(|| "invalid value of Jsonb".to_owned())?,
1127            ),
1128            // pgwire binary parameters are untrusted and must be re-canonicalized.
1129            DataType::Variant => Self::Variant(VariantVal::from_serialized_untrusted(bytes)?),
1130            DataType::Int256 => Self::Int256(Int256::from_binary(bytes)?),
1131            DataType::Vector(_) | DataType::Struct(_) | DataType::List(_) | DataType::Map(_) => {
1132                return Err(format!("unsupported data type: {}", data_type).into());
1133            }
1134        };
1135        Ok(res)
1136    }
1137
1138    /// Creates a scalar from pgwire "TEXT" format.
1139    ///
1140    /// The counterpart of [`ToText`].
1141    pub fn from_text(s: &str, data_type: &DataType) -> Result<Self, BoxedError> {
1142        Ok(match data_type {
1143            DataType::Boolean => str_to_bool(s)?.into(),
1144            DataType::Int16 => i16::from_str(s)?.into(),
1145            DataType::Int32 => i32::from_str(s)?.into(),
1146            DataType::Int64 => i64::from_str(s)?.into(),
1147            DataType::Int256 => Int256::from_str(s)?.into(),
1148            DataType::Serial => Serial::from(i64::from_str(s)?).into(),
1149            DataType::Decimal => Decimal::from_str(s)?.into(),
1150            DataType::Float32 => F32::from_str(s)?.into(),
1151            DataType::Float64 => F64::from_str(s)?.into(),
1152            DataType::Varchar => s.into(),
1153            DataType::Date => Date::from_str(s)?.into(),
1154            DataType::Timestamp => Timestamp::from_str(s)?.into(),
1155            // We only handle the case with timezone here, and leave the implicit session timezone case
1156            // for later phase.
1157            DataType::Timestamptz => Timestamptz::from_str(s)?.into(),
1158            DataType::Time => Time::from_str(s)?.into(),
1159            DataType::Interval => Interval::from_str(s)?.into(),
1160            DataType::List(_) => ListValue::from_str(s, data_type)?.into(),
1161            DataType::Struct(st) => StructValue::from_str(s, st)?.into(),
1162            DataType::Jsonb => JsonbVal::from_str(s)?.into(),
1163            DataType::Variant => VariantVal::from_str(s)?.into(),
1164            DataType::Bytea => {
1165                let mut buf = Vec::new();
1166                str_to_bytea(s, &mut buf)?;
1167                buf.into()
1168            }
1169            DataType::Vector(size) => VectorVal::from_text(s, *size)?.into(),
1170            DataType::Map(_m) => return Err("map from text is not supported".into()),
1171        })
1172    }
1173
1174    pub fn from_text_for_test(s: &str, data_type: &DataType) -> Result<Self, BoxedError> {
1175        Ok(match data_type {
1176            DataType::Map(map_type) => MapValue::from_str_for_test(s, map_type)?.into(),
1177            _ => ScalarImpl::from_text(s, data_type)?,
1178        })
1179    }
1180}
1181
1182impl From<ScalarRefImpl<'_>> for ScalarImpl {
1183    fn from(scalar_ref: ScalarRefImpl<'_>) -> Self {
1184        scalar_ref.into_scalar_impl()
1185    }
1186}
1187
1188impl<'a> From<&'a ScalarImpl> for ScalarRefImpl<'a> {
1189    fn from(scalar: &'a ScalarImpl) -> Self {
1190        scalar.as_scalar_ref_impl()
1191    }
1192}
1193
1194impl ScalarImpl {
1195    /// Converts [`ScalarImpl`] to [`ScalarRefImpl`]
1196    pub fn as_scalar_ref_impl(&self) -> ScalarRefImpl<'_> {
1197        dispatch_scalar_variants!(self, inner, { inner.as_scalar_ref().into() })
1198    }
1199}
1200
1201impl ScalarRefImpl<'_> {
1202    /// Converts [`ScalarRefImpl`] to [`ScalarImpl`]
1203    pub fn into_scalar_impl(self) -> ScalarImpl {
1204        dispatch_scalar_ref_variants!(self, inner, { inner.to_owned_scalar().into() })
1205    }
1206}
1207
1208impl Hash for ScalarImpl {
1209    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
1210        dispatch_scalar_variants!(self, inner, { inner.as_scalar_ref().hash_scalar(state) })
1211    }
1212}
1213
1214impl Hash for ScalarRefImpl<'_> {
1215    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
1216        dispatch_scalar_ref_variants!(self, inner, { inner.hash_scalar(state) })
1217    }
1218}
1219
1220/// Feeds the raw scalar reference of `datum` to the given `state`, which should behave the same
1221/// as [`crate::array::Array::hash_at`], where NULL value will be carefully handled.
1222///
1223/// **FIXME**: the result of this function might be different from [`std::hash::Hash`] due to the
1224/// type alias of `DatumRef = Option<_>`, we should manually implement [`std::hash::Hash`] for
1225/// [`DatumRef`] in the future when it becomes a newtype. (#477)
1226#[inline(always)]
1227pub fn hash_datum(datum: impl ToDatumRef, state: &mut impl std::hash::Hasher) {
1228    match datum.to_datum_ref() {
1229        Some(scalar_ref) => scalar_ref.hash(state),
1230        None => NULL_VAL_FOR_HASH.hash(state),
1231    }
1232}
1233
1234impl ScalarRefImpl<'_> {
1235    pub fn binary_format(&self, data_type: &DataType) -> to_binary::Result<Bytes> {
1236        use self::to_binary::ToBinary;
1237        self.to_binary_with_type(data_type)
1238    }
1239
1240    pub fn text_format(&self, data_type: &DataType) -> String {
1241        self.to_text_with_type(data_type)
1242    }
1243
1244    /// Serialize the scalar into the `memcomparable` format.
1245    pub fn serialize(
1246        &self,
1247        ser: &mut memcomparable::Serializer<impl BufMut>,
1248    ) -> memcomparable::Result<()> {
1249        match self {
1250            Self::Int16(v) => v.serialize(ser)?,
1251            Self::Int32(v) => v.serialize(ser)?,
1252            Self::Int64(v) => v.serialize(ser)?,
1253            Self::Serial(v) => v.serialize(ser)?,
1254            Self::Float32(v) => v.serialize(ser)?,
1255            Self::Float64(v) => v.serialize(ser)?,
1256            Self::Utf8(v) => v.serialize(ser)?,
1257            Self::Bytea(v) => ser.serialize_bytes(v)?,
1258            Self::Bool(v) => v.serialize(ser)?,
1259            Self::Decimal(v) => ser.serialize_decimal((*v).into())?,
1260            Self::Interval(v) => v.serialize(ser)?,
1261            Self::Date(v) => v.0.num_days_from_ce().serialize(ser)?,
1262            Self::Timestamp(v) => {
1263                v.0.and_utc().timestamp().serialize(&mut *ser)?;
1264                v.0.and_utc().timestamp_subsec_nanos().serialize(ser)?;
1265            }
1266            Self::Timestamptz(v) => v.serialize(ser)?,
1267            Self::Time(v) => {
1268                v.0.num_seconds_from_midnight().serialize(&mut *ser)?;
1269                v.0.nanosecond().serialize(ser)?;
1270            }
1271            Self::Int256(v) => v.memcmp_serialize(ser)?,
1272            Self::Jsonb(v) => v.memcmp_serialize(ser)?,
1273            Self::Variant(v) => v.memcmp_serialize(ser)?,
1274            Self::Struct(v) => v.memcmp_serialize(ser)?,
1275            Self::List(v) => v.memcmp_serialize(ser)?,
1276            Self::Map(v) => v.memcmp_serialize(ser)?,
1277            Self::Vector(v) => v.memcmp_serialize(ser)?,
1278        };
1279        Ok(())
1280    }
1281}
1282
1283impl ScalarImpl {
1284    /// Serialize the scalar into the `memcomparable` format.
1285    pub fn serialize(
1286        &self,
1287        ser: &mut memcomparable::Serializer<impl BufMut>,
1288    ) -> memcomparable::Result<()> {
1289        self.as_scalar_ref_impl().serialize(ser)
1290    }
1291
1292    /// Deserialize the scalar from the `memcomparable` format.
1293    pub fn deserialize(
1294        ty: &DataType,
1295        de: &mut memcomparable::Deserializer<impl Buf>,
1296    ) -> memcomparable::Result<Self> {
1297        use DataType as Ty;
1298        Ok(match ty {
1299            Ty::Int16 => Self::Int16(i16::deserialize(de)?),
1300            Ty::Int32 => Self::Int32(i32::deserialize(de)?),
1301            Ty::Int64 => Self::Int64(i64::deserialize(de)?),
1302            Ty::Int256 => Self::Int256(Int256::memcmp_deserialize(de)?),
1303            Ty::Serial => Self::Serial(Serial::from(i64::deserialize(de)?)),
1304            Ty::Float32 => Self::Float32(f32::deserialize(de)?.into()),
1305            Ty::Float64 => Self::Float64(f64::deserialize(de)?.into()),
1306            Ty::Varchar => Self::Utf8(Box::<str>::deserialize(de)?),
1307            Ty::Bytea => Self::Bytea(serde_bytes::ByteBuf::deserialize(de)?.into_vec().into()),
1308            Ty::Boolean => Self::Bool(bool::deserialize(de)?),
1309            Ty::Decimal => Self::Decimal(de.deserialize_decimal()?.into()),
1310            Ty::Interval => Self::Interval(Interval::deserialize(de)?),
1311            Ty::Time => Self::Time({
1312                let secs = u32::deserialize(&mut *de)?;
1313                let nano = u32::deserialize(de)?;
1314                Time::with_secs_nano(secs, nano)
1315                    .map_err(|e| memcomparable::Error::Message(e.to_report_string()))?
1316            }),
1317            Ty::Timestamp => Self::Timestamp({
1318                let secs = i64::deserialize(&mut *de)?;
1319                let nsecs = u32::deserialize(de)?;
1320                Timestamp::with_secs_nsecs(secs, nsecs)
1321                    .map_err(|e| memcomparable::Error::Message(e.to_report_string()))?
1322            }),
1323            Ty::Timestamptz => Self::Timestamptz(Timestamptz::deserialize(de)?),
1324            Ty::Date => Self::Date({
1325                let days = i32::deserialize(de)?;
1326                Date::with_days_since_ce(days)
1327                    .map_err(|e| memcomparable::Error::Message(e.to_report_string()))?
1328            }),
1329            Ty::Jsonb => Self::Jsonb(JsonbVal::memcmp_deserialize(de)?),
1330            Ty::Variant => Self::Variant(VariantVal::memcmp_deserialize(de)?),
1331            Ty::Struct(t) => StructValue::memcmp_deserialize(t.types(), de)?.to_scalar_value(),
1332            Ty::List(t) => ListValue::memcmp_deserialize(t, de)?.to_scalar_value(),
1333            Ty::Map(t) => MapValue::memcmp_deserialize(t, de)?.to_scalar_value(),
1334            Ty::Vector(dimension) => {
1335                VectorVal::memcmp_deserialize(*dimension, de)?.to_scalar_value()
1336            }
1337        })
1338    }
1339
1340    pub fn as_integral(&self) -> i64 {
1341        match self {
1342            Self::Int16(v) => *v as i64,
1343            Self::Int32(v) => *v as i64,
1344            Self::Int64(v) => *v,
1345            _ => panic!(
1346                "Can't convert ScalarImpl::{} to a integral",
1347                self.get_ident()
1348            ),
1349        }
1350    }
1351}
1352
1353/// Returns whether the `literal` matches the `data_type`.
1354pub fn literal_type_match(data_type: &DataType, literal: Option<&ScalarImpl>) -> bool {
1355    match literal {
1356        None => true,
1357        Some(scalar) => scalar_ref_type_match(data_type, scalar.as_scalar_ref_impl()),
1358    }
1359}
1360
1361/// Returns whether the scalar ref matches the `data_type`.
1362///
1363/// This is a lightweight "shape check" intended for callers that need to avoid panics on
1364/// malformed input. For nested types, it checks element/field types recursively.
1365pub fn scalar_ref_type_match(data_type: &DataType, scalar: ScalarRefImpl<'_>) -> bool {
1366    match (data_type, scalar) {
1367        (DataType::List(list_type), ScalarRefImpl::List(v)) => {
1368            v.elem_type().equals_datatype(list_type.elem())
1369        }
1370        (DataType::Map(map_type), ScalarRefImpl::Map(v)) => v
1371            .inner()
1372            .elem_type()
1373            .equals_datatype(&map_type.clone().into_struct()),
1374        (DataType::Vector(size), ScalarRefImpl::Vector(v)) => v.dimension() == *size,
1375        (DataType::Struct(struct_type), ScalarRefImpl::Struct(v)) => {
1376            struct_ref_type_match(struct_type, v)
1377        }
1378
1379        _ => {
1380            macro_rules! matches {
1381                ($( { $data_type:ident, $variant_name:ident, $suffix_name:ident, $scalar:ty, $scalar_ref:ty, $array:ty, $builder:ty }),*) => {
1382                    match (data_type, scalar) {
1383                        $(
1384                            (DataType::$data_type { .. }, ScalarRefImpl::$variant_name(_)) => true,
1385                            (DataType::$data_type { .. }, _) => false, // keep exhaustive over DataType variants
1386                        )*
1387                    }
1388                }
1389            }
1390            for_all_variants! { matches }
1391        }
1392    }
1393}
1394
1395/// Returns whether the `datum` matches the `data_type`.
1396#[inline(always)]
1397pub fn datum_ref_type_match(data_type: &DataType, datum: DatumRef<'_>) -> bool {
1398    match datum {
1399        None => true,
1400        Some(scalar) => scalar_ref_type_match(data_type, scalar),
1401    }
1402}
1403
1404fn struct_ref_type_match(expected: &StructType, value: StructRef<'_>) -> bool {
1405    match value {
1406        StructRef::Indexed { arr, .. } => {
1407            // `StructRef::Indexed` comes with a `StructArray`, whose type can be compared directly.
1408            crate::array::Array::data_type(arr).equals_datatype(&DataType::Struct(expected.clone()))
1409        }
1410        StructRef::ValueRef { val } => {
1411            let fields = val.fields();
1412            if fields.len() != expected.len() {
1413                return false;
1414            }
1415            expected
1416                .types()
1417                .zip_eq_fast(fields.iter())
1418                .all(|(ty, datum)| datum_ref_type_match(ty, datum.to_datum_ref()))
1419        }
1420    }
1421}
1422
1423#[cfg(test)]
1424mod tests {
1425    use std::hash::{BuildHasher, Hasher};
1426
1427    use strum::IntoEnumIterator;
1428
1429    use super::*;
1430    use crate::util::hash_util::Crc32FastBuilder;
1431
1432    #[test]
1433    fn test_size() {
1434        use static_assertions::const_assert_eq;
1435
1436        use crate::array::*;
1437
1438        macro_rules! assert_item_size_eq {
1439            ($array:ty, $size:literal) => {
1440                const_assert_eq!(std::mem::size_of::<<$array as Array>::OwnedItem>(), $size);
1441            };
1442        }
1443
1444        assert_item_size_eq!(StructArray, 16); // Box<[Datum]>
1445        assert_item_size_eq!(ListArray, 8); // Box<ArrayImpl>
1446        assert_item_size_eq!(Utf8Array, 16); // Box<str>
1447        assert_item_size_eq!(IntervalArray, 16);
1448        assert_item_size_eq!(TimestampArray, 12);
1449
1450        // TODO: try to reduce the memory usage of `Decimal`, `ScalarImpl` and `Datum`.
1451        assert_item_size_eq!(DecimalArray, 20);
1452
1453        const_assert_eq!(std::mem::size_of::<ScalarImpl>(), 24);
1454        const_assert_eq!(std::mem::size_of::<ScalarRefImpl<'_>>(), 24);
1455        const_assert_eq!(std::mem::size_of::<Datum>(), 24);
1456        const_assert_eq!(std::mem::size_of::<StructType>(), 8);
1457        const_assert_eq!(std::mem::size_of::<DataType>(), 16);
1458    }
1459
1460    #[test]
1461    fn test_data_type_display() {
1462        let d: DataType =
1463            StructType::new(vec![("i", DataType::Int32), ("j", DataType::Varchar)]).into();
1464        assert_eq!(
1465            format!("{}", d),
1466            "struct<i integer, j character varying>".to_owned()
1467        );
1468    }
1469
1470    #[test]
1471    fn test_hash_implementation() {
1472        fn test(datum: Datum, data_type: DataType) {
1473            assert!(literal_type_match(&data_type, datum.as_ref()));
1474
1475            let mut builder = data_type.create_array_builder(6);
1476            for _ in 0..3 {
1477                builder.append_null();
1478                builder.append(&datum);
1479            }
1480            let array = builder.finish();
1481
1482            let hash_from_array = {
1483                let mut state = Crc32FastBuilder.build_hasher();
1484                array.hash_at(3, &mut state);
1485                state.finish()
1486            };
1487
1488            let hash_from_datum = {
1489                let mut state = Crc32FastBuilder.build_hasher();
1490                hash_datum(&datum, &mut state);
1491                state.finish()
1492            };
1493
1494            let hash_from_datum_ref = {
1495                let mut state = Crc32FastBuilder.build_hasher();
1496                hash_datum(datum.to_datum_ref(), &mut state);
1497                state.finish()
1498            };
1499
1500            assert_eq!(hash_from_array, hash_from_datum);
1501            assert_eq!(hash_from_datum, hash_from_datum_ref);
1502        }
1503
1504        for name in DataTypeName::iter() {
1505            let (scalar, data_type) = match name {
1506                DataTypeName::Boolean => (ScalarImpl::Bool(true), DataType::Boolean),
1507                DataTypeName::Int16 => (ScalarImpl::Int16(233), DataType::Int16),
1508                DataTypeName::Int32 => (ScalarImpl::Int32(233333), DataType::Int32),
1509                DataTypeName::Int64 => (ScalarImpl::Int64(233333333333), DataType::Int64),
1510                DataTypeName::Int256 => (
1511                    ScalarImpl::Int256(233333333333_i64.into()),
1512                    DataType::Int256,
1513                ),
1514                DataTypeName::Serial => (ScalarImpl::Serial(233333333333.into()), DataType::Serial),
1515                DataTypeName::Float32 => (ScalarImpl::Float32(23.33.into()), DataType::Float32),
1516                DataTypeName::Float64 => (
1517                    ScalarImpl::Float64(23.333333333333.into()),
1518                    DataType::Float64,
1519                ),
1520                DataTypeName::Decimal => (
1521                    ScalarImpl::Decimal("233.33".parse().unwrap()),
1522                    DataType::Decimal,
1523                ),
1524                DataTypeName::Date => (
1525                    ScalarImpl::Date(Date::from_ymd_uncheck(2333, 3, 3)),
1526                    DataType::Date,
1527                ),
1528                DataTypeName::Varchar => (ScalarImpl::Utf8("233".into()), DataType::Varchar),
1529                DataTypeName::Bytea => (
1530                    ScalarImpl::Bytea("\\x233".as_bytes().into()),
1531                    DataType::Bytea,
1532                ),
1533                DataTypeName::Time => (
1534                    ScalarImpl::Time(Time::from_hms_uncheck(2, 3, 3)),
1535                    DataType::Time,
1536                ),
1537                DataTypeName::Timestamp => (
1538                    ScalarImpl::Timestamp(Timestamp::from_timestamp_uncheck(23333333, 2333)),
1539                    DataType::Timestamp,
1540                ),
1541                DataTypeName::Timestamptz => (
1542                    ScalarImpl::Timestamptz(Timestamptz::from_micros(233333333).unwrap()),
1543                    DataType::Timestamptz,
1544                ),
1545                DataTypeName::Interval => (
1546                    ScalarImpl::Interval(Interval::from_month_day_usec(2, 3, 3333)),
1547                    DataType::Interval,
1548                ),
1549                DataTypeName::Jsonb => (ScalarImpl::Jsonb(JsonbVal::null()), DataType::Jsonb),
1550                DataTypeName::Variant => {
1551                    (ScalarImpl::Variant(VariantVal::null()), DataType::Variant)
1552                }
1553                DataTypeName::Struct => (
1554                    ScalarImpl::Struct(StructValue::new(vec![
1555                        ScalarImpl::Int64(233).into(),
1556                        ScalarImpl::Float64(23.33.into()).into(),
1557                    ])),
1558                    DataType::Struct(StructType::new(vec![
1559                        ("a", DataType::Int64),
1560                        ("b", DataType::Float64),
1561                    ])),
1562                ),
1563                DataTypeName::List => (
1564                    ScalarImpl::List(ListValue::from_iter([233i64, 2333])),
1565                    DataType::Int64.list(),
1566                ),
1567                DataTypeName::Vector => (
1568                    ScalarImpl::Vector(VectorVal::from_iter(
1569                        (0..VectorVal::TEST_VECTOR_DIMENSION)
1570                            .map(|i| ((i + 1) as f32).try_into().unwrap()),
1571                    )),
1572                    DataType::Vector(VectorVal::TEST_VECTOR_DIMENSION),
1573                ),
1574                DataTypeName::Map => {
1575                    // map is not hashable
1576                    continue;
1577                }
1578            };
1579
1580            test(Some(scalar), data_type.clone());
1581            test(None, data_type);
1582        }
1583    }
1584
1585    #[test]
1586    fn test_data_type_from_str() {
1587        assert_eq!(DataType::from_str("bool").unwrap(), DataType::Boolean);
1588        assert_eq!(DataType::from_str("boolean").unwrap(), DataType::Boolean);
1589        assert_eq!(DataType::from_str("BOOL").unwrap(), DataType::Boolean);
1590        assert_eq!(DataType::from_str("BOOLEAN").unwrap(), DataType::Boolean);
1591
1592        assert_eq!(DataType::from_str("int2").unwrap(), DataType::Int16);
1593        assert_eq!(DataType::from_str("smallint").unwrap(), DataType::Int16);
1594        assert_eq!(DataType::from_str("INT2").unwrap(), DataType::Int16);
1595        assert_eq!(DataType::from_str("SMALLINT").unwrap(), DataType::Int16);
1596
1597        assert_eq!(DataType::from_str("int4").unwrap(), DataType::Int32);
1598        assert_eq!(DataType::from_str("integer").unwrap(), DataType::Int32);
1599        assert_eq!(DataType::from_str("int4").unwrap(), DataType::Int32);
1600        assert_eq!(DataType::from_str("INT4").unwrap(), DataType::Int32);
1601        assert_eq!(DataType::from_str("INTEGER").unwrap(), DataType::Int32);
1602        assert_eq!(DataType::from_str("INT").unwrap(), DataType::Int32);
1603
1604        assert_eq!(DataType::from_str("int8").unwrap(), DataType::Int64);
1605        assert_eq!(DataType::from_str("bigint").unwrap(), DataType::Int64);
1606        assert_eq!(DataType::from_str("INT8").unwrap(), DataType::Int64);
1607        assert_eq!(DataType::from_str("BIGINT").unwrap(), DataType::Int64);
1608
1609        assert_eq!(DataType::from_str("rw_int256").unwrap(), DataType::Int256);
1610        assert_eq!(DataType::from_str("RW_INT256").unwrap(), DataType::Int256);
1611
1612        assert_eq!(DataType::from_str("float4").unwrap(), DataType::Float32);
1613        assert_eq!(DataType::from_str("real").unwrap(), DataType::Float32);
1614        assert_eq!(DataType::from_str("FLOAT4").unwrap(), DataType::Float32);
1615        assert_eq!(DataType::from_str("REAL").unwrap(), DataType::Float32);
1616
1617        assert_eq!(DataType::from_str("float8").unwrap(), DataType::Float64);
1618        assert_eq!(
1619            DataType::from_str("double precision").unwrap(),
1620            DataType::Float64
1621        );
1622        assert_eq!(DataType::from_str("FLOAT8").unwrap(), DataType::Float64);
1623        assert_eq!(
1624            DataType::from_str("DOUBLE PRECISION").unwrap(),
1625            DataType::Float64
1626        );
1627
1628        assert_eq!(DataType::from_str("decimal").unwrap(), DataType::Decimal);
1629        assert_eq!(DataType::from_str("DECIMAL").unwrap(), DataType::Decimal);
1630        assert_eq!(DataType::from_str("numeric").unwrap(), DataType::Decimal);
1631        assert_eq!(DataType::from_str("NUMERIC").unwrap(), DataType::Decimal);
1632
1633        assert_eq!(DataType::from_str("date").unwrap(), DataType::Date);
1634        assert_eq!(DataType::from_str("DATE").unwrap(), DataType::Date);
1635
1636        assert_eq!(DataType::from_str("varchar").unwrap(), DataType::Varchar);
1637        assert_eq!(DataType::from_str("VARCHAR").unwrap(), DataType::Varchar);
1638
1639        assert_eq!(DataType::from_str("time").unwrap(), DataType::Time);
1640        assert_eq!(
1641            DataType::from_str("time without time zone").unwrap(),
1642            DataType::Time
1643        );
1644        assert_eq!(DataType::from_str("TIME").unwrap(), DataType::Time);
1645        assert_eq!(
1646            DataType::from_str("TIME WITHOUT TIME ZONE").unwrap(),
1647            DataType::Time
1648        );
1649
1650        assert_eq!(
1651            DataType::from_str("timestamp").unwrap(),
1652            DataType::Timestamp
1653        );
1654        assert_eq!(
1655            DataType::from_str("timestamp without time zone").unwrap(),
1656            DataType::Timestamp
1657        );
1658        assert_eq!(
1659            DataType::from_str("TIMESTAMP").unwrap(),
1660            DataType::Timestamp
1661        );
1662        assert_eq!(
1663            DataType::from_str("TIMESTAMP WITHOUT TIME ZONE").unwrap(),
1664            DataType::Timestamp
1665        );
1666
1667        assert_eq!(
1668            DataType::from_str("timestamptz").unwrap(),
1669            DataType::Timestamptz
1670        );
1671        assert_eq!(
1672            DataType::from_str("timestamp with time zone").unwrap(),
1673            DataType::Timestamptz
1674        );
1675        assert_eq!(
1676            DataType::from_str("TIMESTAMPTZ").unwrap(),
1677            DataType::Timestamptz
1678        );
1679        assert_eq!(
1680            DataType::from_str("TIMESTAMP WITH TIME ZONE").unwrap(),
1681            DataType::Timestamptz
1682        );
1683
1684        assert_eq!(DataType::from_str("interval").unwrap(), DataType::Interval);
1685        assert_eq!(DataType::from_str("INTERVAL").unwrap(), DataType::Interval);
1686
1687        assert_eq!(
1688            DataType::from_str("int2[]").unwrap(),
1689            DataType::Int16.list()
1690        );
1691        assert_eq!(DataType::from_str("int[]").unwrap(), DataType::Int32.list());
1692        assert_eq!(
1693            DataType::from_str("int8[]").unwrap(),
1694            DataType::Int64.list()
1695        );
1696        assert_eq!(
1697            DataType::from_str("float4[]").unwrap(),
1698            DataType::Float32.list()
1699        );
1700        assert_eq!(
1701            DataType::from_str("float8[]").unwrap(),
1702            DataType::Float64.list()
1703        );
1704        assert_eq!(
1705            DataType::from_str("decimal[]").unwrap(),
1706            DataType::Decimal.list()
1707        );
1708        assert_eq!(
1709            DataType::from_str("varchar[]").unwrap(),
1710            DataType::Varchar.list()
1711        );
1712        assert_eq!(DataType::from_str("variant").unwrap(), DataType::Variant);
1713        assert_eq!(DataType::from_str("VARIANT").unwrap(), DataType::Variant);
1714        assert_eq!(
1715            DataType::from_str("variant[]").unwrap(),
1716            DataType::Variant.list()
1717        );
1718
1719        assert_eq!(DataType::from_str("date[]").unwrap(), DataType::Date.list());
1720        assert_eq!(DataType::from_str("time[]").unwrap(), DataType::Time.list());
1721        assert_eq!(
1722            DataType::from_str("timestamp[]").unwrap(),
1723            DataType::Timestamp.list()
1724        );
1725        assert_eq!(
1726            DataType::from_str("timestamptz[]").unwrap(),
1727            DataType::Timestamptz.list()
1728        );
1729        assert_eq!(
1730            DataType::from_str("interval[]").unwrap(),
1731            DataType::Interval.list()
1732        );
1733
1734        assert_eq!(
1735            DataType::from_str("record").unwrap(),
1736            DataType::Struct(StructType::unnamed(vec![]))
1737        );
1738        assert_eq!(
1739            DataType::from_str("struct<a int4, b varchar>").unwrap(),
1740            DataType::Struct(StructType::new(vec![
1741                ("a", DataType::Int32),
1742                ("b", DataType::Varchar)
1743            ]))
1744        );
1745    }
1746
1747    #[test]
1748    fn test_can_alter() {
1749        let cannots = [
1750            (DataType::Int32, None),
1751            (DataType::Int32.list(), None),
1752            (
1753                MapType::from_kv(DataType::Varchar, DataType::Int32.list()).into(),
1754                None,
1755            ),
1756            (
1757                StructType::new([("a", DataType::Int32)]).into(),
1758                Some(false),
1759            ),
1760            (
1761                MapType::from_kv(
1762                    DataType::Varchar,
1763                    StructType::new([("a", DataType::Int32)]).into(),
1764                )
1765                .into(),
1766                Some(false),
1767            ),
1768        ];
1769        for (cannot, why) in cannots {
1770            assert_eq!(cannot.can_alter(), why, "{cannot:?}");
1771        }
1772
1773        let cans = [
1774            StructType::new([("a", DataType::Int32), ("b", DataType::Int32.list())])
1775                .with_ids([ColumnId::new(1), ColumnId::new(2)])
1776                .into(),
1777            DataType::list(DataType::Struct(
1778                StructType::new([("a", DataType::Int32)]).with_ids([ColumnId::new(1)]),
1779            )),
1780            MapType::from_kv(
1781                DataType::Varchar,
1782                StructType::new([("a", DataType::Int32)])
1783                    .with_ids([ColumnId::new(1)])
1784                    .into(),
1785            )
1786            .into(),
1787        ];
1788        for can in cans {
1789            assert_eq!(can.can_alter(), Some(true), "{can:?}");
1790        }
1791    }
1792}
risingwave_common/types/mod.rs

risingwave_common/types/
mod.rs
