[fix](array-type) forbid implicit cast of array type while load (#15325)

* forbit array cast while load

* add regression test

Co-authored-by: cambyzju <zhuxiaoli01@baidu.com>

fe/fe-core/src/main/java/org/apache/doris/load/Load.java

@@ -1061,6 +1061,17 @@ public class Load {
                     throw new AnalysisException("Don't support aggregation function in load expression");
                 }
             }
+
+            // Array type do not support cast now
+            Type exprReturnType = expr.getType();
+            if (exprReturnType.isArrayType()) {
+                Type schemaType = tbl.getColumn(entry.getKey()).getType();
+                if (exprReturnType != schemaType) {
+                    throw new AnalysisException("Don't support load from type:" + exprReturnType + " to type:"
+                            + schemaType + " for column:" + entry.getKey());
+                }
+            }
+
             exprsByName.put(entry.getKey(), expr);
         }
 

regression-test/suites/load_p0/stream_load/test_stream_load.groovy

@@ -650,5 +650,27 @@ suite("test_stream_load", "p0") {
     order_qt_all102 "SELECT * from ${tableName8}" // 8
     sql """truncate table ${tableName8}"""
     sql """sync"""
+
+    // malformat with mismatch array type
+    streamLoad {
+        table "${tableName8}"
+
+        set 'column_separator', '|'
+        set 'columns', 'k1,k2,k3,k4,k5,k6,k7,k8,k9,b10,k11,k10=array_remove(cast(k5 as array<bigint>), 1)'
+
+        file 'array_normal.csv'
+        time 10000 // limit inflight 10s
+
+        check { result, exception, startTime, endTime ->
+            if (exception != null) {
+                throw exception
+            }
+            log.info("Stream load result: ${result}".toString())
+            def json = parseJson(result)
+            assertEquals("fail", json.Status.toLowerCase())
+            assertTrue(json.Message.contains('Don\'t support load from type'))
+        }
+    }
+    sql "sync"
 }