This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 3c408c0ab5 [python] Introduce hll_sketch aggregator function (#9274)
3c408c0ab5 is described below
commit 3c408c0ab510a7c0ae7879e7792de1a486922db4
Author: Xiangyi Zhu <[email protected]>
AuthorDate: Thu Aug 20 10:03:17 2026 +0800
[python] Introduce hll_sketch aggregator function (#9274)
---
.../pypaimon/read/merge_engine_support.py | 5 +-
.../pypaimon/read/reader/aggregate/aggregators.py | 76 +++++++++++++++++--
.../read/reader/aggregation_merge_function.py | 4 +-
.../pypaimon/tests/test_aggregation_e2e.py | 15 ++--
.../pypaimon/tests/test_field_aggregators.py | 88 +++++++++++++++++++++-
.../tests/test_optional_datasketches_dependency.py | 10 +++
paimon-python/setup.py | 4 +
7 files changed, 181 insertions(+), 21 deletions(-)
diff --git a/paimon-python/pypaimon/read/merge_engine_support.py
b/paimon-python/pypaimon/read/merge_engine_support.py
index daf11b6a2f..a7d8337854 100644
--- a/paimon-python/pypaimon/read/merge_engine_support.py
+++ b/paimon-python/pypaimon/read/merge_engine_support.py
@@ -69,6 +69,7 @@ _AGGREGATION_SUPPORTED_AGG_FUNCS = frozenset([
"merge_map_with_keytime",
"merge_map",
"theta_sketch",
+ "hll_sketch",
"rbm32",
])
_FIELDS_PREFIX = "fields."
@@ -215,7 +216,7 @@ def check_supported(table) -> None:
"built-in aggregators ({}); retract opt-ins "
"(aggregation.remove-record-on-delete, "
"fields.<f>.ignore-retract) "
- "and other aggregators (hll_sketch / rbm64) "
+ "and other aggregators (rbm64) "
"are not yet supported. "
"Open an issue to track support.".format(
", ".join(sorted(unsupported)),
@@ -266,7 +267,7 @@ def aggregation_unsupported_options(table) -> Set[str]:
``builtin_seq_comparator``).
3. Out-of-scope aggregator selections: ``fields.<f>.aggregate-
function`` and ``fields.default-aggregate-function`` set to an
- identifier this engine doesn't support yet (e.g. ``hll_sketch``).
+ identifier this engine doesn't support yet (e.g. ``rbm64``).
"""
flagged: Set[str] = set()
raw = table.options.options.to_map()
diff --git a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
index a00ccffef7..7d3ff90256 100644
--- a/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
+++ b/paimon-python/pypaimon/read/reader/aggregate/aggregators.py
@@ -22,16 +22,18 @@ Each class registers itself with the global registry at
import time
via :func:`register_aggregator`, so importing
``pypaimon.read.reader.aggregate`` makes all of them discoverable.
-This module ships 19 aggregators — the primary-key placeholder plus
-the 18 most commonly-used value aggregators: ``primary_key`` /
+This module ships 20 aggregators — the primary-key placeholder plus
+the 19 most commonly-used value aggregators: ``primary_key`` /
``last_value`` / ``last_non_null_value`` / ``first_value`` /
``first_non_null_value`` / ``sum`` / ``max`` / ``min`` / ``bool_or``
/ ``bool_and`` / ``product`` / ``listagg`` / ``collect`` /
``merge_map`` / ``merge_map_with_keytime`` / ``nested_update`` /
-``nested_partial_update`` / ``theta_sketch`` / ``rbm32``. Other
-aggregators (``hll_sketch`` / ``rbm64``) are intentionally deferred —
-the registry will report them as unsupported so users see a clear
-error rather than a silent fallback.
+``nested_partial_update`` / ``theta_sketch`` / ``hll_sketch`` /
+``rbm32``. The remaining aggregator (``rbm64``) is intentionally
+deferred — Java serializes it with ``Roaring64Bitmap``'s private ART
+format, which has no portable counterpart in ``pyroaring``, so the
+bytes are not interchangeable. The registry reports it as unsupported
+so users see a clear error rather than a silent fallback.
"""
from typing import Any, List, Dict, Optional, Tuple, Union, Set
@@ -69,6 +71,7 @@ NAME_COLLECT = "collect"
NAME_MERGE_MAP_WITH_KEYTIME = "merge_map_with_keytime"
NAME_MERGE_MAP = "merge_map"
NAME_THETA_SKETCH = "theta_sketch"
+NAME_HLL_SKETCH = "hll_sketch"
NAME_RBM32 = "rbm32"
@@ -1432,6 +1435,64 @@ class FieldThetaSketchAgg(FieldAggregator):
return union.get_result().serialize()
+class FieldHllSketchAgg(FieldAggregator):
+ """Aggregator for HyperLogLog sketches.
+
+ Mirrors Java's ``FieldHllSketchAgg`` / ``HllSketchUtil.union``: the
+ union is seeded from ``input_field`` and then updated with
+ ``accumulator``, and the result is emitted as a compact ``HLL_4``
+ sketch. Both sides use Apache DataSketches, so the serialized bytes
+ are interchangeable with Java's ``HllSketch.toCompactByteArray()``.
+ """
+
+ def __init__(self, name: str, field_type: DataType):
+ super().__init__(name, field_type)
+ if _atomic_base_name(field_type) not in ("VARBINARY", "BYTES"):
+ raise ValueError(
+ "Data type for hll sketch column must be 'VarBinaryType' but
was '{}'.".format(field_type)
+ )
+
+ def agg(self, accumulator: Any, input_field: Any) -> Any:
+ if accumulator is None or input_field is None:
+ return input_field if accumulator is None else accumulator
+
+ if not isinstance(accumulator, (bytes, bytearray)):
+ raise TypeError(
+ "HllSketch accumulator must be bytes, got
{}".format(type(accumulator))
+ )
+
+ if not isinstance(input_field, (bytes, bytearray)):
+ raise TypeError(
+ "HllSketch input must be bytes, got
{}".format(type(input_field))
+ )
+
+ if isinstance(accumulator, bytearray):
+ accumulator = bytes(accumulator)
+ if isinstance(input_field, bytearray):
+ input_field = bytes(input_field)
+
+ try:
+ from _datasketches import hll_sketch, hll_union, tgt_hll_type
+ except ImportError as exc:
+ raise ImportError(
+ "The hll_sketch aggregator requires the 'datasketches' "
+ "package. Install it with "
+ "\"pip install 'pypaimon[hll-sketch]'\"."
+ ) from exc
+
+ sketch1 = hll_sketch.deserialize(accumulator)
+ sketch2 = hll_sketch.deserialize(input_field)
+
+ # Java builds the union from the input sketch (``Union.heapify``)
+ # and folds the accumulator in afterwards; keep that order so the
+ # union's lg_k is taken from the same side as Java's.
+ union = hll_union(sketch2.lg_config_k)
+ union.update(sketch2)
+ union.update(sketch1)
+
+ return bytes(union.get_result(tgt_hll_type.HLL_4).serialize_compact())
+
+
class FieldRoaringBitmap32Agg(FieldAggregator):
"""roaring bitmap 32 aggregate a field of a row."""
@@ -1546,6 +1607,9 @@ register_aggregator(
register_aggregator(
NAME_THETA_SKETCH, _build_no_type_check(FieldThetaSketchAgg,
NAME_THETA_SKETCH)
)
+register_aggregator(
+ NAME_HLL_SKETCH, _build_no_type_check(FieldHllSketchAgg, NAME_HLL_SKETCH)
+)
register_aggregator(
NAME_RBM32, _build_roaring_bitmap(FieldRoaringBitmap32Agg, NAME_RBM32)
)
diff --git a/paimon-python/pypaimon/read/reader/aggregation_merge_function.py
b/paimon-python/pypaimon/read/reader/aggregation_merge_function.py
index 3d4b64f674..d5437be78e 100644
--- a/paimon-python/pypaimon/read/reader/aggregation_merge_function.py
+++ b/paimon-python/pypaimon/read/reader/aggregation_merge_function.py
@@ -27,8 +27,8 @@ aggregation (sum / max / min / last_value / ...) per column.
This is the **core merge semantics only**. Retract on DELETE /
UPDATE_BEFORE rows (with ``aggregation.remove-record-on-delete`` and
-``fields.<field>.ignore-retract`` opt-ins) and 2 additional
-aggregators (``hll_sketch`` / ``rbm64``)
+``fields.<field>.ignore-retract`` opt-ins) and 1 additional
+aggregator (``rbm64``)
are intentionally deferred. Non-INSERT row
kinds raise ``NotImplementedError`` at :meth:`add` time so we never
silently corrupt data with a half-implemented contract, and
diff --git a/paimon-python/pypaimon/tests/test_aggregation_e2e.py
b/paimon-python/pypaimon/tests/test_aggregation_e2e.py
index fc69104352..da22421475 100644
--- a/paimon-python/pypaimon/tests/test_aggregation_e2e.py
+++ b/paimon-python/pypaimon/tests/test_aggregation_e2e.py
@@ -294,20 +294,19 @@ class AggregationMergeEngineE2ETest(unittest.TestCase):
)
def test_out_of_scope_field_aggregator_rejected(self):
- # hll_sketch is one of the aggregator identifiers this engine
- # doesn't support yet. The guard must reject the config rather
- # than let the per-field factory build a (silently wrong)
- # fallback.
+ # rbm64 is the aggregator identifier this engine doesn't support
+ # yet. The guard must reject the config rather than let the
+ # per-field factory build a (silently wrong) fallback.
self._create_and_expect_unsupported(
- 'agg_reject_hll_sketch',
- {'fields.label.aggregate-function': 'hll_sketch'},
+ 'agg_reject_rbm64',
+ {'fields.label.aggregate-function': 'rbm64'},
'fields.label.aggregate-function',
)
def test_out_of_scope_default_aggregator_rejected(self):
self._create_and_expect_unsupported(
- 'agg_reject_default_hll_sketch',
- {'fields.default-aggregate-function': 'hll_sketch'},
+ 'agg_reject_default_rbm64',
+ {'fields.default-aggregate-function': 'rbm64'},
'fields.default-aggregate-function',
)
diff --git a/paimon-python/pypaimon/tests/test_field_aggregators.py
b/paimon-python/pypaimon/tests/test_field_aggregators.py
index 9fe0047cc6..1cf661c68b 100644
--- a/paimon-python/pypaimon/tests/test_field_aggregators.py
+++ b/paimon-python/pypaimon/tests/test_field_aggregators.py
@@ -31,7 +31,7 @@ from decimal import Decimal as BigDecimal
from functools import reduce
from typing import List
-from _datasketches import update_theta_sketch
+from _datasketches import hll_sketch, update_theta_sketch
from pypaimon.common.options import CoreOptions, Options
from pypaimon.data import Timestamp, Decimal
@@ -55,6 +55,7 @@ from pypaimon.read.reader.aggregate.aggregators import (
FieldMergeMapWithKeyTimeAgg,
FieldMergeMapAgg,
FieldThetaSketchAgg,
+ FieldHllSketchAgg,
FieldRoaringBitmap32Agg,
)
from pypaimon.schema.data_types import AtomicType, DataField, RowType,
ArrayType, MapType
@@ -2542,6 +2543,69 @@ class FieldThetaSketchAggTest(unittest.TestCase):
self.assertEqual(agg.agg(acc2, input_val), acc2)
+class FieldHllSketchAggTest(unittest.TestCase):
+ """HLL sketches are approximate and their sparse ("list mode")
+ payload keeps coupons in insertion order, so merged bytes are not
+ stable across insertion orders. Assert on estimates instead, and pin
+ cross-language compatibility with fixtures produced by Java's
+ ``HllSketchUtil`` (datasketches-java 4.2.0, default lgK=12).
+ """
+
+ JAVA_SKETCH_OF_1_2_3 = bytes.fromhex(
+ "0201070c030803002bf2fb06862ff90d75816607")
+ JAVA_SKETCH_OF_3_4_5 = bytes.fromhex(
+ "0201070c030803007581660781bc5d067b65e608")
+ JAVA_UNION_OF_BOTH = bytes.fromhex(
+ "0201070c030805007581660781bc5d067b65e6082bf2fb06862ff90d")
+
+ @staticmethod
+ def sketch_of(*values: int) -> bytes:
+ sketch = hll_sketch(12)
+
+ for value in values:
+ sketch.update(value)
+
+ return bytes(sketch.serialize_compact())
+
+ @staticmethod
+ def estimate(data: bytes) -> float:
+ return hll_sketch.deserialize(data).get_estimate()
+
+ def test_field_hll_sketch_agg(self):
+ agg = _make("hll_sketch", "VARBINARY(20)")
+ self.assertIsInstance(agg, FieldHllSketchAgg)
+
+ input_val = self.sketch_of(1)
+ acc = self.sketch_of(2, 3)
+
+ self.assertIsNone(agg.agg(None, None))
+ self.assertEqual(agg.agg(None, input_val), input_val)
+ self.assertEqual(agg.agg(acc, None), acc)
+
+ merged = agg.agg(acc, input_val)
+ self.assertAlmostEqual(self.estimate(merged), 3.0, places=6)
+
+ # Folding in a sketch already covered by the accumulator keeps
+ # the distinct count stable.
+ self.assertAlmostEqual(
+ self.estimate(agg.agg(merged, input_val)), 3.0, places=6)
+
+ def test_merges_java_produced_sketches(self):
+ agg = _make("hll_sketch", "VARBINARY(20)")
+
+ merged = agg.agg(self.JAVA_SKETCH_OF_1_2_3, self.JAVA_SKETCH_OF_3_4_5)
+
+ # {1,2,3} ∪ {3,4,5} == 5 distinct, and Java's own union of the
+ # same pair agrees.
+ self.assertAlmostEqual(self.estimate(merged), 5.0, places=6)
+ self.assertAlmostEqual(
+ self.estimate(self.JAVA_UNION_OF_BOTH), 5.0, places=6)
+
+ def test_rejects_non_binary_column(self):
+ with self.assertRaises(ValueError):
+ _make("hll_sketch", "INT")
+
+
class FieldRoaringBitmap32AggTest(unittest.TestCase):
def test_field_roaring_bitmap32_agg(self):
@@ -2582,8 +2646,8 @@ class FieldRoaringBitmap32AggTest(unittest.TestCase):
class RegistrationTest(unittest.TestCase):
- """Sanity check that all 10 expected aggregators (the primary-key
- placeholder plus 9 value aggregators) are registered when the
+ """Sanity check that all 20 expected aggregators (the primary-key
+ placeholder plus 19 value aggregators) are registered when the
package is imported. Guards against future refactors silently
dropping a registration.
"""
@@ -2594,6 +2658,13 @@ class RegistrationTest(unittest.TestCase):
"first_value", "first_non_null_value",
"sum", "max", "min",
"bool_or", "bool_and",
+ "product",
+ "listagg",
+ "collect",
+ "nested_update", "nested_partial_update",
+ "merge_map", "merge_map_with_keytime",
+ "theta_sketch", "hll_sketch",
+ "rbm32",
])
def test_all_expected_aggregators_registered(self):
@@ -2603,6 +2674,17 @@ class RegistrationTest(unittest.TestCase):
self.assertEqual(missing, set(),
"Missing built-in aggregators: {}".format(missing))
+ def test_merge_engine_guard_lists_the_same_aggregators(self):
+ # ``merge_engine_support`` duplicates the identifier list on
+ # purpose (it must stay import-free of the read pipeline), so
+ # the two can drift. Adding an aggregator without updating the
+ # guard would leave it rejected at read time even though it is
+ # registered; pin them together here.
+ from pypaimon.read.merge_engine_support import (
+ _AGGREGATION_SUPPORTED_AGG_FUNCS)
+ self.assertEqual(set(_AGGREGATION_SUPPORTED_AGG_FUNCS),
+ set(self.EXPECTED))
+
if __name__ == '__main__':
unittest.main()
diff --git
a/paimon-python/pypaimon/tests/test_optional_datasketches_dependency.py
b/paimon-python/pypaimon/tests/test_optional_datasketches_dependency.py
index 37297d1949..f5fd6d705d 100644
--- a/paimon-python/pypaimon/tests/test_optional_datasketches_dependency.py
+++ b/paimon-python/pypaimon/tests/test_optional_datasketches_dependency.py
@@ -61,6 +61,16 @@ class OptionalDataSketchesDependencyTest(unittest.TestCase):
else:
raise AssertionError(
"theta_sketch should require datasketches")
+
+ hll_agg = create_field_aggregator(
+ AtomicType("VARBINARY"), "value", "hll_sketch", options)
+ try:
+ hll_agg.agg(b"first", b"second")
+ except ImportError as exc:
+ assert "pypaimon[hll-sketch]" in str(exc)
+ else:
+ raise AssertionError(
+ "hll_sketch should require datasketches")
"""
)
env = os.environ.copy()
diff --git a/paimon-python/setup.py b/paimon-python/setup.py
index a5b1b6209e..27e5efe6e2 100644
--- a/paimon-python/setup.py
+++ b/paimon-python/setup.py
@@ -193,6 +193,10 @@ setup(
'datasketches>=4,<5; python_version<"3.9"',
'datasketches>=5,<6; python_version>="3.9"',
],
+ 'hll-sketch': [
+ 'datasketches>=4,<5; python_version<"3.9"',
+ 'datasketches>=5,<6; python_version>="3.9"',
+ ],
'sql': [
'pypaimon-rust>=0.3.0; python_version>="3.10"',
'datafusion>=54,<55; python_version>="3.10"',
