[FLINK-38286][table] Fix: MAP function with duplicate keys produces non-deterministic results

Author: raminqaf

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/GenerateUtils.scala

@@ -802,4 +802,19 @@ object GenerateUtils {
     compares.mkString
   }
 
+  /**
+   * Groups the input sequence by the key function, and keeps the order of the first appearance of
+   * each key.
+   */
+  def groupByOrdered[A, K](xs: collection.Seq[A])(
+      f: A => K): collection.Seq[(K, collection.Seq[A])] = {
+    val m = collection.mutable.LinkedHashMap.empty[K, collection.mutable.ArrayBuffer[A]]
+    xs.foreach {
+      x =>
+        val k = f(x)
+        m.getOrElseUpdate(k, collection.mutable.ArrayBuffer.empty[A]) += x
+    }
+    m.toSeq.map { case (k, buffer) => (k, buffer) }
+  }
+
 }

flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala

@@ -1510,28 +1510,23 @@ object ScalarOperatorGens {
     val mapType = resultType.asInstanceOf[MapType]
     val baseMap = newName(ctx, "map")
 
-    // prepare map key array
-    val keyElements = elements
+    val keyValues = elements
       .grouped(2)
       .map { case Seq(key, value) => (key, value) }
       .toSeq
-      .groupBy(_._1)
-      .map(_._2.last)
-      .keys
-      .toSeq
+
+    val deduplicatedKeyValues = groupByOrdered(keyValues)(_._1).map {
+      case (_, pairs) =>
+        pairs.last // Take the last occurrence
+    }
+    // prepare map key array
+    val keyElements = deduplicatedKeyValues.map(_._1)
     val keyType = mapType.getKeyType
     val keyExpr = generateArray(ctx, new ArrayType(keyType), keyElements)
     val isKeyFixLength = isPrimitive(keyType)
 
     // prepare map value array
-    val valueElements = elements
-      .grouped(2)
-      .map { case Seq(key, value) => (key, value) }
-      .toSeq
-      .groupBy(_._1)
-      .map(_._2.last)
-      .values
-      .toSeq
+    val valueElements = deduplicatedKeyValues.map(_._2)
     val valueType = mapType.getValueType
     val valueExpr = generateArray(ctx, new ArrayType(valueType), valueElements)
     val isValueFixLength = isPrimitive(valueType)

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/functions/MapFunctionITCase.java

@@ -97,9 +97,16 @@ private Stream<TestSetSpec> mapTestCases() {
                                 DECIMAL(10, 4).notNull(),
                                 BOOLEAN().notNull())
                         .testResult(
+                                resultSpec(
+                                        map(
+                                                1, 1, 1, 2, 1, 9, 1, 3, 2, 24, 2, 29, 1, 0, 2, 22,
+                                                1, 8, 2, 25, 2, 20),
+                                        "MAP[1, 1, 1, 2, 1, 9, 1, 3, 2, 24, 2, 29, 1, 0, 2, 22, 1, 8, 2, 25, 2, 20]",
+                                        Map.ofEntries(Map.entry(1, 8), Map.entry(2, 20)),
+                                        DataTypes.MAP(INT().notNull(), INT().notNull()).notNull()),
                                 resultSpec(
                                         map($("f0"), $("f0"), $("f0"), $("f1")),
-                                        "MAP[f0, f1]",
+                                        "MAP[f0, f0, f0, f1]",
                                         Collections.singletonMap(1, 2),
                                         DataTypes.MAP(INT().notNull(), INT().notNull()).notNull()),
                                 resultSpec(