Skip to content

Commit c3a5395

Browse files
committed
Implemented DDSketch
1 parent 3fa6bdc commit c3a5395

12 files changed

Lines changed: 948 additions & 0 deletions

File tree

QuantileFlow/__init__.py

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,13 @@
2020
"""
2121
from QuantileFlow.momentsketch.core import MomentSketch
2222
from QuantileFlow.hdrhistogram.core import HDRHistogram
23+
from QuantileFlow.ddsketch.core import DDSketch
2324

2425
__version__ = "0.0.3"
2526
__all__ = [
2627
"MomentSketch",
2728
"HDRHistogram",
29+
"DDSketch",
2830
]
2931

3032
if __name__ == "__main__":

QuantileFlow/ddsketch/__init__.py

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
"""
2+
DDSketch: Distributed and Mergeable Quantile Sketch with Relative Error Guarantees
3+
4+
This module provides an implementation of the DDSketch algorithm for computing approximate
5+
quantiles with a user-defined relative error bound. Key features include:
6+
7+
- Configurable accuracy: Set the desired relative error guarantee for quantile estimation
8+
- Mergeable: Sketches can be combined for distributed applications
9+
- Space efficient: Uses compact bucket structures to minimize memory usage
10+
- Fast updates: Insert operations are O(1) time complexity
11+
- Robust: Maintains error guarantees across the entire value range
12+
13+
The implementation includes different mapping schemes:
14+
- Logarithmic: The canonical implementation with provable relative error guarantees
15+
- Linear interpolation: Approximation using linear interpolation for improved performance
16+
- Cubic interpolation: Approximation using cubic interpolation for better memory efficiency
17+
"""
18+
19+
from QuantileFlow.ddsketch.core import DDSketch
20+
from QuantileFlow.ddsketch.mapping.logarithmic import LogarithmicMapping
21+
from QuantileFlow.ddsketch.mapping.linear_interpolation import LinearInterpolationMapping
22+
from QuantileFlow.ddsketch.mapping.cubic_interpolation import CubicInterpolationMapping
23+
from QuantileFlow.ddsketch.storage.contiguous import ContiguousStorage
24+
from QuantileFlow.ddsketch.storage.sparse import SparseStorage
25+
from QuantileFlow.ddsketch.storage.base import BucketManagementStrategy, Storage
26+
27+
__all__ = [
28+
"DDSketch",
29+
"LogarithmicMapping",
30+
"LinearInterpolationMapping",
31+
"CubicInterpolationMapping",
32+
"ContiguousStorage",
33+
"SparseStorage",
34+
"BucketManagementStrategy",
35+
"Storage"
36+
]

QuantileFlow/ddsketch/core.py

Lines changed: 195 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,195 @@
1+
"""Core DDSketch implementation."""
2+
3+
from typing import Literal, Union
4+
from .mapping.logarithmic import LogarithmicMapping
5+
from .mapping.linear_interpolation import LinearInterpolationMapping
6+
from .mapping.cubic_interpolation import CubicInterpolationMapping
7+
from .storage.base import BucketManagementStrategy
8+
from .storage.contiguous import ContiguousStorage
9+
from .storage.sparse import SparseStorage
10+
11+
class DDSketch:
12+
"""
13+
DDSketch implementation for quantile approximation with relative-error guarantees.
14+
15+
This implementation supports different mapping schemes and storage types for
16+
optimal performance in different scenarios. It can handle both positive and
17+
negative values, and provides configurable bucket management strategies.
18+
19+
Reference:
20+
"DDSketch: A Fast and Fully-Mergeable Quantile Sketch with Relative-Error Guarantees"
21+
by Charles Masson, Jee E. Rim and Homin K. Lee
22+
"""
23+
24+
def __init__(
25+
self,
26+
relative_accuracy: float,
27+
mapping_type: Literal['logarithmic', 'lin_interpol', 'cub_interpol'] = 'logarithmic',
28+
max_buckets: int = 2048,
29+
bucket_strategy: BucketManagementStrategy = BucketManagementStrategy.FIXED,
30+
cont_neg: bool = True
31+
):
32+
"""
33+
Initialize DDSketch.
34+
35+
Args:
36+
relative_accuracy: The relative accuracy guarantee (alpha).
37+
Must be between 0 and 1.
38+
mapping_type: The type of mapping scheme to use:
39+
- 'logarithmic': Basic logarithmic mapping
40+
- 'lin_interpol': Linear interpolation mapping
41+
- 'cub_interpol': Cubic interpolation mapping
42+
max_buckets: Maximum number of buckets per store (default 2048).
43+
If cont_neg is True, each store will have max_buckets buckets.
44+
bucket_strategy: Strategy for managing bucket count.
45+
If FIXED, uses ContiguousStorage, otherwise uses SparseStorage.
46+
cont_neg: Whether to handle negative values (default True).
47+
48+
Raises:
49+
ValueError: If relative_accuracy is not between 0 and 1.
50+
"""
51+
if not 0 < relative_accuracy < 1:
52+
raise ValueError("relative_accuracy must be between 0 and 1")
53+
54+
self.relative_accuracy = relative_accuracy
55+
self.cont_neg = cont_neg
56+
57+
58+
# Initialize mapping scheme
59+
if mapping_type == 'logarithmic':
60+
self.mapping = LogarithmicMapping(relative_accuracy)
61+
elif mapping_type == 'lin_interpol':
62+
self.mapping = LinearInterpolationMapping(relative_accuracy)
63+
elif mapping_type == 'cub_interpol':
64+
self.mapping = CubicInterpolationMapping(relative_accuracy)
65+
66+
# Choose storage type based on strategy
67+
if bucket_strategy == BucketManagementStrategy.FIXED:
68+
self.positive_store = ContiguousStorage(max_buckets)
69+
self.negative_store = ContiguousStorage(max_buckets) if cont_neg else None
70+
else:
71+
self.positive_store = SparseStorage(strategy=bucket_strategy)
72+
self.negative_store = SparseStorage(strategy=bucket_strategy) if cont_neg else None
73+
74+
self.count = 0
75+
self.zero_count = 0
76+
77+
def insert(self, value: Union[int, float]) -> None:
78+
"""
79+
Insert a value into the sketch.
80+
81+
Args:
82+
value: The value to insert.
83+
84+
Raises:
85+
ValueError: If value is negative and cont_neg is False.
86+
"""
87+
if value == 0:
88+
self.zero_count += 1
89+
elif value > 0:
90+
bucket_idx = self.mapping.compute_bucket_index(value)
91+
self.positive_store.add(bucket_idx)
92+
elif value < 0 and self.cont_neg:
93+
bucket_idx = self.mapping.compute_bucket_index(-value)
94+
self.negative_store.add(bucket_idx)
95+
elif value < 0:
96+
raise ValueError("Negative values not supported when cont_neg is False")
97+
self.count += 1
98+
99+
def delete(self, value: Union[int, float]) -> None:
100+
"""
101+
Delete a value from the sketch.
102+
103+
Args:
104+
value: The value to delete.
105+
106+
Raises:
107+
ValueError: If value is negative and cont_neg is False.
108+
"""
109+
if self.count == 0:
110+
return
111+
112+
deleted = False
113+
if value == 0 and self.zero_count > 0:
114+
self.zero_count -= 1
115+
deleted = True
116+
elif value > 0:
117+
bucket_idx = self.mapping.compute_bucket_index(value)
118+
deleted = self.positive_store.remove(bucket_idx)
119+
elif value < 0 and self.cont_neg:
120+
bucket_idx = self.mapping.compute_bucket_index(-value)
121+
deleted = self.negative_store.remove(bucket_idx)
122+
elif value < 0:
123+
raise ValueError("Negative values not supported when cont_neg is False")
124+
125+
if deleted:
126+
self.count -= 1
127+
128+
def quantile(self, q: float) -> float:
129+
"""
130+
Compute the approximate quantile.
131+
132+
Args:
133+
q: The desired quantile (between 0 and 1).
134+
135+
Returns:
136+
The approximate value at the specified quantile.
137+
138+
Raises:
139+
ValueError: If q is not between 0 and 1 or if the sketch is empty.
140+
"""
141+
if not 0 <= q <= 1:
142+
raise ValueError("Quantile must be between 0 and 1")
143+
if self.count == 0:
144+
raise ValueError("Cannot compute quantile of empty sketch")
145+
146+
rank = q * (self.count - 1)
147+
148+
if self.cont_neg:
149+
neg_count = self.negative_store.total_count
150+
if rank < neg_count:
151+
# Handle negative values
152+
curr_count = 0
153+
if self.negative_store.min_index is not None:
154+
for idx in range(self.negative_store.max_index, self.negative_store.min_index - 1, -1):
155+
bucket_count = self.negative_store.get_count(idx)
156+
curr_count += bucket_count
157+
if curr_count > rank:
158+
return -self.mapping.compute_value_from_index(idx)
159+
rank -= neg_count
160+
161+
if rank < self.zero_count:
162+
return 0
163+
rank -= self.zero_count
164+
165+
curr_count = 0
166+
if self.positive_store.min_index is not None:
167+
for idx in range(self.positive_store.min_index, self.positive_store.max_index + 1):
168+
bucket_count = self.positive_store.get_count(idx)
169+
curr_count += bucket_count
170+
if curr_count > rank:
171+
return self.mapping.compute_value_from_index(idx)
172+
173+
return float('inf')
174+
175+
def merge(self, other: 'DDSketch') -> None:
176+
"""
177+
Merge another DDSketch into this one.
178+
179+
Args:
180+
other: Another DDSketch instance to merge with this one.
181+
182+
Raises:
183+
ValueError: If the sketches are incompatible.
184+
"""
185+
if self.relative_accuracy != other.relative_accuracy:
186+
raise ValueError("Cannot merge sketches with different relative accuracies")
187+
188+
self.positive_store.merge(other.positive_store)
189+
if self.cont_neg and other.cont_neg:
190+
self.negative_store.merge(other.negative_store)
191+
elif other.cont_neg and sum(other.negative_store.counts.values()) > 0:
192+
raise ValueError("Cannot merge sketch containing negative values when cont_neg is False")
193+
194+
self.zero_count += other.zero_count
195+
self.count += other.count
Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
"""
2+
Mapping schemes for DDSketch algorithm.
3+
4+
This module provides various mapping schemes for the DDSketch algorithm:
5+
6+
- LogarithmicMapping: The canonical implementation with provable relative error guarantees
7+
- LinearInterpolationMapping: Faster approximation using linear interpolation
8+
- CubicInterpolationMapping: Memory-efficient approximation using cubic interpolation
9+
10+
All mapping schemes derive from the MappingScheme base class and provide methods to
11+
compute the bucket index for a given value and to recover a value from a bucket index.
12+
"""
13+
14+
from QuantileFlow.ddsketch.mapping.base import MappingScheme
15+
from QuantileFlow.ddsketch.mapping.logarithmic import LogarithmicMapping
16+
from QuantileFlow.ddsketch.mapping.linear_interpolation import LinearInterpolationMapping
17+
from QuantileFlow.ddsketch.mapping.cubic_interpolation import CubicInterpolationMapping
18+
19+
__all__ = [
20+
"MappingScheme",
21+
"LogarithmicMapping",
22+
"LinearInterpolationMapping",
23+
"CubicInterpolationMapping"
24+
]
Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
"""Base class for DDSketch mapping schemes."""
2+
3+
from abc import ABC, abstractmethod
4+
5+
6+
class MappingScheme(ABC):
7+
"""Abstract base class for different mapping schemes."""
8+
9+
@abstractmethod
10+
def compute_bucket_index(self, value: float) -> int:
11+
"""Compute the bucket index for a given value."""
12+
pass
13+
14+
@abstractmethod
15+
def compute_value_from_index(self, index: int) -> float:
16+
"""Compute the representative value for a given bucket index."""
17+
pass

0 commit comments

Comments
 (0)