diff --git a/amber/src/main/python/core/models/tuple.py b/amber/src/main/python/core/models/tuple.py index 1f319b6a4f4..e7f0614336f 100644 --- a/amber/src/main/python/core/models/tuple.py +++ b/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(): diff --git a/amber/src/test/python/core/architecture/sendsemantics/test_partitioners.py b/amber/src/test/python/core/architecture/sendsemantics/test_partitioners.py index 7d70eb37de1..57cff689406 100644 --- a/amber/src/test/python/core/architecture/sendsemantics/test_partitioners.py +++ b/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"))) diff --git a/amber/src/test/python/core/models/test_tuple.py b/amber/src/test/python/core/models/test_tuple.py index aff43faa881..730f1ff266a 100644 --- a/amber/src/test/python/core/models/test_tuple.py +++ b/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={