[format][python] Support binary keys for MAP BLOB

docs/docs/concepts/spec/fileformat.md

@@ -896,11 +896,14 @@ their encodings are:
 | `DECIMAL(p, s)`, `p > 18` | Minimal-length signed big-endian two's-complement unscaled integer |
 | `DATE` | Four-byte little-endian signed count of days since 1970-01-01 |
 | `TIME(p)` | Four-byte little-endian signed count of milliseconds since midnight |
+| `BINARY`, `VARBINARY` (`BYTES`) | Raw bytes |
 | `CHAR`, `VARCHAR` | UTF-8 bytes |
 
-The DECIMAL scale is defined by the field type and is not stored in each key. An empty
-map has an entry count of zero and is distinct from a null map. The `TIME(p)` encoding
-uses Paimon's millisecond internal representation and does not add nanosecond precision.
+The DECIMAL scale is defined by the field type and is not stored in each key. `BINARY`
+and `VARBINARY` keys are not padded, truncated, or validated against the declared length.
+An empty map has an entry count of zero and is distinct from a null map. The `TIME(p)`
+encoding uses Paimon's millisecond internal representation and does not add nanosecond
+precision.
 
 At the outer file index level, `-1` represents a null field and `-2` represents a
 field placeholder used by data evolution.

docs/docs/multimodal-table/blob.mdx

@@ -87,8 +87,9 @@ Paimon supports three storage modes for BLOB fields, selected via **comment dire
 This allows one table to mix different storage modes for different BLOB columns.
 `ARRAY<BLOB>` and `MAP<K, BLOB>` are supported only by `__BLOB_FIELD`;
 descriptor-only and blob-view comment directives accept scalar BLOB fields only.
-Map keys support the integer family, `BOOLEAN`, `DECIMAL`, `DATE`, `TIME`, `CHAR`, and
-`VARCHAR`. Use non-null keys for compatibility across Flink, Spark, and Python.
+Map keys support the integer family, `BOOLEAN`, `DECIMAL`, `DATE`, `TIME`, `BINARY`,
+`VARBINARY` (`BYTES`), `CHAR`, and `VARCHAR`. Use non-null keys for compatibility across
+Flink, Spark, and Python.
 
 ## Table Options
 

docs/docs/primary-key-table/blob-storage.md

@@ -94,8 +94,8 @@ array order, a null array, and null elements are preserved. An empty array write
 
 `MAP<K, BLOB>` is externalized value by value. Keys remain in the normal data file and every non-null value is replaced
 with a descriptor to managed storage. A null map, an empty map, and null values are preserved. Supported key types are
-the integer family, `BOOLEAN`, `DECIMAL`, `DATE`, `TIME`, `CHAR`, and `VARCHAR`; `blob-descriptor-field` and
-`blob-view-field` remain scalar-only declarations.
+the integer family, `BOOLEAN`, `DECIMAL`, `DATE`, `TIME`, `BINARY`, `VARBINARY` (`BYTES`), `CHAR`, and `VARCHAR`;
+`blob-descriptor-field` and `blob-view-field` remain scalar-only declarations.
 
 `blob.target-file-size` controls when a writer rolls to a new managed payload pack. A pack can contain payloads from
 multiple rows, and a row descriptor records its URI, offset, and length.

paimon-common/src/main/java/org/apache/paimon/data/GenericMap.java

@@ -23,6 +23,10 @@
 import org.apache.paimon.types.MultisetType;
 
 import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
