risingwave_expr_impl/scalar/
cast.rs

1// Copyright 2023 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::str::FromStr;
16use std::sync::Arc;
17
18use futures_util::FutureExt;
19use itertools::Itertools;
20use risingwave_common::array::{DataChunk, ListRef, ListValue, StructRef, StructValue, VectorVal};
21use risingwave_common::cast;
22use risingwave_common::row::OwnedRow;
23use risingwave_common::types::{
24    DataType, F64, Int256, JsonbRef, JsonbVal, MapRef, MapValue, ScalarRef as _, Serial,
25    Timestamptz, ToText, VariantRef, VariantVal,
26};
27use risingwave_common::util::iter_util::ZipEqFast;
28use risingwave_common::util::row_id::row_id_to_unix_millis;
29use risingwave_expr::expr::{Context, InputRefExpression, SyncExpressionBoxExt, build_func};
30use risingwave_expr::{ExprError, Result, function};
31use risingwave_pb::expr::expr_node::PbType;
32use thiserror_ext::AsReport;
33
34#[function("cast(varchar) -> *int")]
35#[function("cast(varchar) -> decimal")]
36#[function("cast(varchar) -> *float")]
37#[function("cast(varchar) -> int256")]
38#[function("cast(varchar) -> date")]
39#[function("cast(varchar) -> time")]
40#[function("cast(varchar) -> timestamp")]
41#[function("cast(varchar) -> interval")]
42#[function("cast(varchar) -> jsonb")]
43#[function("cast(varchar) -> variant")]
44pub fn str_parse<T>(elem: &str, ctx: &Context) -> Result<T>
45where
46    T: FromStr,
47    <T as FromStr>::Err: std::fmt::Display,
48{
49    elem.trim().parse().map_err(|err: <T as FromStr>::Err| {
50        ExprError::Parse(format!("{} {}", ctx.return_type, err).into())
51    })
52}
53
54// TODO: introduce `FromBinary` and support all types
55#[function("pgwire_recv(bytea) -> int8")]
56pub fn pgwire_recv(elem: &[u8]) -> Result<i64> {
57    let fixed_length =
58        <[u8; 8]>::try_from(elem).map_err(|e| ExprError::Parse(e.to_report_string().into()))?;
59    Ok(i64::from_be_bytes(fixed_length))
60}
61
62#[function("cast(int2) -> int256")]
63#[function("cast(int4) -> int256")]
64#[function("cast(int8) -> int256")]
65pub fn to_int256<T: TryInto<Int256>>(elem: T) -> Result<Int256> {
66    elem.try_into()
67        .map_err(|_| ExprError::CastOutOfRange("int256"))
68}
69
70#[function("cast(jsonb) -> boolean")]
71pub fn jsonb_to_bool(v: JsonbRef<'_>) -> Result<Option<bool>> {
72    if v.is_jsonb_null() {
73        Ok(None)
74    } else {
75        v.as_bool()
76            .map(Some)
77            .map_err(|e| ExprError::Parse(e.into()))
78    }
79}
80
81/// Note that PostgreSQL casts JSON numbers from arbitrary precision `numeric` but we use `f64`.
82/// This is less powerful but still meets RFC 8259 interoperability.
83#[function("cast(jsonb) -> int2")]
84#[function("cast(jsonb) -> int4")]
85#[function("cast(jsonb) -> int8")]
86#[function("cast(jsonb) -> decimal")]
87#[function("cast(jsonb) -> float4")]
88#[function("cast(jsonb) -> float8")]
89pub fn jsonb_to_number<T: TryFrom<F64>>(v: JsonbRef<'_>) -> Result<Option<T>> {
90    if v.is_jsonb_null() {
91        Ok(None)
92    } else {
93        v.as_number()
94            .map_err(|e| ExprError::Parse(e.into()))?
95            .try_into()
96            .map(Some)
97            .map_err(|_| ExprError::NumericOutOfRange)
98    }
99}
100
101#[function("cast(int4) -> int2")]
102#[function("cast(int8) -> int2")]
103#[function("cast(int8) -> int4")]
104#[function("cast(int8) -> serial")]
105#[function("cast(serial) -> int8")]
106#[function("cast(float4) -> int2")]
107#[function("cast(float8) -> int2")]
108#[function("cast(float4) -> int4")]
109#[function("cast(float8) -> int4")]
110#[function("cast(float4) -> int8")]
111#[function("cast(float8) -> int8")]
112#[function("cast(float8) -> float4")]
113#[function("cast(decimal) -> int2")]
114#[function("cast(decimal) -> int4")]
115#[function("cast(decimal) -> int8")]
116#[function("cast(decimal) -> float4")]
117#[function("cast(decimal) -> float8")]
118#[function("cast(float4) -> decimal")]
119#[function("cast(float8) -> decimal")]
120pub fn try_cast<T1, T2>(elem: T1) -> Result<T2>
121where
122    T1: TryInto<T2> + std::fmt::Debug + Copy,
123{
124    elem.try_into()
125        .map_err(|_| ExprError::CastOutOfRange(std::any::type_name::<T2>()))
126}
127
128#[function("cast(boolean) -> int4")]
129#[function("cast(int2) -> int4")]
130#[function("cast(int2) -> int8")]
131#[function("cast(int2) -> float4")]
132#[function("cast(int2) -> float8")]
133#[function("cast(int2) -> decimal")]
134#[function("cast(int4) -> int8")]
135#[function("cast(int4) -> float4")]
136#[function("cast(int4) -> float8")]
137#[function("cast(int4) -> decimal")]
138#[function("cast(int8) -> float4")]
139#[function("cast(int8) -> float8")]
140#[function("cast(int8) -> decimal")]
141#[function("cast(float4) -> float8")]
142#[function("cast(date) -> timestamp")]
143#[function("cast(time) -> interval")]
144#[function("cast(timestamp) -> date")]
145#[function("cast(timestamp) -> time")]
146#[function("cast(interval) -> time")]
147#[function("cast(varchar) -> varchar")]
148#[function("cast(int256) -> float8")]
149pub fn cast<T1, T2>(elem: T1) -> T2
150where
151    T1: Into<T2>,
152{
153    elem.into()
154}
155
156#[function("cast(jsonb) -> variant")]
157pub fn jsonb_to_variant(elem: JsonbRef<'_>) -> Result<VariantVal> {
158    VariantVal::from_jsonb(elem).map_err(|e| ExprError::Parse(e.to_report_string().into()))
159}
160
161#[function("cast(variant) -> jsonb")]
162pub fn variant_to_jsonb(elem: VariantRef<'_>) -> Result<JsonbVal> {
163    elem.to_jsonb()
164        .map_err(|e| ExprError::Parse(e.to_report_string().into()))
165}
166
167/// Extract the timestamp from row id.
168#[function("cast(serial) -> timestamptz")]
169pub fn serial_to_timestamptz(elem: Serial) -> Result<Timestamptz> {
170    let unix_ms = row_id_to_unix_millis(elem.as_row_id()).ok_or(ExprError::NumericOutOfRange)?;
171    Timestamptz::from_millis(unix_ms).ok_or(ExprError::NumericOutOfRange)
172}
173
174#[function("cast(varchar) -> boolean")]
175pub fn str_to_bool(input: &str) -> Result<bool> {
176    cast::str_to_bool(input).map_err(|err| ExprError::Parse(err.into()))
177}
178
179#[function("cast(int4) -> boolean")]
180pub fn int_to_bool(input: i32) -> bool {
181    input != 0
182}
183
184/// For most of the types, cast them to varchar is the same as their pgwire "TEXT" format.
185/// So we use `ToText` to cast type to varchar.
186#[function("cast(*int) -> varchar")]
187#[function("cast(decimal) -> varchar")]
188#[function("cast(*float) -> varchar")]
189#[function("cast(int256) -> varchar")]
190#[function("cast(time) -> varchar")]
191#[function("cast(date) -> varchar")]
192#[function("cast(interval) -> varchar")]
193#[function("cast(timestamp) -> varchar")]
194#[function("cast(jsonb) -> varchar")]
195#[function("cast(variant) -> varchar")]
196#[function("cast(bytea) -> varchar")]
197#[function("cast(anyarray) -> varchar")]
198#[function("cast(vector) -> varchar")]
199pub fn general_to_text(elem: impl ToText, mut writer: &mut impl std::fmt::Write) {
200    elem.write(&mut writer).unwrap();
201}
202
203// TODO: use `ToBinary` and support all types
204#[function("pgwire_send(int8) -> bytea")]
205fn pgwire_send(elem: i64, writer: &mut impl std::io::Write) {
206    writer.write_all(&elem.to_be_bytes()).unwrap();
207}
208
209#[function("cast(boolean) -> varchar")]
210pub fn bool_to_varchar(input: bool, writer: &mut impl std::fmt::Write) {
211    writer
212        .write_str(if input { "true" } else { "false" })
213        .unwrap();
214}
215
216/// `bool_out` is different from `cast(boolean) -> varchar` to produce a single char. `PostgreSQL`
217/// uses different variants of bool-to-string in different situations.
218#[function("bool_out(boolean) -> varchar")]
219pub fn bool_out(input: bool, writer: &mut impl std::fmt::Write) {
220    writer.write_str(if input { "t" } else { "f" }).unwrap();
221}
222
223#[function("cast(varchar) -> bytea")]
224pub fn str_to_bytea(elem: &str, writer: &mut impl std::io::Write) -> Result<()> {
225    cast::str_to_bytea(elem, writer).map_err(|err| ExprError::Parse(err.into()))
226}
227
228#[function("cast(varchar) -> anyarray", type_infer = "unreachable")]
229fn str_to_list(input: &str, ctx: &Context) -> Result<ListValue> {
230    ListValue::from_str(input, &ctx.return_type).map_err(|err| ExprError::Parse(err.into()))
231}
232
233#[function("cast(varchar) -> vector", type_infer = "unreachable")]
234fn str_to_vector(input: &str, ctx: &Context) -> Result<VectorVal> {
235    let DataType::Vector(size) = &ctx.return_type else {
236        unreachable!()
237    };
238    VectorVal::from_text(input, *size).map_err(|err| ExprError::Parse(err.into()))
239}
240
241/// Cast array with `source_elem_type` into array with `target_elem_type` by casting each element.
242#[function("cast(anyarray) -> anyarray", type_infer = "unreachable")]
243fn list_cast(input: ListRef<'_>, ctx: &Context) -> Result<ListValue> {
244    let cast = build_func(
245        PbType::Cast,
246        ctx.return_type.as_list_elem().clone(),
247        vec![InputRefExpression::new(ctx.arg_types[0].as_list_elem().clone(), 0).boxed()],
248    )
249    .unwrap();
250    let items = Arc::new(input.to_owned_scalar().into_array());
251    let len = items.len();
252    let list = cast
253        .eval(&DataChunk::new(vec![items], len))
254        .now_or_never()
255        .unwrap()?;
256    Ok(ListValue::new(Arc::try_unwrap(list).unwrap()))
257}
258
259/// Cast struct of `source_elem_type` to `target_elem_type` by casting each element.
260#[function("cast(struct) -> struct", type_infer = "unreachable")]
261fn struct_cast(input: StructRef<'_>, ctx: &Context) -> Result<StructValue> {
262    let fields = (input.iter_fields_ref())
263        .zip_eq_fast(ctx.arg_types[0].as_struct().types())
264        .zip_eq_fast(ctx.return_type.as_struct().types())
265        .map(|((datum_ref, source_field_type), target_field_type)| {
266            if source_field_type == target_field_type {
267                return Ok(datum_ref.map(|scalar_ref| scalar_ref.into_scalar_impl()));
268            }
269            let cast = build_func(
270                PbType::Cast,
271                target_field_type.clone(),
272                vec![InputRefExpression::new(source_field_type.clone(), 0).boxed()],
273            )
274            .unwrap();
275            let value = match datum_ref {
276                Some(scalar_ref) => cast
277                    .eval_row(&OwnedRow::new(vec![Some(scalar_ref.into_scalar_impl())]))
278                    .now_or_never()
279                    .unwrap()?,
280                None => None,
281            };
282            Ok(value) as Result<_>
283        })
284        .try_collect()?;
285    Ok(StructValue::new(fields))
286}
287
288/// Cast array with `source_elem_type` into array with `target_elem_type` by casting each element.
289#[function("cast(anymap) -> anymap", type_infer = "unreachable")]
290fn map_cast(map: MapRef<'_>, ctx: &Context) -> Result<MapValue> {
291    let new_ctx = Context {
292        arg_types: vec![ctx.arg_types[0].clone().as_map().clone().into_list()],
293        return_type: ctx.return_type.as_map().clone().into_list(),
294        variadic: ctx.variadic,
295    };
296    list_cast(map.into_inner(), &new_ctx).map(MapValue::from_entries)
297}
298
299#[cfg(test)]
300mod tests {
301    use chrono::NaiveDateTime;
302    use risingwave_common::array::*;
303    use risingwave_common::types::*;
304    use risingwave_expr::expr::build_from_pretty;
305
306    use super::*;
307
308    #[test]
309    fn integer_cast_to_bool() {
310        assert!(int_to_bool(32));
311        assert!(int_to_bool(-32));
312        assert!(!int_to_bool(0));
313    }
314
315    #[test]
316    fn number_to_string() {
317        macro_rules! test {
318            ($fn:ident($value:expr), $right:literal) => {
319                let mut writer = String::new();
320                $fn($value, &mut writer);
321                assert_eq!(writer, $right);
322            };
323        }
324
325        test!(bool_to_varchar(true), "true");
326        test!(bool_to_varchar(true), "true");
327        test!(bool_to_varchar(false), "false");
328
329        test!(general_to_text(32), "32");
330        test!(general_to_text(-32), "-32");
331        test!(general_to_text(i32::MIN), "-2147483648");
332        test!(general_to_text(i32::MAX), "2147483647");
333
334        test!(general_to_text(i16::MIN), "-32768");
335        test!(general_to_text(i16::MAX), "32767");
336
337        test!(general_to_text(i64::MIN), "-9223372036854775808");
338        test!(general_to_text(i64::MAX), "9223372036854775807");
339
340        test!(general_to_text(F64::from(32.12)), "32.12");
341        test!(general_to_text(F64::from(-32.14)), "-32.14");
342
343        test!(general_to_text(F32::from(32.12_f32)), "32.12");
344        test!(general_to_text(F32::from(-32.14_f32)), "-32.14");
345
346        test!(general_to_text(Decimal::try_from(1.222).unwrap()), "1.222");
347
348        test!(general_to_text(Decimal::NaN), "NaN");
349    }
350
351    #[test]
352    fn test_str_to_list() {
353        // Empty List
354        let ctx = Context {
355            arg_types: vec![DataType::Varchar],
356            return_type: DataType::from_str("int[]").unwrap(),
357            variadic: false,
358        };
359        assert_eq!(
360            str_to_list("{}", &ctx).unwrap(),
361            ListValue::empty(&DataType::Varchar)
362        );
363
364        let list123 = ListValue::from_iter([1, 2, 3]);
365
366        // Single List
367        let ctx = Context {
368            arg_types: vec![DataType::Varchar],
369            return_type: DataType::from_str("int[]").unwrap(),
370            variadic: false,
371        };
372        assert_eq!(str_to_list("{1, 2, 3}", &ctx).unwrap(), list123);
373
374        // Nested List
375        let nested_list123 = ListValue::from_iter([list123]);
376        let ctx = Context {
377            arg_types: vec![DataType::Varchar],
378            return_type: DataType::from_str("int[][]").unwrap(),
379            variadic: false,
380        };
381        assert_eq!(str_to_list("{{1, 2, 3}}", &ctx).unwrap(), nested_list123);
382
383        let nested_list445566 = ListValue::from_iter([ListValue::from_iter([44, 55, 66])]);
384
385        let double_nested_list123_445566 =
386            ListValue::from_iter([nested_list123.clone(), nested_list445566.clone()]);
387
388        // Double nested List
389        let ctx = Context {
390            arg_types: vec![DataType::Varchar],
391            return_type: DataType::from_str("int[][][]").unwrap(),
392            variadic: false,
393        };
394        assert_eq!(
395            str_to_list("{{{1, 2, 3}}, {{44, 55, 66}}}", &ctx).unwrap(),
396            double_nested_list123_445566
397        );
398
399        // Cast previous double nested lists to double nested varchar lists
400        let ctx = Context {
401            arg_types: vec![DataType::from_str("int[][]").unwrap()],
402            return_type: DataType::from_str("varchar[][]").unwrap(),
403            variadic: false,
404        };
405        let double_nested_varchar_list123_445566 = ListValue::from_iter([
406            list_cast(nested_list123.as_scalar_ref(), &ctx).unwrap(),
407            list_cast(nested_list445566.as_scalar_ref(), &ctx).unwrap(),
408        ]);
409
410        // Double nested Varchar List
411        let ctx = Context {
412            arg_types: vec![DataType::Varchar],
413            return_type: DataType::from_str("varchar[][][]").unwrap(),
414            variadic: false,
415        };
416        assert_eq!(
417            str_to_list("{{{1, 2, 3}}, {{44, 55, 66}}}", &ctx).unwrap(),
418            double_nested_varchar_list123_445566
419        );
420    }
421
422    #[test]
423    fn test_invalid_str_to_list() {
424        // Unbalanced input
425        let ctx = Context {
426            arg_types: vec![DataType::Varchar],
427            return_type: DataType::from_str("int[]").unwrap(),
428            variadic: false,
429        };
430        assert!(str_to_list("{{}", &ctx).is_err());
431        assert!(str_to_list("{}}", &ctx).is_err());
432        assert!(str_to_list("{{1, 2, 3}, {4, 5, 6}", &ctx).is_err());
433        assert!(str_to_list("{{1, 2, 3}, 4, 5, 6}}", &ctx).is_err());
434    }
435
436    #[test]
437    fn test_struct_cast() {
438        let ctx = Context {
439            arg_types: vec![DataType::Struct(StructType::new(vec![
440                ("a", DataType::Varchar),
441                ("b", DataType::Float32),
442            ]))],
443            return_type: DataType::Struct(StructType::new(vec![
444                ("a", DataType::Int32),
445                ("b", DataType::Int32),
446            ])),
447            variadic: false,
448        };
449        assert_eq!(
450            struct_cast(
451                StructValue::new(vec![
452                    Some("1".into()),
453                    Some(F32::from(0.0).to_scalar_value()),
454                ])
455                .as_scalar_ref(),
456                &ctx,
457            )
458            .unwrap(),
459            StructValue::new(vec![
460                Some(1i32.to_scalar_value()),
461                Some(0i32.to_scalar_value()),
462            ])
463        );
464    }
465
466    #[test]
467    fn test_timestamp() {
468        assert_eq!(
469            try_cast::<_, Timestamp>(Date::from_ymd_uncheck(1994, 1, 1)).unwrap(),
470            Timestamp::new(
471                NaiveDateTime::parse_from_str("1994-1-1 0:0:0", "%Y-%m-%d %H:%M:%S").unwrap()
472            )
473        )
474    }
475
476    #[tokio::test]
477    async fn test_unary() {
478        test_unary_bool::<BoolArray, _>(|x| !x, PbType::Not).await;
479        test_unary_date::<TimestampArray, _>(|x| try_cast(x).unwrap(), PbType::Cast).await;
480        let ctx_str_to_int16 = Context {
481            arg_types: vec![DataType::Varchar],
482            return_type: DataType::Int16,
483            variadic: false,
484        };
485        test_str_to_int16::<I16Array, _>(|x| str_parse(x, &ctx_str_to_int16).unwrap()).await;
486    }
487
488    #[tokio::test]
489    async fn test_i16_to_i32() {
490        let mut input = Vec::<Option<i16>>::new();
491        let mut target = Vec::<Option<i32>>::new();
492        for i in 0..100i16 {
493            if i % 2 == 0 {
494                target.push(Some(i as i32));
495                input.push(Some(i));
496            } else {
497                input.push(None);
498                target.push(None);
499            }
500        }
501        let col1 = I16Array::from_iter(&input).into_ref();
502        let data_chunk = DataChunk::new(vec![col1], 100);
503        let expr = build_from_pretty("(cast:int4 $0:int2)");
504        let res = expr.eval(&data_chunk).await.unwrap();
505        let arr: &I32Array = res.as_ref().into();
506        for (idx, item) in arr.iter().enumerate() {
507            let x = target[idx].as_ref().map(|x| x.as_scalar_ref());
508            assert_eq!(x, item);
509        }
510
511        for i in 0..input.len() {
512            let row = OwnedRow::new(vec![input[i].map(|int| int.to_scalar_value())]);
513            let result = expr.eval_row(&row).await.unwrap();
514            let expected = target[i].map(|int| int.to_scalar_value());
515            assert_eq!(result, expected);
516        }
517    }
518
519    #[tokio::test]
520    async fn test_neg() {
521        let input = [Some(1), Some(0), Some(-1)];
522        let target = [Some(-1), Some(0), Some(1)];
523
524        let col1 = I32Array::from_iter(&input).into_ref();
525        let data_chunk = DataChunk::new(vec![col1], 3);
526        let expr = build_from_pretty("(neg:int4 $0:int4)");
527        let res = expr.eval(&data_chunk).await.unwrap();
528        let arr: &I32Array = res.as_ref().into();
529        for (idx, item) in arr.iter().enumerate() {
530            let x = target[idx].as_ref().map(|x| x.as_scalar_ref());
531            assert_eq!(x, item);
532        }
533
534        for i in 0..input.len() {
535            let row = OwnedRow::new(vec![input[i].map(|int| int.to_scalar_value())]);
536            let result = expr.eval_row(&row).await.unwrap();
537            let expected = target[i].map(|int| int.to_scalar_value());
538            assert_eq!(result, expected);
539        }
540    }
541
542    async fn test_str_to_int16<A, F>(f: F)
543    where
544        A: Array,
545        for<'a> &'a A: std::convert::From<&'a ArrayImpl>,
546        for<'a> <A as Array>::RefItem<'a>: PartialEq,
547        F: Fn(&str) -> <A as Array>::OwnedItem,
548    {
549        let mut input = Vec::<Option<Box<str>>>::new();
550        let mut target = Vec::<Option<<A as Array>::OwnedItem>>::new();
551        for i in 0..1u32 {
552            if i % 2 == 0 {
553                let s = i.to_string().into_boxed_str();
554                target.push(Some(f(&s)));
555                input.push(Some(s));
556            } else {
557                input.push(None);
558                target.push(None);
559            }
560        }
561        let col1_data = &input.iter().map(|x| x.as_ref().map(|x| &**x)).collect_vec();
562        let col1 = Utf8Array::from_iter(col1_data).into_ref();
563        let data_chunk = DataChunk::new(vec![col1], 1);
564        let expr = build_from_pretty("(cast:int2 $0:varchar)");
565        let res = expr.eval(&data_chunk).await.unwrap();
566        let arr: &A = res.as_ref().into();
567        for (idx, item) in arr.iter().enumerate() {
568            let x = target[idx].as_ref().map(|x| x.as_scalar_ref());
569            assert_eq!(x, item);
570        }
571
572        for i in 0..input.len() {
573            let row = OwnedRow::new(vec![
574                input[i].as_ref().cloned().map(|str| str.to_scalar_value()),
575            ]);
576            let result = expr.eval_row(&row).await.unwrap();
577            let expected = target[i].as_ref().cloned().map(|x| x.to_scalar_value());
578            assert_eq!(result, expected);
579        }
580    }
581
582    async fn test_unary_bool<A, F>(f: F, kind: PbType)
583    where
584        A: Array,
585        for<'a> &'a A: std::convert::From<&'a ArrayImpl>,
586        for<'a> <A as Array>::RefItem<'a>: PartialEq,
587        F: Fn(bool) -> <A as Array>::OwnedItem,
588    {
589        let mut input = Vec::<Option<bool>>::new();
590        let mut target = Vec::<Option<<A as Array>::OwnedItem>>::new();
591        for i in 0..100 {
592            if i % 2 == 0 {
593                input.push(Some(true));
594                target.push(Some(f(true)));
595            } else if i % 3 == 0 {
596                input.push(Some(false));
597                target.push(Some(f(false)));
598            } else {
599                input.push(None);
600                target.push(None);
601            }
602        }
603
604        let col1 = BoolArray::from_iter(&input).into_ref();
605        let data_chunk = DataChunk::new(vec![col1], 100);
606        let expr = build_from_pretty(format!("({kind:?}:boolean $0:boolean)"));
607        let res = expr.eval(&data_chunk).await.unwrap();
608        let arr: &A = res.as_ref().into();
609        for (idx, item) in arr.iter().enumerate() {
610            let x = target[idx].as_ref().map(|x| x.as_scalar_ref());
611            assert_eq!(x, item);
612        }
613
614        for i in 0..input.len() {
615            let row = OwnedRow::new(vec![input[i].map(|b| b.to_scalar_value())]);
616            let result = expr.eval_row(&row).await.unwrap();
617            let expected = target[i].as_ref().cloned().map(|x| x.to_scalar_value());
618            assert_eq!(result, expected);
619        }
620    }
621
622    async fn test_unary_date<A, F>(f: F, kind: PbType)
623    where
624        A: Array,
625        for<'a> &'a A: std::convert::From<&'a ArrayImpl>,
626        for<'a> <A as Array>::RefItem<'a>: PartialEq,
627        F: Fn(Date) -> <A as Array>::OwnedItem,
628    {
629        let mut input = Vec::<Option<Date>>::new();
630        let mut target = Vec::<Option<<A as Array>::OwnedItem>>::new();
631        for i in 0..100 {
632            if i % 2 == 0 {
633                let date = Date::from_num_days_from_ce_uncheck(i);
634                input.push(Some(date));
635                target.push(Some(f(date)));
636            } else {
637                input.push(None);
638                target.push(None);
639            }
640        }
641
642        let col1 = DateArray::from_iter(&input).into_ref();
643        let data_chunk = DataChunk::new(vec![col1], 100);
644        let expr = build_from_pretty(format!("({kind:?}:timestamp $0:date)"));
645        let res = expr.eval(&data_chunk).await.unwrap();
646        let arr: &A = res.as_ref().into();
647        for (idx, item) in arr.iter().enumerate() {
648            let x = target[idx].as_ref().map(|x| x.as_scalar_ref());
649            assert_eq!(x, item);
650        }
651
652        for i in 0..input.len() {
653            let row = OwnedRow::new(vec![input[i].map(|d| d.to_scalar_value())]);
654            let result = expr.eval_row(&row).await.unwrap();
655            let expected = target[i].as_ref().cloned().map(|x| x.to_scalar_value());
656            assert_eq!(result, expected);
657        }
658    }
659}
risingwave_expr_impl/scalar/cast.rs

risingwave_expr_impl/scalar/
cast.rs
