feat: support create table with map type

support cast

improve check

e2e_test/batch/distribution_mode.slt

@@ -4,7 +4,7 @@ SET RW_IMPLICIT_FLUSH TO true;
 statement ok
 SET QUERY_MODE TO distributed;
 
-include ./basic/*.slt.part
+include ./basic/**/*.slt.part
 include ./duckdb/all.slt.part
 include ./order/*.slt.part
 include ./join/*.slt.part

e2e_test/batch/local_mode.slt

@@ -4,7 +4,7 @@ SET RW_IMPLICIT_FLUSH TO true;
 statement ok
 SET QUERY_MODE TO local;
 
-include ./basic/*.slt.part
+include ./basic/**/*.slt.part
 include ./duckdb/all.slt.part
 include ./order/*.slt.part
 include ./join/*.slt.part

e2e_test/batch/types/map.slt.part

@@ -0,0 +1,120 @@
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+
+statement error
+create table t (m map (float, float));
+----
+db error: ERROR: Failed to run the query
+
+Caused by:
+  invalid map key type: double precision
+
+
+query error
+select map_from_entries(array[1.0,2.0,3.0], array[1,2,3]);
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Failed to bind expression: map_from_entries(ARRAY[1.0, 2.0, 3.0], ARRAY[1, 2, 3])
+  2: Expr error
+  3: invalid map key type: numeric
+
+
+query error
+select map_from_entries(array[1,1,3], array[1,2,3]);
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Expr error
+  2: error while evaluating expression `map('{1,1,3}', '{1,2,3}')`
+  3: map keys must be unique
+
+
+query ?
+select map_from_entries(array[1,2,3], array[1,null,3]);
+----
+{"1":1,"2":NULL,"3":3}
+
+
+query error
+select map_from_entries(array[1,null,3], array[1,2,3]);
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Expr error
+  2: error while evaluating expression `map('{1,NULL,3}', '{1,2,3}')`
+  3: map keys must not be NULL
+
+
+query error
+select map_from_entries(array[1,3], array[1,2,3]);
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Expr error
+  2: error while evaluating expression `map('{1,3}', '{1,2,3}')`
+  3: map keys and values have different length
+
+
+query error
+select map_from_entries(array[1,2], array[1,2]) = map_from_entries(array[2,1], array[2,1]);
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Failed to bind expression: map_from_entries(ARRAY[1, 2], ARRAY[1, 2]) = map_from_entries(ARRAY[2, 1], ARRAY[2, 1])
+  2: function equal(map(integer,integer), map(integer,integer)) does not exist
+
+
+statement ok
+create table t (
+    m1 map(varchar, float),
+    m2 map(int, bool),
+    m3 map(varchar, map(varchar, varchar)),
+    l map(varchar,int)[],
+    s struct<m map(varchar, struct<x int>)>,
+);
+
+
+statement ok
+insert into t values (
+    map_from_entries(array['a','b','c'], array[1.0,2.0,3.0]::float[]),
+    map_from_entries(array[1,2,3], array[true,false,true]),
+    map_from_entries(array['a','b'],
+        array[
+            map_from_entries(array['a1'], array['a2']),
+            map_from_entries(array['b1'], array['b2'])
+        ]
+    ),
+    array[
+        map_from_entries(array['a','b','c'], array[1,2,3]),
+        map_from_entries(array['d','e','f'], array[4,5,6])
+    ],
+    row(
+        map_from_entries(array['a','b','c'], array[row(1),row(2),row(3)]::struct<x int>[])
+    )
+);
+
+# cast(map(character varying,integer)) -> map(character varying,double precision)
+query ?
+select map_from_entries(array['a','b','c'], array[1,2,3])::map(varchar,float);
+----
+{"a":1,"b":2,"c":3}
+
+
+statement ok
+insert into t(m1) values (map_from_entries(array['a','b','c'], array[1,2,3]));
+
+query ????? rowsort
+select * from t;
+----
+{"a":1,"b":2,"c":3}	NULL	NULL	NULL	NULL
+{"a":1,"b":2,"c":3}	{"1":t,"2":f,"3":t}	{"a":{"a1":a2},"b":{"b1":b2}}	{"{\"a\":1,\"b\":2,\"c\":3}","{\"d\":4,\"e\":5,\"f\":6}"}	("{""a"":(1),""b"":(2),""c"":(3)}")
+
+statement ok
+drop table t;

src/expr/impl/src/scalar/cast.rs

@@ -21,7 +21,7 @@ use itertools::Itertools;
 use risingwave_common::array::{ArrayImpl, DataChunk, ListRef, ListValue, StructRef, StructValue};
 use risingwave_common::cast;
 use risingwave_common::row::OwnedRow;
-use risingwave_common::types::{Int256, JsonbRef, ToText, F64};
+use risingwave_common::types::{Int256, JsonbRef, MapRef, MapValue, ToText, F64};
 use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_expr::expr::{build_func, Context, ExpressionBoxExt, InputRefExpression};
 use risingwave_expr::{function, ExprError, Result};
@@ -241,6 +241,17 @@ fn struct_cast(input: StructRef<'_>, ctx: &Context) -> Result<StructValue> {
     Ok(StructValue::new(fields))
 }
 
+/// Cast array with `source_elem_type` into array with `target_elem_type` by casting each element.
+#[function("cast(anymap) -> anymap", type_infer = "panic")]
+fn map_cast(map: MapRef<'_>, ctx: &Context) -> Result<MapValue> {
+    let new_ctx = Context {
+        arg_types: vec![ctx.arg_types[0].clone().as_map().clone().into_list()],
+        return_type: ctx.return_type.as_map().clone().into_list(),
+        variadic: ctx.variadic,
+    };
+    list_cast(map.into_inner(), &new_ctx).map(MapValue::from_list_entries)
+}
+
 #[cfg(test)]
 mod tests {
     use chrono::NaiveDateTime;

src/frontend/planner_test/tests/testdata/output/insert.yaml

@@ -80,7 +80,11 @@
   sql: |
     create table t (v1 real, v2 int);
     insert into t values (22.33, true);
-  binder_error: 'Bind error: cannot cast type "boolean" to "integer" in Assign context'
+  binder_error: |
+    failed to cast the 2nd column
+
+    Caused by:
+      cannot cast type "boolean" to "integer" in Assign context
 - name: simple insert
   sql: |
     create table t (v1 int, v2 int);
@@ -175,7 +179,11 @@
   sql: |
     create table t (v1 timestamp, v2 real);
     insert into t select time '01:02:03', 4.5 from t;
-  binder_error: 'Bind error: cannot cast type "time without time zone" to "timestamp without time zone" in Assign context'
+  binder_error: |
+    failed to cast the 1st column
+
+    Caused by:
+      cannot cast type "time without time zone" to "timestamp without time zone" in Assign context
 - name: insert into select mismatch columns length
   sql: |
     create table t (v1 int, v2 real);

src/frontend/src/binder/expr/mod.rs

@@ -14,7 +14,7 @@
 
 use itertools::Itertools;
 use risingwave_common::catalog::{ColumnDesc, ColumnId, PG_CATALOG_SCHEMA_NAME};
-use risingwave_common::types::DataType;
+use risingwave_common::types::{DataType, MapType};
 use risingwave_common::util::iter_util::zip_eq_fast;
 use risingwave_common::{bail_no_function, bail_not_implemented, not_implemented};
 use risingwave_pb::plan_common::{AdditionalColumn, ColumnDescVersion};
@@ -999,6 +999,11 @@ pub fn bind_data_type(data_type: &AstDataType) -> Result<DataType> {
                 .collect::<Result<Vec<_>>>()?,
             types.iter().map(|f| f.name.real_value()).collect_vec(),
         ),
+        AstDataType::Map(kv) => {
+            let key = bind_data_type(&kv.0)?;
+            let value = bind_data_type(&kv.1)?;
+            DataType::Map(MapType::try_from_kv(key, value)?)
+        }
         AstDataType::Custom(qualified_type_name) => {
             let idents = qualified_type_name
                 .0

src/frontend/src/binder/insert.rs

@@ -14,6 +14,7 @@
 
 use std::collections::{BTreeMap, HashMap, HashSet};
 
+use anyhow::Context;
 use itertools::Itertools;
 use risingwave_common::catalog::{ColumnCatalog, Schema, TableVersionId};
 use risingwave_common::types::DataType;
@@ -26,6 +27,7 @@ use crate::binder::{Binder, Clause};
 use crate::catalog::TableId;
 use crate::error::{ErrorCode, Result, RwError};
 use crate::expr::{ExprImpl, InputRef};
+use crate::handler::create_mv::ordinal;
 use crate::user::UserId;
 
 #[derive(Debug, Clone)]
@@ -197,7 +199,7 @@ impl Binder {
         let bound_query;
         let cast_exprs;
 
-        let bounded_column_nums = match source.as_simple_values() {
+        let bound_column_nums = match source.as_simple_values() {
             None => {
                 bound_query = self.bind_query(source)?;
                 let actual_types = bound_query.data_types();
@@ -234,7 +236,7 @@ impl Binder {
             cols_to_insert_in_table.len()
         };
 
-        let (err_msg, default_column_indices) = match num_target_cols.cmp(&bounded_column_nums) {
+        let (err_msg, default_column_indices) = match num_target_cols.cmp(&bound_column_nums) {
             std::cmp::Ordering::Equal => (None, default_column_indices),
             std::cmp::Ordering::Greater => {
                 if has_user_specified_columns {
@@ -248,7 +250,7 @@ impl Binder {
                     //      insert into t values (7)
                     // this kind of usage is fine, null values will be provided
                     // implicitly.
-                    (None, col_indices_to_insert.split_off(bounded_column_nums))
+                    (None, col_indices_to_insert.split_off(bound_column_nums))
                 }
             }
             std::cmp::Ordering::Less => {
@@ -312,10 +314,22 @@ impl Binder {
         let msg = match expected_types.len().cmp(&exprs.len()) {
             std::cmp::Ordering::Less => "INSERT has more expressions than target columns",
             _ => {
+                let expr_len = exprs.len();
                 return exprs
                     .into_iter()
                     .zip_eq_fast(expected_types.iter().take(expr_num))
-                    .map(|(e, t)| e.cast_assign(t.clone()).map_err(Into::into))
+                    .enumerate()
+                    .map(|(i, (e, t))| {
+                        let res = e.cast_assign(t.clone());
+                        if expr_len > 1 {
+                            res.with_context(|| {
+                                format!("failed to cast the {} column", ordinal(i + 1))
+                            })
+                            .map_err(Into::into)
+                        } else {
+                            res.map_err(Into::into)
+                        }
+                    })
                     .try_collect();
             }
         };

src/frontend/src/binder/select.rs

@@ -714,8 +714,8 @@ fn data_type_to_alias(data_type: &AstDataType) -> Option<String> {
         AstDataType::Jsonb => "jsonb".to_string(),
         AstDataType::Array(ty) => return data_type_to_alias(ty),
         AstDataType::Custom(ty) => format!("{}", ty),
-        AstDataType::Struct(_) => {
-            // Note: Postgres doesn't have anonymous structs
+        AstDataType::Struct(_) | AstDataType::Map(_) => {
+            // Note: Postgres doesn't have maps and anonymous structs
             return None;
         }
     };

src/frontend/src/expr/type_inference/cast.rs

@@ -114,6 +114,7 @@ pub fn align_array_and_element(
 pub fn cast_ok(source: &DataType, target: &DataType, allows: CastContext) -> bool {
     cast_ok_struct(source, target, allows)
         || cast_ok_array(source, target, allows)
+        || cast_ok_map(source, target, allows)
         || cast_ok_base(source, target, allows)
 }
 
@@ -161,6 +162,17 @@ fn cast_ok_array(source: &DataType, target: &DataType, allows: CastContext) -> b
     }
 }
 
+fn cast_ok_map(source: &DataType, target: &DataType, allows: CastContext) -> bool {
+    match (source, target) {
+        (DataType::Map(source_elem), DataType::Map(target_elem)) => cast_ok(
+            &source_elem.clone().into_list(),
+            &target_elem.clone().into_list(),
+            allows,
+        ),
+        _ => false,
+    }
+}
+
 pub fn cast_map_array() -> Vec<(DataTypeName, DataTypeName, CastContext)> {
     CAST_MAP
         .iter()

src/frontend/src/handler/create_mv.rs

@@ -278,7 +278,7 @@ It only indicates the physical clustering of the data, which may improve the per
     ))
 }
 
-fn ordinal(i: usize) -> String {
+pub fn ordinal(i: usize) -> String {
     let s = i.to_string();
     let suffix = if s.ends_with('1') && !s.ends_with("11") {
         "st"

src/sqlparser/src/ast/data_type.rs

@@ -70,6 +70,8 @@ pub enum DataType {
     Array(Box<DataType>),
     /// Structs
     Struct(Vec<StructField>),
+    /// Map(key_type, value_type)
+    Map(Box<(DataType, DataType)>),
 }
 
 impl fmt::Display for DataType {
@@ -110,6 +112,9 @@ impl fmt::Display for DataType {
             DataType::Struct(defs) => {
                 write!(f, "STRUCT<{}>", display_comma_separated(defs))
             }
+            DataType::Map(kv) => {
+                write!(f, "MAP({},{})", kv.0, kv.1)
+            }
         }
     }
 }

src/sqlparser/src/parser.rs

@@ -3632,8 +3632,7 @@ impl Parser<'_> {
         .parse_next(self)
     }
 
-    /// Parse a SQL datatype (in the context of a CREATE TABLE statement for example) and convert
-    /// into an array of that datatype if needed
+    /// Parse a SQL datatype (in the context of a CREATE TABLE statement for example)
     pub fn parse_data_type(&mut self) -> PResult<DataType> {
         parser_v2::data_type(self)
     }