[CALCITE-7433] Invalid unparse for cast to map type in Spark

Author: xuzifu666

core/src/main/java/org/apache/calcite/sql/dialect/SparkSqlDialect.java

@@ -169,6 +169,9 @@ public SparkSqlDialect(SqlDialect.Context context) {
     case ARRAY:
       return RelToSqlConverterUtil.getCastSpecAngleBracketArrayType(this, type,
           SqlParserPos.ZERO);
+    case MAP:
+      return RelToSqlConverterUtil.getCastSpecSparkSqlMapType(this, type,
+          SqlParserPos.ZERO);
     case MULTISET:
       throw new UnsupportedOperationException("Spark dialect does not support cast to "
           + type.getSqlTypeName());

core/src/main/java/org/apache/calcite/util/RelToSqlConverterUtil.java

@@ -390,6 +390,24 @@ public static SqlDataTypeSpec getCastSpecAngleBracketArrayType(SqlDialect dialec
     return new SqlDataTypeSpec(sqlArrayTypeNameSpec, SqlParserPos.ZERO);
   }
 
+  /**
+   * Transformation Map type from {@code MAP<VARCHAR,INTEGER>} to {@code Map<STRING,INTEGER>}.
+   */
+  public static SqlDataTypeSpec getCastSpecSparkSqlMapType(SqlDialect dialect,
+      RelDataType type, SqlParserPos pos) {
+    MapSqlType mapSqlType = (MapSqlType) type;
+    SqlDataTypeSpec keySpec = (SqlDataTypeSpec) dialect.getCastSpec(mapSqlType.getKeyType());
+    SqlDataTypeSpec valueSpec =
+        (SqlDataTypeSpec) dialect.getCastSpec(mapSqlType.getValueType());
+    SqlDataTypeSpec nonNullKeySpec =
+        requireNonNull(keySpec, "keySpec").withNullable(false);
+    SqlDataTypeSpec nonNullValueSpec =
+        requireNonNull(valueSpec, "valueSpec").withNullable(false);
+    SqlMapTypeNameSpec sqlMapTypeNameSpec =
+        new SqlMapTypeNameSpec(nonNullKeySpec, nonNullValueSpec, pos);
+    return new SqlDataTypeSpec(sqlMapTypeNameSpec, SqlParserPos.ZERO);
+  }
+
   /**
    * ClickHouseSqlMapTypeNameSpec to parse or unparse SQL MAP type to {@code Map(VARCHAR, VARCHAR)}.
    */

core/src/test/java/org/apache/calcite/rel/rel2sql/RelToSqlConverterTest.java

@@ -2853,6 +2853,44 @@ private SqlDialect nonOrdinalDialect() {
         .withSpark().ok(expectedSpark1);
   }
 
+  /** Test case for
+   * <a href="https://issues.apache.org/jira/browse/CALCITE-7433">[CALCITE-7433]
+   * Invalid unparse for cast to map type in Spark</a>.
+   */
+  @Test void testCastMapSpark() {
+    final String query = "select cast(MAP['a',1,'b',2,'c',3]"
+        + " as MAP<varchar,integer>)";
+    final String expectedSpark =
+        "SELECT CAST(MAP ('a', 1, 'b', 2, 'c', 3) AS MAP< STRING, INTEGER >)\n"
+            + "FROM (VALUES (0)) `t` (`ZERO`)";
+    sql(query)
+        .withSpark().ok(expectedSpark);
+
+    final String query1 = "select cast(MAP['a',ARRAY[1,2,3]]"
+        + " as MAP<varchar,integer array>)";
+    final String expectedSpark1 =
+        "SELECT CAST(MAP ('a', ARRAY (1, 2, 3)) AS MAP< STRING, ARRAY< INTEGER > >)\n"
+            + "FROM (VALUES (0)) `t` (`ZERO`)";
+    sql(query1)
+        .withSpark().ok(expectedSpark1);
+
+    final String query2 = "select cast(MAP['a',ARRAY[1.0,2.0,3.0]]"
+        + " as MAP<varchar,real array>)";
+    final String expectedSpark2 =
+        "SELECT CAST(MAP ('a', ARRAY (1.0, 2.0, 3.0)) AS MAP< STRING, ARRAY< REAL > >)\n"
+            + "FROM (VALUES (0)) `t` (`ZERO`)";
+    sql(query2)
+        .withSpark().ok(expectedSpark2);
+
+    final String query3 = "select cast(MAP['a',MAP['b','c']]"
+        + " as MAP<varchar,MAP<varchar,varchar>>)";
+    final String expectedSpark3 =
+        "SELECT CAST(MAP ('a', MAP ('b', 'c')) AS MAP< STRING, MAP< STRING, STRING > >)\n"
+            + "FROM (VALUES (0)) `t` (`ZERO`)";
+    sql(query3)
+        .withSpark().ok(expectedSpark3);
+  }
+
   /** Test case for
    * <a href="https://issues.apache.org/jira/browse/CALCITE-7055">[CALCITE-7055]
    * Invalid unparse for cast to array type in StarRocks</a>.