[Improve][Transform] Improve inner sql query for map field type (#7718)

Author: liunaijie

seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/inner_query.conf

@@ -53,6 +53,7 @@ transform {
         c_row.c_string c_string,
         c_row.c_inner_row.c_inner_string c_inner_string,
         c_row.c_inner_row.c_inner_timestamp c_inner_timestamp,
+        c_row.c_inner_row.c_map inner_map,
         c_row.c_inner_row.c_map.innerQuery map_val,
         c_row.c_inner_row.c_map.notExistKey map_not_exist_val
         from fake"""
@@ -94,11 +95,20 @@ sink {
             {rule_type = NOT_NULL}
           ]
         },
+        {
+          field_name = "inner_map"
+          field_type = "map<string, string>"
+          field_value = [
+            {
+              equals_to = {innerQuery=innerQuery}
+            }
+          ]
+        },
         {
           field_name = "map_val"
           field_type = "string"
           field_value = [
-            {rule_type = NOT_NULL}
+            {equals_to = "innerQuery"}
           ]
         },
         {

seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java

@@ -144,12 +144,15 @@ public SeaTunnelDataType<?> getExpressionType(Expression expression) {
                     if (filedTypeRes instanceof SeaTunnelRowType) {
                         parRowType = (SeaTunnelRowType) filedTypeRes;
                     } else if (filedTypeRes instanceof MapType) {
-                        //  for map type. only support it's the latest struct.
-                        if (i != deep - 2) {
+                        if (i < deep - 2) {
                             throw new IllegalArgumentException(
-                                    "For now, we only support map struct is the latest struct in inner query function! Please modify your query!");
+                                    "For now, when you query map field with inner query, it must be latest field or latest struct field! Please modify your query!");
+                        }
+                        if (i == deep - 1) {
+                            return filedTypeRes;
+                        } else {
+                            return ((MapType<?, ?>) filedTypeRes).getValueType();
                         }
-                        return ((MapType<?, ?>) filedTypeRes).getValueType();
                     }
                 }
                 return filedTypeRes;