feat(pyamber): hash large binary tuples

amber/src/main/python/core/models/tuple.py

@@ -172,6 +172,11 @@ def java_hash_bytes(bytes: Iterator[int], init: int, salt: int):
     return h
 
 
+def java_hash_string(value: str, salt: int):
+    units = (unit[0] for unit in struct.iter_unpack(">H", value.encode("utf-16-be")))
+    return java_hash_bytes(units, 0, salt)
+
+
 class Tuple:
     """
     Lazy-Tuple implementation.
@@ -474,6 +479,7 @@ def __hash__(self) -> int:
             AttributeType.STRING: lambda f: java_hash_bytes(map(ord, f), 0, salt),
             AttributeType.TIMESTAMP: lambda f: java_hash_long(int(f.timestamp())),
             AttributeType.BINARY: lambda f: java_hash_bytes(f, 1, salt),
+            AttributeType.LARGE_BINARY: lambda f: salt + java_hash_string(f.uri, salt),
         }
 
         for name, field in self.as_key_value_pairs():

amber/src/test/python/core/architecture/sendsemantics/test_partitioners.py

@@ -275,6 +275,27 @@ def test_full_batch_yields_and_clears_only_that_slot(self):
         for _, batch in p.receivers:
             assert len(batch) < p.batch_size
 
+    @pytest.mark.parametrize("uri", ["s3://bucket/object", "s3://bucket/\U0001f600"])
+    def test_large_binary_key_routes_to_same_receiver(self, uri):
+        from core.models.type.large_binary import largebinary
+
+        schema = Schema(raw_schema={"k": "LARGE_BINARY", "v": "STRING"})
+        p = self._partitioner(batch_size=1)
+        first = Tuple({"k": largebinary(uri), "v": "first"}, schema)
+        second = Tuple({"k": largebinary(uri), "v": "second"}, schema)
+
+        first_route = [
+            (receiver, list(batch)) for receiver, batch in p.add_tuple_to_batch(first)
+        ]
+        second_route = [
+            (receiver, list(batch)) for receiver, batch in p.add_tuple_to_batch(second)
+        ]
+
+        assert len(first_route) == len(second_route) == 1
+        assert first_route[0][0] == second_route[0][0]
+        assert first_route[0][1] == [first]
+        assert second_route[0][1] == [second]
+
     def test_no_hash_attribute_names_falls_back_to_whole_tuple(self):
         p = self._partitioner(hash_keys=())
         list(p.add_tuple_to_batch(_hashable_tuple(k=1, v="a")))

amber/src/test/python/core/models/test_tuple.py

@@ -476,6 +476,23 @@ def test_finalize_rejects_numpy_bool_false_in_int_field(self):
         with pytest.raises(TypeError, match="Unmatched type"):
             tuple_.finalize(Schema(raw_schema={"count": "INTEGER"}))
 
+    @pytest.mark.parametrize(
+        ("uri", "java_hash"),
+        [
+            ("s3://bucket/object", -46745592),
+            ("s3://bucket/\U0001f600", 1443831724),
+            ("s3://bucket/key", 1754254834),
+        ],
+    )
+    def test_large_binary_hash_matches_java(self, uri, java_hash):
+        from core.models.type.large_binary import largebinary
+
+        schema = Schema(raw_schema={"blob": "LARGE_BINARY"})
+        blob = Tuple({"blob": largebinary(uri)}, schema)
+
+        assert hash(blob) == java_hash
+        assert hash(Tuple({"blob": None}, schema)) == 31
+
     def test_hash(self):
         schema = Schema(
             raw_schema={