mz_expr/scalar/func/impls/

string.rs

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

use std::borrow::Cow;
use std::fmt;
use std::sync::LazyLock;

use chrono::{DateTime, NaiveDateTime, NaiveTime, Utc};
use mz_expr_derive::sqlfunc;
use mz_lowertest::MzReflect;
use mz_ore::cast::CastFrom;
use mz_ore::result::ResultExt;
use mz_ore::str::StrExt;
use mz_repr::adt::char::{Char, format_str_trim};
use mz_repr::adt::date::Date;
use mz_repr::adt::interval::Interval;
use mz_repr::adt::jsonb::Jsonb;
use mz_repr::adt::numeric::{self, Numeric, NumericMaxScale};
use mz_repr::adt::pg_legacy_name::PgLegacyName;
use mz_repr::adt::regex::Regex;
use mz_repr::adt::system::{Oid, PgLegacyChar};
use mz_repr::adt::timestamp::{CheckedTimestamp, TimestampPrecision};
use mz_repr::adt::varchar::{VarChar, VarCharMaxLength};
use mz_repr::{Datum, RowArena, SqlColumnType, SqlScalarType, strconv};
use serde::{Deserialize, Serialize};
use uuid::Uuid;

use crate::func::{binary, regexp_match_static};
use crate::scalar::func::{
    EagerUnaryFunc, LazyUnaryFunc, array_create_scalar, regexp_split_to_array_re,
};
use crate::{EvalError, MirScalarExpr, UnaryFunc, like_pattern};

#[sqlfunc(
    sqlname = "text_to_boolean",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastBoolToString)
)]
fn cast_string_to_bool<'a>(a: &'a str) -> Result<bool, EvalError> {
    strconv::parse_bool(a).err_into()
}

#[sqlfunc(
    sqlname = "text_to_\"char\"",
    preserves_uniqueness = true,
    inverse = to_unary!(super::CastPgLegacyCharToString)
)]
fn cast_string_to_pg_legacy_char<'a>(a: &'a str) -> PgLegacyChar {
    PgLegacyChar(a.as_bytes().get(0).copied().unwrap_or(0))
}

#[sqlfunc(sqlname = "text_to_name", preserves_uniqueness = true)]
fn cast_string_to_pg_legacy_name<'a>(a: &'a str) -> PgLegacyName<String> {
    PgLegacyName(strconv::parse_pg_legacy_name(a))
}

#[sqlfunc(
    sqlname = "text_to_bytea",
    preserves_uniqueness = true,
    inverse = to_unary!(super::CastBytesToString)
)]
fn cast_string_to_bytes<'a>(a: &'a str) -> Result<Vec<u8>, EvalError> {
    strconv::parse_bytes(a).err_into()
}

#[sqlfunc(
    sqlname = "text_to_smallint",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastInt16ToString)
)]
fn cast_string_to_int16<'a>(a: &'a str) -> Result<i16, EvalError> {
    strconv::parse_int16(a).err_into()
}

#[sqlfunc(
    sqlname = "text_to_integer",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastInt32ToString)
)]
fn cast_string_to_int32<'a>(a: &'a str) -> Result<i32, EvalError> {
    strconv::parse_int32(a).err_into()
}

#[sqlfunc(
    sqlname = "text_to_bigint",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastInt64ToString)
)]
fn cast_string_to_int64<'a>(a: &'a str) -> Result<i64, EvalError> {
    strconv::parse_int64(a).err_into()
}

#[sqlfunc(
    sqlname = "text_to_real",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastFloat32ToString)
)]
fn cast_string_to_float32<'a>(a: &'a str) -> Result<f32, EvalError> {
    strconv::parse_float32(a).err_into()
}

#[sqlfunc(
    sqlname = "text_to_double",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastFloat64ToString)
)]
fn cast_string_to_float64<'a>(a: &'a str) -> Result<f64, EvalError> {
    strconv::parse_float64(a).err_into()
}

