(seatunnel) branch dev updated: [BugFix][Spark-translation] map type cast error (#6552)

[fanjia]

This is an automated email from the ASF dual-hosted git repository.

fanjia pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git


The following commit(s) were added to refs/heads/dev by this push:
     new b1dcd4a2bc [BugFix][Spark-translation] map type cast error (#6552)
b1dcd4a2bc is described below

commit b1dcd4a2bc9292f1c46571d342195e53078f7942
Author: Carl-Zhou-CN <1058249...@qq.com>
AuthorDate: Thu Mar 21 13:27:33 2024 +0800

    [BugFix][Spark-translation] map type cast error (#6552)
---
 .../src/test/resources/fake_to_assert.conf           |  9 +++++++++
 .../spark/serialization/SeaTunnelRowConverter.java   | 20 ++++++++++----------
 2 files changed, 19 insertions(+), 10 deletions(-)

diff --git 
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fake-e2e/src/test/resources/fake_to_assert.conf
 
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fake-e2e/src/test/resources/fake_to_assert.conf
index 084ddbb533..cab504d3b8 100644
--- 
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fake-e2e/src/test/resources/fake_to_assert.conf
+++ 
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-fake-e2e/src/test/resources/fake_to_assert.conf
@@ -67,8 +67,17 @@ source {
   }
 }
 
+transform {
+    Sql {
+        source_table_name = "fake"
+        result_table_name = "tmp1"
+        query = """select * from fake"""
+    }
+}
+
 sink {
   Assert {
+    source_table_name = "tmp1"
     rules {
       row_rules = [
         {
diff --git 
a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/SeaTunnelRowConverter.java
 
b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/SeaTunnelRowConverter.java
index 000e0baa06..db23d4912a 100644
--- 
a/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/SeaTunnelRowConverter.java
+++ 
b/seatunnel-translation/seatunnel-translation-spark/seatunnel-translation-spark-common/src/main/java/org/apache/seatunnel/translation/spark/serialization/SeaTunnelRowConverter.java
@@ -31,7 +31,7 @@ import org.apache.spark.sql.types.StructType;
 import org.apache.spark.unsafe.types.UTF8String;
 
 import scala.Tuple2;
-import scala.collection.immutable.HashMap.HashTrieMap;
+import scala.collection.immutable.AbstractMap;
 import scala.collection.mutable.WrappedArray;
 
 import java.io.IOException;
@@ -179,7 +179,7 @@ public class SeaTunnelRowConverter extends 
RowConverter<SeaTunnelRow> {
             case STRING:
                 return field.toString();
             case MAP:
-                return reconvertMap((HashTrieMap<?, ?>) field, (MapType<?, ?>) 
dataType);
+                return reconvertMap((AbstractMap<?, ?>) field, (MapType<?, ?>) 
dataType);
             case ARRAY:
                 return reconvertArray((WrappedArray.ofRef<?>) field, 
(ArrayType<?, ?>) dataType);
             default:
@@ -206,23 +206,23 @@ public class SeaTunnelRowConverter extends 
RowConverter<SeaTunnelRow> {
     }
 
     /**
-     * Convert HashTrieMap to LinkedHashMap
+     * Convert AbstractMap to LinkedHashMap
      *
-     * @param hashTrieMap HashTrieMap data
+     * @param abstractMap AbstractMap data
      * @param mapType fields type map
      * @return java.util.LinkedHashMap
-     * @see HashTrieMap
+     * @see AbstractMap
      */
-    private Map<Object, Object> reconvertMap(HashTrieMap<?, ?> hashTrieMap, 
MapType<?, ?> mapType) {
-        if (hashTrieMap == null || hashTrieMap.size() == 0) {
+    private Map<Object, Object> reconvertMap(AbstractMap<?, ?> abstractMap, 
MapType<?, ?> mapType) {
+        if (abstractMap == null || abstractMap.size() == 0) {
             return Collections.emptyMap();
         }
-        int num = hashTrieMap.size();
+        int num = abstractMap.size();
         Map<Object, Object> newMap = new LinkedHashMap<>(num);
         SeaTunnelDataType<?> keyType = mapType.getKeyType();
         SeaTunnelDataType<?> valueType = mapType.getValueType();
-        scala.collection.immutable.List<?> keyList = 
hashTrieMap.keySet().toList();
-        scala.collection.immutable.List<?> valueList = 
hashTrieMap.values().toList();
+        scala.collection.immutable.List<?> keyList = 
abstractMap.keySet().toList();
+        scala.collection.immutable.List<?> valueList = 
abstractMap.values().toList();
         for (int i = 0; i < num; i++) {
             Object key = keyList.apply(i);
             Object value = valueList.apply(i);