1 /***********************************************************************
2  * Copyright (c) 2013-2024 Commonwealth Computer Research, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * http://www.opensource.org/licenses/apache2.0.php.
7  ***********************************************************************/
8 
9 /**
10   * Adapted from:
11   *
12   * stream-lib
13   * Copyright (C) 2011 Clearspring Technologies, Inc.
14   * Copyright 2016 AddThis
15   *
16   * This product includes software developed by AddThis.
17   *
18   * This product also includes code adapted from:
19   *
20   * Apache Solr (http://lucene.apache.org/solr/)
21   * Copyright 2014 The Apache Software Foundation
22   *
23   * Apache Mahout (http://mahout.apache.org/)
24   * Copyright 2014 The Apache Software Foundation
25   *
26   */
27 
28 package org.locationtech.geomesa.utils.clearspring
29 
30 import com.clearspring.analytics.hash.MurmurHash
31 import com.clearspring.analytics.stream.cardinality.ICardinality
32 
33 /**
34   * Scala implementation of HyperLogLog (HLL) algorithm from this paper:
35   * <p/>
36   * http://algo.inria.fr/flajolet/Publications/FlFuGaMe07.pdf
37   * <p/>
38   * HLL is an improved version of LogLog that is capable of estimating
39   * the cardinality of a set with accuracy = 1.04/sqrt(m) where
40   * m = math.pow(2, b).  So we can control accuracy vs space usage by increasing
41   * or decreasing b.
42   * <p/>
43   * The main benefit of using HLL over LL is that it only requires 64%
44   * of the space that LL does to get the same accuracy.
45   * <p/>
46   * This implementation implements a single counter.  If a large (millions)
47   * number of counters are required you may want to refer to:
48   * <p/>
49   * http://dsiutils.dsi.unimi.it/
50   * <p/>
51   * It has a more complex implementation of HLL that supports multiple counters
52   * in a single object, drastically reducing the java overhead from creating
53   * a large number of objects.
54   * <p/>
55   * This implementation leveraged a javascript implementation that Yammer has
56   * been working on:
57   * <p/>
58   * https://github.com/yammer/probablyjs
59   * <p>
60   * Note that this implementation does not include the long range correction function
61   * defined in the original paper.  Empirical evidence shows that the correction
62   * function causes more harm than good.
63   * </p>
64   * <p/>
65   * <p>
66   * Users have different motivations to use different types of hashing functions.
67   * Rather than try to keep up with all available hash functions and to remove
68   * the concern of causing future binary incompatibilities this class allows clients
69   * to offer the value in hashed int or long form.  This way clients are free
70   * to change their hash function on their own time line.  We recommend using Google's
71   * Guava Murmur3_128 implementation as it provides good performance and speed when
72   * high precision is required.  In our tests the 32bit MurmurHash function included
73   * in this project is faster and produces better results than the 32 bit murmur3
74   * implementation google provides.
75   * </p>
76   */
77 /**
78   * Create a new HyperLogLog instance.  The log2m parameter defines the accuracy of
79   * the counter.  The larger the log2m the better the accuracy.
80   * <p/>
81   * accuracy = 1.04/sqrt(math.pow(2, log2m))
82   *
83   * @param log2m - the number of bits to use as the basis for the HLL instance
84   */
85 class HyperLogLog private (val log2m: Int, val registerSet: RegisterSet) extends ICardinality {
86 
87   require(log2m >= 0 && log2m <= 30, s"log2m argument $log2m is outside the acceptable range [0, 30]")
88 
89   private val alphaMM: Double = HyperLogLog.getAlphaMM(log2m, 1 << log2m)
90 
91   override def offerHashed(hashedValue: Long): Boolean = {
92     // j becomes the binary address determined by the first b log2m of x
93     // j will be between 0 and 2^log2m
94     val j = (hashedValue >>> (java.lang.Long.SIZE - log2m)).asInstanceOf[Int]
95     val r = java.lang.Long.numberOfLeadingZeros((hashedValue << log2m) | (1 << (log2m - 1)) + 1) + 1
96     registerSet.updateIfGreater(j, r)
97   }
98 
99   override def offerHashed(hashedValue: Int): Boolean = {
100     // j becomes the binary address determined by the first b log2m of x
101     // j will be between 0 and 2^log2m
102     val j = hashedValue >>> (Integer.SIZE - log2m)
103     val r = Integer.numberOfLeadingZeros((hashedValue << log2m) | (1 << (log2m - 1)) + 1) + 1
104     registerSet.updateIfGreater(j, r)
105   }
106 
107   override def offer(o: Any): Boolean = offerHashed(MurmurHash.hash(o))
108 
109   override def cardinality: Long = {
110     var registerSum = 0d
111     var zeros = 0.0
112 
113     var j = 0
114     while (j < registerSet.count) {
115       val value = registerSet.get(j)
116       registerSum += 1.0 / (1 << value)
117       if (value == 0) {
118         zeros += 1
119       }
120       j += 1
121     }
122 
123     val estimate = alphaMM * (1 / registerSum)
124 
125     if (estimate <= (5.0 / 2.0) * registerSet.count) {
126       // Small Range Estimate
127       math.round(HyperLogLog.linearCounting(registerSet.count, zeros))
128     } else {
129       math.round(estimate)
130     }
131   }
132 
133   override def sizeof(): Int = registerSet.size * 4
134 
135   /**
136     * Add all the elements of the other set to this set.
137     * <p/>
138     * This operation does not imply a loss of precision.
139     *
140     * @param other A compatible Hyperloglog instance (same log2m)
141     * @throws IllegalArgumentException if other is not compatible
142     */
143   def +=(other: HyperLogLog): Unit = {
144     require(registerSet.size == other.registerSet.size, "Cannot merge estimators of different sizes")
145     registerSet.merge(other.registerSet)
146   }
147 
148   override def merge(estimators: ICardinality*): HyperLogLog = {
149     val merged = new HyperLogLog(log2m, new RegisterSet(this.registerSet.count)())
150     merged += this
151     estimators.foreach {
152       case h: HyperLogLog => merged += h
153       case e => throw new IllegalArgumentException(s"Cannot merge estimators of different class: $e")
154     }
155     merged
156   }
157 
158   override def getBytes: Array[Byte] = throw new NotImplementedError
159 }
160 
161 object HyperLogLog {
162 
163   def apply(log2m: Int): HyperLogLog = new HyperLogLog(log2m, new RegisterSet(1 << log2m)())
164 
165   def apply(log2m: Int, register: Array[Int]): HyperLogLog =
166     new HyperLogLog(log2m, new RegisterSet(1 << log2m)(register))
167 
168   private def getAlphaMM(p: Int, m: Int): Double = {
169     // See the paper.
170     p match {
171       case 4 => 0.673 * m * m
172       case 5 => 0.697 * m * m
173       case 6 => 0.709 * m * m
174       case _ => (0.7213 / (1 + 1.079 / m)) * m * m
175     }
176   }
177 
178   private def linearCounting(m: Int, V: Double): Double = m * math.log(m / V)
179 }