#[sqlfunc(
    sqlname = "text_to_oid",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastOidToString)
)]
fn cast_string_to_oid<'a>(a: &'a str) -> Result<Oid, EvalError> {
    Ok(Oid(strconv::parse_oid(a)?))
}

#[sqlfunc(
    sqlname = "text_to_uint2",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastUint16ToString)
)]
fn cast_string_to_uint16(a: &str) -> Result<u16, EvalError> {
    strconv::parse_uint16(a).err_into()
}

#[sqlfunc(
    sqlname = "text_to_uint4",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastUint32ToString)
)]
fn cast_string_to_uint32(a: &str) -> Result<u32, EvalError> {
    strconv::parse_uint32(a).err_into()
}

#[sqlfunc(
    sqlname = "text_to_uint8",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastUint64ToString)
)]
fn cast_string_to_uint64(a: &str) -> Result<u64, EvalError> {
    strconv::parse_uint64(a).err_into()
}

#[sqlfunc(sqlname = "reverse")]
fn reverse<'a>(a: &'a str) -> String {
    a.chars().rev().collect()
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct CastStringToNumeric(pub Option<NumericMaxScale>);

impl EagerUnaryFunc for CastStringToNumeric {
    type Input<'a> = &'a str;
    type Output<'a> = Result<Numeric, EvalError>;

    fn call<'a>(&self, a: Self::Input<'a>) -> Self::Output<'a> {
        let mut d = strconv::parse_numeric(a)?;
        if let Some(scale) = self.0 {
            if numeric::rescale(&mut d.0, scale.into_u8()).is_err() {
                return Err(EvalError::NumericFieldOverflow);
            }
        }
        Ok(d.into_inner())
    }

    fn output_type(&self, input: SqlColumnType) -> SqlColumnType {
        SqlScalarType::Numeric { max_scale: self.0 }.nullable(input.nullable)
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        to_unary!(super::CastNumericToString)
    }
}

impl fmt::Display for CastStringToNumeric {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("text_to_numeric")
    }
}

#[sqlfunc(
    sqlname = "text_to_date",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastDateToString)
)]
fn cast_string_to_date<'a>(a: &'a str) -> Result<Date, EvalError> {
    strconv::parse_date(a).err_into()
}

#[sqlfunc(
    sqlname = "text_to_time",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastTimeToString)
)]
fn cast_string_to_time<'a>(a: &'a str) -> Result<NaiveTime, EvalError> {
    strconv::parse_time(a).err_into()
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct CastStringToTimestamp(pub Option<TimestampPrecision>);

impl EagerUnaryFunc for CastStringToTimestamp {
    type Input<'a> = &'a str;
    type Output<'a> = Result<CheckedTimestamp<NaiveDateTime>, EvalError>;

    fn call<'a>(&self, a: Self::Input<'a>) -> Self::Output<'a> {
        let out = strconv::parse_timestamp(a)?;
        let updated = out.round_to_precision(self.0)?;
        Ok(updated)
    }

    fn output_type(&self, input: SqlColumnType) -> SqlColumnType {
        SqlScalarType::Timestamp { precision: self.0 }.nullable(input.nullable)
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        to_unary!(super::CastTimestampToString)
    }
}

impl fmt::Display for CastStringToTimestamp {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("text_to_timestamp")
    }
}

#[sqlfunc(sqlname = "try_parse_monotonic_iso8601_timestamp")]
// TODO: Pretty sure this preserves uniqueness, but not 100%.
//
// Ironically, even though this has "monotonic" in the name, it's not quite
// eligible for `#[is_monotone = true]` because any input could also be
// mapped to null. So, handle it via SpecialUnary in the interpreter.
fn try_parse_monotonic_iso8601_timestamp<'a>(
    a: &'a str,
) -> Option<CheckedTimestamp<NaiveDateTime>> {
    let ts = mz_persist_types::timestamp::try_parse_monotonic_iso8601_timestamp(a)?;
    let ts = CheckedTimestamp::from_timestamplike(ts)
        .expect("monotonic_iso8601 range is a subset of CheckedTimestamp domain");
    Some(ts)
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct CastStringToTimestampTz(pub Option<TimestampPrecision>);

impl EagerUnaryFunc for CastStringToTimestampTz {
    type Input<'a> = &'a str;
    type Output<'a> = Result<CheckedTimestamp<DateTime<Utc>>, EvalError>;

    fn call<'a>(&self, a: Self::Input<'a>) -> Self::Output<'a> {
        let out = strconv::parse_timestamptz(a)?;
        let updated = out.round_to_precision(self.0)?;
        Ok(updated)
    }

    fn output_type(&self, input: SqlColumnType) -> SqlColumnType {
        SqlScalarType::TimestampTz { precision: self.0 }.nullable(input.nullable)
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        to_unary!(super::CastTimestampTzToString)
    }
}

