mz_expr/scalar/func/impls/
map.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use itertools::Itertools;
11use mz_expr_derive::sqlfunc;
12use mz_repr::{Datum, DatumList, DatumMap, ExcludeNull, RowArena, SqlScalarType};
13use serde::{Deserialize, Serialize};
14
15use crate::EvalError;
16use crate::scalar::func::stringify_datum;
17
18#[derive(
19    Ord,
20    PartialOrd,
21    Clone,
22    Debug,
23    Eq,
24    PartialEq,
25    Serialize,
26    Deserialize,
27    Hash
28)]
29pub struct CastMapToString {
30    pub ty: SqlScalarType,
31}
32
33// TODO? If we moved typeconv into expr, we could evaluate the inverse of this cast
34#[sqlfunc(CastMapToString, sqlname = "maptostr", preserves_uniqueness = true)]
35fn cast_map_to_string<'a>(&self, a: ExcludeNull<Datum<'a>>) -> Result<String, EvalError> {
36    let mut buf = String::new();
37    stringify_datum(&mut buf, *a, &self.ty)?;
38    Ok(buf)
39}
40
41#[sqlfunc(sqlname = "map_length")]
42fn map_length<'a>(a: DatumMap<'a>) -> Result<i32, EvalError> {
43    let count = a.iter().count();
44    count
45        .try_into()
46        .map_err(|_| EvalError::Int32OutOfRange(count.to_string().into()))
47}
48
49#[derive(
50    Ord,
51    PartialOrd,
52    Clone,
53    Debug,
54    Eq,
55    PartialEq,
56    Serialize,
57    Deserialize,
58    Hash
59)]
60pub struct MapBuildFromRecordList {
61    pub value_type: SqlScalarType,
62}
63
64#[sqlfunc(
65    MapBuildFromRecordList,
66    sqlname = "map_build",
67    output_type_expr = SqlScalarType::Map {
68        value_type: Box::new(self.value_type.clone()),
69        custom_id: None,
70    }
71    .nullable(true),
72    // NOTE: The body never returns NULL, but a non-nullable output changes the column
73    // types of existing materialized views that call this function. A Kafka sink with
74    // Avro on such a view regenerates its schema on upgrade, and the schema registry
75    // rejects the narrowed schema as incompatible, which stalls the sink. Iceberg sinks
76    // require nullability to match the table exactly (`field_is_compatible`).
77    introduces_nulls = true
78)]
79fn map_build_from_record_list<'a>(
80    &self,
81    a: DatumList<'a>,
82    temp_storage: &'a RowArena,
83) -> DatumMap<'a> {
84    let mut map = std::collections::BTreeMap::new();
85
86    for i in a.iter() {
87        if i.is_null() {
88            continue;
89        }
90
91        for (k, v) in i.unwrap_list().iter().tuples() {
92            if k.is_null() {
93                continue;
94            }
95            map.insert(k.unwrap_str(), v);
96        }
97    }
98
99    temp_storage
100        .make_datum(|packer| packer.push_dict(map))
101        .unwrap_map()
102}
mz_expr/scalar/func/impls/map.rs

mz_expr/scalar/func/impls/
map.rs