@@ -46,25 +50,58 @@ public final class GenericMap implements InternalMap, Serializable {
     private static final long serialVersionUID = 1L;
 
     private final Map<?, ?> map;
+    private final boolean binaryKeys;
 
     /**
      * Creates an instance of {@link GenericMap} using the given Java map.
      *
      * <p>Note: All keys and values of the map must be internal data structures.
      */
     public GenericMap(Map<?, ?> map) {
-        this.map = map;
+        this(map, false);
+    }
+
+    private GenericMap(Map<?, ?> map, boolean binaryKeys) {
+        this.binaryKeys = binaryKeys;
+        this.map = binaryKeys ? normalizeBinaryKeys(map) : map;
+    }
+
+    /**
+     * Creates a map whose binary keys use content equality.
+     *
+     * @since 2.1
+     */
+    public static GenericMap fromBinaryKeyMap(Map<?, ?> map) {
+        return new GenericMap(map, true);
+    }
+
+    private static Map<BinaryKey, Object> normalizeBinaryKeys(Map<?, ?> map) {
+        Map<BinaryKey, Object> binaryMap = new LinkedHashMap<>();
+        for (Map.Entry<?, ?> entry : map.entrySet()) {
+            Object key = entry.getKey();
+            if (key != null && !(key instanceof byte[])) {
+                throw new IllegalArgumentException("Binary key must be byte[].");
+            }
+            binaryMap.put(copyBinaryKey(key), entry.getValue());
+        }
+        return binaryMap;
     }
 
     /**
      * Returns the value to which the specified key is mapped, or {@code null} if this map contains
      * no mapping for the key. The returned value is in internal data structure.
      */
     public Object get(Object key) {
+        if (binaryKeys) {
+            return isBinaryKey(key) ? map.get(lookupBinaryKey(key)) : null;
+        }
         return map.get(key);
     }
 
     public boolean contains(Object key) {
+        if (binaryKeys) {
+            return isBinaryKey(key) && map.containsKey(lookupBinaryKey(key));
+        }
         return map.containsKey(key);
     }
 
@@ -75,7 +112,11 @@ public int size() {
 
     @Override
     public InternalArray keyArray() {
-        Object[] keys = map.keySet().toArray();
+        Object[] keys = new Object[map.size()];
+        int index = 0;
+        for (Object key : map.keySet()) {
+            keys[index++] = copyUnwrappedBinaryKey(key);
+        }
         return new GenericArray(keys);
     }
 
@@ -94,18 +135,50 @@ public boolean equals(Object o) {
             return false;
         }
         // deepEquals for values of byte[]
-        return deepEquals(map, ((GenericMap) o).map);
+        return deepEquals(this, (GenericMap) o);
+    }
+
+    private static boolean deepEquals(GenericMap m1, GenericMap m2) {
+        if (m1.map.size() != m2.map.size()) {
+            return false;
+        }
+        if ((m1.binaryKeys && m2.binaryKeys) || (!m1.hasBinaryKeys() && !m2.hasBinaryKeys())) {
+            return deepEquals(m1.map, m2.map);
+        }
+
+        List<Map.Entry<?, ?>> entries2 = new ArrayList<>(m2.map.entrySet());
+        boolean[] matched = new boolean[entries2.size()];
+        for (Map.Entry<?, ?> entry1 : m1.map.entrySet()) {
+            boolean found = false;
+            for (int i = 0; i < entries2.size(); i++) {
+                if (matched[i]) {
+                    continue;
+                }
+                Map.Entry<?, ?> entry2 = entries2.get(i);
+                if (Objects.deepEquals(
+                                unwrapBinaryKey(entry1.getKey()), unwrapBinaryKey(entry2.getKey()))
+                        && Objects.deepEquals(entry1.getValue(), entry2.getValue())) {
+                    matched[i] = true;
+                    found = true;
+                    break;
+                }
+            }
+            if (!found) {
+                return false;
+            }
+        }
+        return true;
     }
 
-    private static <K, V> boolean deepEquals(Map<K, V> m1, Map<?, ?> m2) {
+    private static boolean deepEquals(Map<?, ?> m1, Map<?, ?> m2) {
         // copied from HashMap.equals but with deepEquals comparison
         if (m1.size() != m2.size()) {
             return false;
         }
         try {
-            for (Map.Entry<K, V> e : m1.entrySet()) {
-                K key = e.getKey();
-                V value = e.getValue();
+            for (Map.Entry<?, ?> entry : m1.entrySet()) {
+                Object key = entry.getKey();
+                Object value = entry.getValue();
                 if (value == null) {
                     if (!(m2.get(key) == null && m2.containsKey(key))) {
                         return false;
@@ -126,9 +199,76 @@ private static <K, V> boolean deepEquals(Map<K, V> m1, Map<?, ?> m2) {
     public int hashCode() {
         int result = 0;
         for (Object key : map.keySet()) {
+            key = unwrapBinaryKey(key);
             // only include key because values can contain byte[]
-            result += 31 * Objects.hashCode(key);
+            result +=
+                    31
+                            * (key instanceof byte[]
+                                    ? Arrays.hashCode((byte[]) key)
+                                    : Objects.hashCode(key));
         }
         return result;
     }
+
+    private boolean hasBinaryKeys() {
+        return binaryKeys || hasBinaryKey(map);
+    }
+
+    private static Object unwrapBinaryKey(Object key) {
+        return key instanceof BinaryKey ? ((BinaryKey) key).bytes : key;
+    }
+
+    private static Object copyUnwrappedBinaryKey(Object key) {
+        return key instanceof BinaryKey ? ((BinaryKey) key).copyBytes() : key;
+    }
+
+    private static boolean isBinaryKey(Object key) {
+        return key == null || key instanceof byte[];
+    }
+
+    private static BinaryKey copyBinaryKey(Object key) {
+        return key == null ? null : new BinaryKey((byte[]) key, true);
+    }
+
+    private static BinaryKey lookupBinaryKey(Object key) {
+        return key == null ? null : new BinaryKey((byte[]) key, false);
+    }
+
+    private static boolean hasBinaryKey(Map<?, ?> map) {
+        for (Object key : map.keySet()) {
+            if (key instanceof byte[]) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    private static final class BinaryKey implements Serializable {
+
+        private static final long serialVersionUID = 1L;
+
+        private final byte[] bytes;
+        private final int hash;
+
+        private BinaryKey(byte[] bytes, boolean copy) {
+            this.bytes = copy ? Arrays.copyOf(bytes, bytes.length) : bytes;
+            this.hash = Arrays.hashCode(this.bytes);
+        }
+
+        private byte[] copyBytes() {
+            return Arrays.copyOf(bytes, bytes.length);
+        }
+
+        @Override
+        public boolean equals(Object object) {
+            return object == this
+                    || (object instanceof BinaryKey
+                            && Arrays.equals(bytes, ((BinaryKey) object).bytes));
+        }
+
+        @Override
+        public int hashCode() {
+            return hash;
+        }
+    }
 }

paimon-common/src/main/java/org/apache/paimon/data/serializer/InternalMapSerializer.java

@@ -100,6 +100,8 @@ public InternalMap copy(InternalMap from) {
     }
 
     private GenericMap copyBlobMap(InternalMap map) {
+        DataTypeRoot keyRoot = keyType.getTypeRoot();
+        boolean binaryKey = keyRoot == DataTypeRoot.BINARY || keyRoot == DataTypeRoot.VARBINARY;
         Map<Object, Object> copied = new LinkedHashMap<>();
         InternalArray keys = map.keyArray();
         InternalArray values = map.valueArray();
@@ -110,7 +112,7 @@ private GenericMap copyBlobMap(InternalMap map) {
                     key == null ? null : keySerializer.copy(key),
                     value == null ? null : valueSerializer.copy(value));
         }
-        return new GenericMap(copied);
+        return binaryKey ? GenericMap.fromBinaryKeyMap(copied) : new GenericMap(copied);
     }
 
     @Override