impl fmt::Display for CastStringToTimestampTz {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("text_to_timestamp_with_time_zone")
    }
}

#[sqlfunc(
    sqlname = "text_to_interval",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastIntervalToString)
)]
fn cast_string_to_interval<'a>(a: &'a str) -> Result<Interval, EvalError> {
    strconv::parse_interval(a).err_into()
}

#[sqlfunc(
    sqlname = "text_to_uuid",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastUuidToString)
)]
fn cast_string_to_uuid<'a>(a: &'a str) -> Result<Uuid, EvalError> {
    strconv::parse_uuid(a).err_into()
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct CastStringToArray {
    // Target array's type.
    pub return_ty: SqlScalarType,
    // The expression to cast the discovered array elements to the array's
    // element type.
    pub cast_expr: Box<MirScalarExpr>,
}

impl LazyUnaryFunc for CastStringToArray {
    fn eval<'a>(
        &'a self,
        datums: &[Datum<'a>],
        temp_storage: &'a RowArena,
        a: &'a MirScalarExpr,
    ) -> Result<Datum<'a>, EvalError> {
        let a = a.eval(datums, temp_storage)?;
        if a.is_null() {
            return Ok(Datum::Null);
        }
        let (datums, dims) = strconv::parse_array(
            a.unwrap_str(),
            || Datum::Null,
            |elem_text| {
                let elem_text = match elem_text {
                    Cow::Owned(s) => temp_storage.push_string(s),
                    Cow::Borrowed(s) => s,
                };
                self.cast_expr
                    .eval(&[Datum::String(elem_text)], temp_storage)
            },
        )?;

        Ok(temp_storage.try_make_datum(|packer| packer.try_push_array(&dims, datums))?)
    }

    /// The output SqlColumnType of this function
    fn output_type(&self, input_type: SqlColumnType) -> SqlColumnType {
        self.return_ty.clone().nullable(input_type.nullable)
    }

    /// Whether this function will produce NULL on NULL input
    fn propagates_nulls(&self) -> bool {
        true
    }

    /// Whether this function will produce NULL on non-NULL input
    fn introduces_nulls(&self) -> bool {
        false
    }

    /// Whether this function preserves uniqueness
    fn preserves_uniqueness(&self) -> bool {
        false
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        to_unary!(super::CastArrayToString {
            ty: self.return_ty.clone(),
        })
    }

    fn is_monotone(&self) -> bool {
        false
    }
}

impl fmt::Display for CastStringToArray {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("strtoarray")
    }
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct CastStringToList {
    // Target list's type
    pub return_ty: SqlScalarType,
    // The expression to cast the discovered list elements to the list's
    // element type.
    pub cast_expr: Box<MirScalarExpr>,
}

impl LazyUnaryFunc for CastStringToList {
    fn eval<'a>(
        &'a self,
        datums: &[Datum<'a>],
        temp_storage: &'a RowArena,
        a: &'a MirScalarExpr,
    ) -> Result<Datum<'a>, EvalError> {
        let a = a.eval(datums, temp_storage)?;
        if a.is_null() {
            return Ok(Datum::Null);
        }
        let parsed_datums = strconv::parse_list(
            a.unwrap_str(),
            matches!(
                self.return_ty.unwrap_list_element_type(),
                SqlScalarType::List { .. }
            ),
            || Datum::Null,
            |elem_text| {
                let elem_text = match elem_text {
                    Cow::Owned(s) => temp_storage.push_string(s),
                    Cow::Borrowed(s) => s,
                };
                self.cast_expr
                    .eval(&[Datum::String(elem_text)], temp_storage)
            },
        )?;

