(fluss) branch main updated: [gateway] Reject duplicate MAP keys in entry arrays (#4205)

[ab...@apache.org]

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

fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git


The following commit(s) were added to refs/heads/main by this push:
     new 1e541870f5 [gateway] Reject duplicate MAP keys in entry arrays (#4205)
1e541870f5 is described below

commit 1e541870f5fd6e9f9e84546064d93f52be9d2a34
Author: Gezi-lzq <lz...@gmail.com>
AuthorDate: Mon Sep 7 06:45:42 2026 +0800

    [gateway] Reject duplicate MAP keys in entry arrays (#4205)
    
    Reject duplicate decoded keys before encoding canonical MAP entry arrays so later values cannot silently overwrite earlier entries.
---
 fluss-gateway/src/protocol/rest/codec/decoder.rs | 36 ++++++++++++++++++++++++
 1 file changed, 36 insertions(+)

diff --git a/fluss-gateway/src/protocol/rest/codec/decoder.rs b/fluss-gateway/src/protocol/rest/codec/decoder.rs
index d4d6a43c57..9605936ba5 100644
--- a/fluss-gateway/src/protocol/rest/codec/decoder.rs
+++ b/fluss-gateway/src/protocol/rest/codec/decoder.rs
@@ -1272,6 +1272,15 @@ fn finish_map(
     value_type: &DataType,
     entries: Vec<(Datum<'static>, Datum<'static>)>,
 ) -> DecodeResult<Datum<'static>> {
+    let mut seen = HashSet::with_capacity(entries.len());
+    for (index, (key, _)) in entries.iter().enumerate() {
+        if !seen.insert(key) {
+            return Err(path
+                .map_part(index, "key")
+                .invalid("duplicates an earlier MAP key"));
+        }
+    }
+
     let mut writer = FlussMapWriter::new(entries.len(), key_type, value_type);
     for (key, value) in entries {
         if let Err(error) = writer.write_entry(key, value) {
@@ -1975,6 +1984,33 @@ mod tests {
         assert!(duplicate.message().contains("duplicate key `a`"));
     }
 
+    #[test]
+    fn rejects_duplicate_keys_in_map_entry_arrays() {
+        for (map_type, json, duplicate_path) in [
+            (
+                DataType::Map(MapType::new(
+                    DataType::String(StringType::new()),
+                    DataType::Int(IntType::new()),
+                )),
+                r#"[{"key":"role","value":1},{"key":"region","value":2},{"key":"role","value":3}]"#,
+                "v[2].key",
+            ),
+            (
+                DataType::Map(MapType::new(
+                    DataType::Int(IntType::new()),
+                    DataType::String(StringType::new()),
+                )),
+                r#"[{"key":7,"value":"first"},{"key":"7","value":"second"}]"#,
+                "v[1].key",
+            ),
+        ] {
+            let error = decode_one(map_type, json).unwrap_err();
+            assert!(!error.is_schema_mismatch(), "{json}");
+            assert!(error.message().contains(duplicate_path), "{json}");
+            assert!(error.message().contains("duplicate"), "{json}");
+        }
+    }
+
     #[test]
     fn keeps_object_shorthand_limited_to_string_map_keys() {
         let map_type = DataType::Map(MapType::new(