        Ok(temp_storage.make_datum(|packer| packer.push_list(parsed_datums)))
    }

    /// The output SqlColumnType of this function
    fn output_type(&self, input_type: SqlColumnType) -> SqlColumnType {
        self.return_ty
            .without_modifiers()
            .nullable(input_type.nullable)
    }

    /// Whether this function will produce NULL on NULL input
    fn propagates_nulls(&self) -> bool {
        true
    }

    /// Whether this function will produce NULL on non-NULL input
    fn introduces_nulls(&self) -> bool {
        false
    }

    /// Whether this function preserves uniqueness
    fn preserves_uniqueness(&self) -> bool {
        false
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        to_unary!(super::CastListToString {
            ty: self.return_ty.clone(),
        })
    }

    fn is_monotone(&self) -> bool {
        false
    }
}

impl fmt::Display for CastStringToList {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("strtolist")
    }
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct CastStringToMap {
    // Target map's value type
    pub return_ty: SqlScalarType,
    // The expression used to cast the discovered values to the map's value
    // type.
    pub cast_expr: Box<MirScalarExpr>,
}

impl LazyUnaryFunc for CastStringToMap {
    fn eval<'a>(
        &'a self,
        datums: &[Datum<'a>],
        temp_storage: &'a RowArena,
        a: &'a MirScalarExpr,
    ) -> Result<Datum<'a>, EvalError> {
        let a = a.eval(datums, temp_storage)?;
        if a.is_null() {
            return Ok(Datum::Null);
        }
        let parsed_map = strconv::parse_map(
            a.unwrap_str(),
            matches!(
                self.return_ty.unwrap_map_value_type(),
                SqlScalarType::Map { .. }
            ),
            |value_text| -> Result<Datum, EvalError> {
                let value_text = match value_text {
                    Some(Cow::Owned(s)) => Datum::String(temp_storage.push_string(s)),
                    Some(Cow::Borrowed(s)) => Datum::String(s),
                    None => Datum::Null,
                };
                self.cast_expr.eval(&[value_text], temp_storage)
            },
        )?;
        let mut pairs: Vec<(String, Datum)> = parsed_map.into_iter().map(|(k, v)| (k, v)).collect();
        pairs.sort_by(|(k1, _v1), (k2, _v2)| k1.cmp(k2));
        pairs.dedup_by(|(k1, _v1), (k2, _v2)| k1 == k2);
        Ok(temp_storage.make_datum(|packer| {
            packer.push_dict_with(|packer| {
                for (k, v) in pairs {
                    packer.push(Datum::String(&k));
                    packer.push(v);
                }
            })
        }))
    }

    /// The output SqlColumnType of this function
    fn output_type(&self, input_type: SqlColumnType) -> SqlColumnType {
        self.return_ty.clone().nullable(input_type.nullable)
    }

    /// Whether this function will produce NULL on NULL input
    fn propagates_nulls(&self) -> bool {
        true
    }

    /// Whether this function will produce NULL on non-NULL input
    fn introduces_nulls(&self) -> bool {
        false
    }

    /// Whether this function preserves uniqueness
    fn preserves_uniqueness(&self) -> bool {
        false
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        to_unary!(super::CastMapToString {
            ty: self.return_ty.clone(),
        })
    }

    fn is_monotone(&self) -> bool {
        false
    }
}

impl fmt::Display for CastStringToMap {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("strtomap")
    }
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct CastStringToChar {
    pub length: Option<mz_repr::adt::char::CharLength>,
    pub fail_on_len: bool,
}

impl EagerUnaryFunc for CastStringToChar {
    type Input<'a> = &'a str;
    type Output<'a> = Result<Char<String>, EvalError>;

    fn call<'a>(&self, a: Self::Input<'a>) -> Self::Output<'a> {
        let s = format_str_trim(a, self.length, self.fail_on_len).map_err(|_| {
            assert!(self.fail_on_len);
            EvalError::StringValueTooLong {
                target_type: "character".into(),
                length: usize::cast_from(self.length.unwrap().into_u32()),
            }
        })?;

        Ok(Char(s))
    }

    fn output_type(&self, input: SqlColumnType) -> SqlColumnType {
        SqlScalarType::Char {
            length: self.length,
        }
        .nullable(input.nullable)
    }

    fn could_error(&self) -> bool {
        self.fail_on_len && self.length.is_some()
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        to_unary!(super::CastCharToString)
    }
}

impl fmt::Display for CastStringToChar {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self.length {
            Some(length) => {
                write!(
                    f,
                    "text_to_char[len={}, fail_on_len={}]",
                    length.into_u32(),
                    self.fail_on_len
                )
            }
            None => f.write_str("text_to_char[len=unbounded]"),
        }
    }
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct CastStringToRange {
    // Target range's type
    pub return_ty: SqlScalarType,
    // The expression to cast the discovered range elements to the range's
    // element type.
    pub cast_expr: Box<MirScalarExpr>,
}

impl LazyUnaryFunc for CastStringToRange {
    fn eval<'a>(
        &'a self,
        datums: &[Datum<'a>],
        temp_storage: &'a RowArena,
        a: &'a MirScalarExpr,
    ) -> Result<Datum<'a>, EvalError> {
        let a = a.eval(datums, temp_storage)?;
        if a.is_null() {
            return Ok(Datum::Null);
        }
        let mut range = strconv::parse_range(a.unwrap_str(), |elem_text| {
            let elem_text = match elem_text {
                Cow::Owned(s) => temp_storage.push_string(s),
                Cow::Borrowed(s) => s,
            };
            self.cast_expr
                .eval(&[Datum::String(elem_text)], temp_storage)
        })?;

        range.canonicalize()?;

        Ok(temp_storage.make_datum(|packer| {
            packer
                .push_range(range)
                .expect("must have already handled errors")
        }))
    }

    /// The output SqlColumnType of this function
    fn output_type(&self, input_type: SqlColumnType) -> SqlColumnType {
        self.return_ty
            .without_modifiers()
            .nullable(input_type.nullable)
    }

    /// Whether this function will produce NULL on NULL input
    fn propagates_nulls(&self) -> bool {
        true
    }

    /// Whether this function will produce NULL on non-NULL input
    fn introduces_nulls(&self) -> bool {
        false
    }

    /// Whether this function preserves uniqueness
    fn preserves_uniqueness(&self) -> bool {
        false
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        to_unary!(super::CastRangeToString {
            ty: self.return_ty.clone(),
        })
    }

    fn is_monotone(&self) -> bool {
        false
    }
}

impl fmt::Display for CastStringToRange {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("strtorange")
    }
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct CastStringToVarChar {
    pub length: Option<VarCharMaxLength>,
    pub fail_on_len: bool,
}

impl EagerUnaryFunc for CastStringToVarChar {
    type Input<'a> = &'a str;
    type Output<'a> = Result<VarChar<&'a str>, EvalError>;

    fn call<'a>(&self, a: Self::Input<'a>) -> Self::Output<'a> {
        let s =
            mz_repr::adt::varchar::format_str(a, self.length, self.fail_on_len).map_err(|_| {
                assert!(self.fail_on_len);
                EvalError::StringValueTooLong {
                    target_type: "character varying".into(),
                    length: usize::cast_from(self.length.unwrap().into_u32()),
                }
            })?;

        Ok(VarChar(s))
    }

    fn output_type(&self, input: SqlColumnType) -> SqlColumnType {
        SqlScalarType::VarChar {
            max_length: self.length,
        }
        .nullable(input.nullable)
    }

    fn could_error(&self) -> bool {
        self.fail_on_len && self.length.is_some()
    }

    fn preserves_uniqueness(&self) -> bool {
        !self.fail_on_len || self.length.is_none()
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        to_unary!(super::CastVarCharToString)
    }
}

impl fmt::Display for CastStringToVarChar {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        match self.length {
            Some(length) => {
                write!(
                    f,
                    "text_to_varchar[len={}, fail_on_len={}]",
                    length.into_u32(),
                    self.fail_on_len
                )
            }
            None => f.write_str("text_to_varchar[len=unbounded]"),
        }
    }
}

// If we support another vector type, this should likely get hoisted into a
// position akin to array parsing.
static INT2VECTOR_CAST_EXPR: LazyLock<MirScalarExpr> = LazyLock::new(|| MirScalarExpr::CallUnary {
    func: UnaryFunc::CastStringToInt16(CastStringToInt16),
    expr: Box::new(MirScalarExpr::column(0)),
});

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct CastStringToInt2Vector;

impl LazyUnaryFunc for CastStringToInt2Vector {
    fn eval<'a>(
        &'a self,
        datums: &[Datum<'a>],
        temp_storage: &'a RowArena,
        a: &'a MirScalarExpr,
    ) -> Result<Datum<'a>, EvalError> {
        let a = a.eval(datums, temp_storage)?;
        if a.is_null() {
            return Ok(Datum::Null);
        }

        let datums = strconv::parse_legacy_vector(a.unwrap_str(), |elem_text| {
            let elem_text = match elem_text {
                Cow::Owned(s) => temp_storage.push_string(s),
                Cow::Borrowed(s) => s,
            };
            INT2VECTOR_CAST_EXPR.eval(&[Datum::String(elem_text)], temp_storage)
        })?;
        array_create_scalar(&datums, temp_storage)
    }

    /// The output SqlColumnType of this function
    fn output_type(&self, input_type: SqlColumnType) -> SqlColumnType {
        SqlScalarType::Int2Vector.nullable(input_type.nullable)
    }

    /// Whether this function will produce NULL on NULL input
    fn propagates_nulls(&self) -> bool {
        true
    }

    /// Whether this function will produce NULL on non-NULL input
    fn introduces_nulls(&self) -> bool {
        false
    }

    /// Whether this function preserves uniqueness
    fn preserves_uniqueness(&self) -> bool {
        false
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        to_unary!(super::CastInt2VectorToString)
    }

    fn is_monotone(&self) -> bool {
        false
    }
}

impl fmt::Display for CastStringToInt2Vector {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        f.write_str("strtoint2vector")
    }
}

#[sqlfunc(
    sqlname = "text_to_jsonb",
    preserves_uniqueness = false,
    inverse = to_unary!(super::CastJsonbToString)
)]
// TODO(jamii): it would be much more efficient to skip the intermediate repr::jsonb::Jsonb.
fn cast_string_to_jsonb<'a>(a: &'a str) -> Result<Jsonb, EvalError> {
    Ok(strconv::parse_jsonb(a)?)
}

#[sqlfunc(sqlname = "btrim")]
fn trim_whitespace<'a>(a: &'a str) -> &'a str {
    a.trim_matches(' ')
}

#[sqlfunc(sqlname = "ltrim")]
fn trim_leading_whitespace<'a>(a: &'a str) -> &'a str {
    a.trim_start_matches(' ')
}

#[sqlfunc(sqlname = "rtrim")]
fn trim_trailing_whitespace<'a>(a: &'a str) -> &'a str {
    a.trim_end_matches(' ')
}

#[sqlfunc(sqlname = "initcap")]
fn initcap<'a>(a: &'a str) -> String {
    let mut out = String::new();
    let mut capitalize_next = true;
    for ch in a.chars() {
        if capitalize_next {
            out.extend(ch.to_uppercase())
        } else {
            out.extend(ch.to_lowercase())
        };
        capitalize_next = !ch.is_alphanumeric();
    }
    out
}

#[sqlfunc(sqlname = "ascii")]
fn ascii<'a>(a: &'a str) -> i32 {
    a.chars()
        .next()
        .and_then(|c| i32::try_from(u32::from(c)).ok())
        .unwrap_or(0)
}

#[sqlfunc(sqlname = "char_length")]
fn char_length<'a>(a: &'a str) -> Result<i32, EvalError> {
    let length = a.chars().count();
    i32::try_from(length).or_else(|_| Err(EvalError::Int32OutOfRange(length.to_string().into())))
}

#[sqlfunc(sqlname = "bit_length")]
fn bit_length_string<'a>(a: &'a str) -> Result<i32, EvalError> {
    let length = a.as_bytes().len() * 8;
    i32::try_from(length).or_else(|_| Err(EvalError::Int32OutOfRange(length.to_string().into())))
}

#[sqlfunc(sqlname = "octet_length")]
fn byte_length_string<'a>(a: &'a str) -> Result<i32, EvalError> {
    let length = a.as_bytes().len();
    i32::try_from(length).or_else(|_| Err(EvalError::Int32OutOfRange(length.to_string().into())))
}

#[sqlfunc]
fn upper<'a>(a: &'a str) -> String {
    a.to_uppercase()
}

#[sqlfunc]
fn lower<'a>(a: &'a str) -> String {
    a.to_lowercase()
}

#[sqlfunc]
fn normalize(text: &str, form_str: &str) -> Result<String, EvalError> {
    use unicode_normalization::UnicodeNormalization;

    match form_str.to_uppercase().as_str() {
        "NFC" => Ok(text.nfc().collect()),
        "NFD" => Ok(text.nfd().collect()),
        "NFKC" => Ok(text.nfkc().collect()),
        "NFKD" => Ok(text.nfkd().collect()),
        _ => Err(EvalError::InvalidParameterValue(
            format!("invalid normalization form: {}", form_str).into(),
        )),
    }
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct IsLikeMatch(pub like_pattern::Matcher);

impl EagerUnaryFunc for IsLikeMatch {
    type Input<'a> = &'a str;
    type Output<'a> = bool;

    fn call<'a>(&self, haystack: Self::Input<'a>) -> Self::Output<'a> {
        self.0.is_match(haystack)
    }

    fn output_type(&self, input: SqlColumnType) -> SqlColumnType {
        SqlScalarType::Bool.nullable(input.nullable)
    }
}

impl fmt::Display for IsLikeMatch {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(
            f,
            "{}like[{}]",
            if self.0.case_insensitive { "i" } else { "" },
            self.0.pattern.escaped()
        )
    }
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct IsRegexpMatch(pub Regex);

impl EagerUnaryFunc for IsRegexpMatch {
    type Input<'a> = &'a str;
    type Output<'a> = bool;

    fn call<'a>(&self, haystack: Self::Input<'a>) -> Self::Output<'a> {
        self.0.is_match(haystack)
    }

    fn output_type(&self, input: SqlColumnType) -> SqlColumnType {
        SqlScalarType::Bool.nullable(input.nullable)
    }
}

impl fmt::Display for IsRegexpMatch {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(
            f,
            "is_regexp_match[{}, case_insensitive={}]",
            self.0.pattern().escaped(),
            self.0.case_insensitive
        )
    }
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct RegexpMatch(pub Regex);

impl LazyUnaryFunc for RegexpMatch {
    fn eval<'a>(
        &'a self,
        datums: &[Datum<'a>],
        temp_storage: &'a RowArena,
        a: &'a MirScalarExpr,
    ) -> Result<Datum<'a>, EvalError> {
        let haystack = a.eval(datums, temp_storage)?;
        if haystack.is_null() {
            return Ok(Datum::Null);
        }
        regexp_match_static(haystack, temp_storage, &self.0)
    }

    /// The output SqlColumnType of this function
    fn output_type(&self, _input_type: SqlColumnType) -> SqlColumnType {
        SqlScalarType::Array(Box::new(SqlScalarType::String)).nullable(true)
    }

    /// Whether this function will produce NULL on NULL input
    fn propagates_nulls(&self) -> bool {
        true
    }

    /// Whether this function will produce NULL on non-NULL input
    fn introduces_nulls(&self) -> bool {
        // Returns null if the regex did not match
        true
    }

    /// Whether this function preserves uniqueness
    fn preserves_uniqueness(&self) -> bool {
        false
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        None
    }

    fn is_monotone(&self) -> bool {
        false
    }
}

impl fmt::Display for RegexpMatch {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(
            f,
            "regexp_match[{}, case_insensitive={}]",
            self.0.pattern().escaped(),
            self.0.case_insensitive
        )
    }
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct RegexpSplitToArray(pub Regex);

impl LazyUnaryFunc for RegexpSplitToArray {
    fn eval<'a>(
        &'a self,
        datums: &[Datum<'a>],
        temp_storage: &'a RowArena,
        a: &'a MirScalarExpr,
    ) -> Result<Datum<'a>, EvalError> {
        let haystack = a.eval(datums, temp_storage)?;
        if haystack.is_null() {
            return Ok(Datum::Null);
        }
        regexp_split_to_array_re(haystack.unwrap_str(), &self.0, temp_storage)
    }

    /// The output SqlColumnType of this function
    fn output_type(&self, input_type: SqlColumnType) -> SqlColumnType {
        SqlScalarType::Array(Box::new(SqlScalarType::String)).nullable(input_type.nullable)
    }

    /// Whether this function will produce NULL on NULL input
    fn propagates_nulls(&self) -> bool {
        true
    }

    /// Whether this function will produce NULL on non-NULL input
    fn introduces_nulls(&self) -> bool {
        false
    }

    /// Whether this function preserves uniqueness
    fn preserves_uniqueness(&self) -> bool {
        false
    }

    fn inverse(&self) -> Option<crate::UnaryFunc> {
        None
    }

    fn is_monotone(&self) -> bool {
        false
    }
}

impl fmt::Display for RegexpSplitToArray {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(
            f,
            "regexp_split_to_array[{}, case_insensitive={}]",
            self.0.pattern().escaped(),
            self.0.case_insensitive
        )
    }
}

#[sqlfunc(sqlname = "mz_panic")]
fn panic<'a>(a: &'a str) -> String {
    print!("{}", a);
    panic!("{}", a)
}

#[sqlfunc(sqlname = "quote_ident", preserves_uniqueness = true)]
fn quote_ident<'a>(a: &'a str) -> Result<String, EvalError> {
    let i = mz_sql_parser::ast::Ident::new(a).map_err(|err| EvalError::InvalidIdentifier {
        ident: a.into(),
        detail: Some(err.to_string().into()),
    })?;
    Ok(i.to_string())
}

#[derive(
    Ord,
    PartialOrd,
    Clone,
    Debug,
    Eq,
    PartialEq,
    Serialize,
    Deserialize,
    Hash,
    MzReflect
)]
pub struct RegexpReplace {
    pub regex: Regex,
    pub limit: usize,
}

impl binary::EagerBinaryFunc for RegexpReplace {
    type Input<'a> = (&'a str, &'a str);
    type Output<'a> = Cow<'a, str>;

    fn call<'a>(
        &self,
        (source, replacement): Self::Input<'a>,
        _temp_storage: &'a RowArena,
    ) -> Self::Output<'a> {
        // WARNING: This function has potential OOM risk if used with an inflationary
        // replacement pattern. It is very difficult to calculate the output size ahead
        // of time because the replacement pattern may depend on capture groups.
        self.regex.replacen(source, self.limit, replacement)
    }

    fn output_type(&self, input_types: &[SqlColumnType]) -> SqlColumnType {
        use mz_repr::AsColumnType;
        let output = <Self::Output<'_> as AsColumnType>::as_column_type();
        let propagates_nulls = binary::EagerBinaryFunc::propagates_nulls(self);
        let nullable = output.nullable;
        let input_nullable = input_types.iter().any(|t| t.nullable);
        output.nullable(nullable || (propagates_nulls && input_nullable))
    }
}

impl fmt::Display for RegexpReplace {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(
            f,
            "regexp_replace[{}, case_insensitive={}, limit={}]",
            self.regex.pattern().escaped(),
            self.regex.case_insensitive,
            self.limit
        )
    }
}