mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-09-24 02:30:51 +00:00
541 lines
17 KiB
C++
541 lines
17 KiB
C++
#pragma once
|
||
|
||
#include <math.h>
|
||
|
||
#include <common/Common.h>
|
||
|
||
#include <DB/IO/WriteBuffer.h>
|
||
#include <DB/IO/WriteHelpers.h>
|
||
#include <DB/IO/ReadBuffer.h>
|
||
#include <DB/IO/ReadHelpers.h>
|
||
#include <DB/IO/VarInt.h>
|
||
|
||
#include <DB/Common/HashTable/HashTableAllocator.h>
|
||
#include <DB/Common/HashTable/Hash.h>
|
||
|
||
|
||
/** Приближённый рассчёт чего-угодно, как правило, построен по следующей схеме:
|
||
* - для рассчёта значения X используется некоторая структура данных;
|
||
* - в структуру данных добавляются не все значения, а только избранные (согласно некоторому критерию избранности);
|
||
* - после обработки всех элементов, структура данных находится в некотором состоянии S;
|
||
* - в качестве приближённого значения X возвращается значание, посчитанное по принципу максимального правдоподобия:
|
||
* при каком реальном значении X, вероятность нахождения структуры данных в полученном состоянии S максимальна.
|
||
*/
|
||
|
||
/** В частности, то, что описано ниже, можно найти по названию BJKST algorithm.
|
||
*/
|
||
|
||
/** Очень простое хэш-множество для приближённого подсчёта количества уникальных значений.
|
||
* Работает так:
|
||
* - вставлять можно UInt64;
|
||
* - перед вставкой, сначала вычисляется хэш-функция UInt64 -> UInt32;
|
||
* - исходное значение не сохраняется (теряется);
|
||
* - далее все операции производятся с этими хэшами;
|
||
* - хэш таблица построена по схеме:
|
||
* - open addressing (один буфер, позиция в буфере вычисляется взятием остатка от деления на его размер);
|
||
* - linear probing (если в ячейке уже есть значение, то берётся ячейка, следующая за ней и т. д.);
|
||
* - отсутствующее значение кодируется нулём; чтобы запомнить наличие в множестве нуля, используется отдельная переменная типа bool;
|
||
* - рост буфера в 2 раза при заполнении более чем на 50%;
|
||
* - если в множестве больше UNIQUES_HASH_MAX_SIZE элементов, то из множества удаляются все элементы,
|
||
* не делящиеся на 2, и затем все элементы, которые не делятся на 2, не вставляются в множество;
|
||
* - если ситуация повторяется, то берутся только элементы делящиеся на 4 и т. п.
|
||
* - метод size() возвращает приблизительное количество элементов, которые были вставлены в множество;
|
||
* - есть методы для быстрого чтения и записи в бинарный и текстовый вид.
|
||
*/
|
||
|
||
|
||
/// Максимальная степень размера буфера перед тем, как значения будут выкидываться
|
||
#define UNIQUES_HASH_MAX_SIZE_DEGREE 17
|
||
|
||
/// Максимальное количество элементов перед тем, как значения будут выкидываться
|
||
#define UNIQUES_HASH_MAX_SIZE (1 << (UNIQUES_HASH_MAX_SIZE_DEGREE - 1))
|
||
|
||
/** Количество младших бит, использующихся для прореживания. Оставшиеся старшие биты используются для определения позиции в хэш-таблице.
|
||
* (старшие биты берутся потому что младшие будут постоянными после выкидывания части значений)
|
||
*/
|
||
#define UNIQUES_HASH_BITS_FOR_SKIP (32 - UNIQUES_HASH_MAX_SIZE_DEGREE)
|
||
|
||
/// Начальная степень размера буфера
|
||
#define UNIQUES_HASH_SET_INITIAL_SIZE_DEGREE 4
|
||
|
||
|
||
/** Эта хэш-функция не самая оптимальная, но состояния UniquesHashSet, посчитанные с ней,
|
||
* хранятся много где на дисках (в Метраже), поэтому она продолжает использоваться.
|
||
*/
|
||
struct UniquesHashSetDefaultHash
|
||
{
|
||
size_t operator() (UInt64 x) const
|
||
{
|
||
return intHash32<0>(x);
|
||
}
|
||
};
|
||
|
||
|
||
template <typename Hash = UniquesHashSetDefaultHash>
|
||
class UniquesHashSet : private HashTableAllocatorWithStackMemory<(1 << UNIQUES_HASH_SET_INITIAL_SIZE_DEGREE) * sizeof(UInt32)>
|
||
{
|
||
private:
|
||
typedef UInt64 Value_t;
|
||
typedef UInt32 HashValue_t;
|
||
typedef HashTableAllocatorWithStackMemory<(1 << UNIQUES_HASH_SET_INITIAL_SIZE_DEGREE) * sizeof(UInt32)> Allocator;
|
||
|
||
UInt32 m_size; /// Количество элементов
|
||
UInt8 size_degree; /// Размер таблицы в виде степени двух
|
||
UInt8 skip_degree; /// Пропускать элементы не делящиеся на 2 ^ skip_degree
|
||
bool has_zero; /// Хэш-таблица содержит элемент со значением хэш-функции = 0.
|
||
|
||
HashValue_t * buf;
|
||
|
||
#ifdef UNIQUES_HASH_SET_COUNT_COLLISIONS
|
||
/// Для профилирования.
|
||
mutable size_t collisions;
|
||
#endif
|
||
|
||
void alloc(UInt8 new_size_degree)
|
||
{
|
||
buf = reinterpret_cast<HashValue_t *>(Allocator::alloc((1 << new_size_degree) * sizeof(buf[0])));
|
||
size_degree = new_size_degree;
|
||
}
|
||
|
||
void free()
|
||
{
|
||
if (buf)
|
||
{
|
||
Allocator::free(buf, buf_size() * sizeof(buf[0]));
|
||
buf = nullptr;
|
||
}
|
||
}
|
||
|
||
inline size_t buf_size() const { return 1 << size_degree; }
|
||
inline size_t max_fill() const { return 1 << (size_degree - 1); }
|
||
inline size_t mask() const { return buf_size() - 1; }
|
||
inline size_t place(HashValue_t x) const { return (x >> UNIQUES_HASH_BITS_FOR_SKIP) & mask(); }
|
||
|
||
/// Значение делится на 2 ^ skip_degree
|
||
inline bool good(HashValue_t hash) const
|
||
{
|
||
return hash == ((hash >> skip_degree) << skip_degree);
|
||
}
|
||
|
||
HashValue_t hash(Value_t key) const
|
||
{
|
||
return Hash()(key);
|
||
}
|
||
|
||
/// Удалить все значения, хэши которых не делятся на 2 ^ skip_degree
|
||
void rehash()
|
||
{
|
||
for (size_t i = 0; i < buf_size(); ++i)
|
||
{
|
||
if (buf[i] && !good(buf[i]))
|
||
{
|
||
buf[i] = 0;
|
||
--m_size;
|
||
}
|
||
}
|
||
|
||
/** После удаления элементов, возможно, освободилось место для элементов,
|
||
* которые были помещены дальше, чем нужно, из-за коллизии.
|
||
* Надо переместить их.
|
||
*/
|
||
for (size_t i = 0; i < buf_size(); ++i)
|
||
{
|
||
if (unlikely(buf[i] && i != place(buf[i])))
|
||
{
|
||
HashValue_t x = buf[i];
|
||
buf[i] = 0;
|
||
reinsertImpl(x);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Увеличить размер буфера в 2 раза или до new_size_degree, если указана ненулевая.
|
||
void resize(size_t new_size_degree = 0)
|
||
{
|
||
size_t old_size = buf_size();
|
||
|
||
if (!new_size_degree)
|
||
new_size_degree = size_degree + 1;
|
||
|
||
/// Расширим пространство.
|
||
buf = reinterpret_cast<HashValue_t *>(Allocator::realloc(buf, old_size * sizeof(buf[0]), (1 << new_size_degree) * sizeof(buf[0])));
|
||
size_degree = new_size_degree;
|
||
|
||
/** Теперь некоторые элементы может потребоваться переместить на новое место.
|
||
* Элемент может остаться на месте, или переместиться в новое место "справа",
|
||
* или переместиться левее по цепочке разрешения коллизий, из-за того, что элементы левее него были перемещены в новое место "справа".
|
||
* Также имеется особый случай:
|
||
* если элемент должен был быть в конце старого буфера, [ x]
|
||
* но находится в начале из-за цепочки разрешения коллизий, [o x]
|
||
* то после ресайза, он сначала снова окажется не на своём месте, [ xo ]
|
||
* и для того, чтобы перенести его куда надо,
|
||
* надо будет после переноса всех элементов из старой половинки [ o x ]
|
||
* обработать ещё хвостик из цепочки разрешения коллизий сразу после неё [ o x ]
|
||
* Именно для этого написано || buf[i] ниже.
|
||
*/
|
||
for (size_t i = 0; i < old_size || buf[i]; ++i)
|
||
{
|
||
HashValue_t x = buf[i];
|
||
if (!x)
|
||
continue;
|
||
|
||
size_t place_value = place(x);
|
||
|
||
/// Элемент на своём месте.
|
||
if (place_value == i)
|
||
continue;
|
||
|
||
while (buf[place_value] && buf[place_value] != x)
|
||
{
|
||
++place_value;
|
||
place_value &= mask();
|
||
|
||
#ifdef UNIQUES_HASH_SET_COUNT_COLLISIONS
|
||
++collisions;
|
||
#endif
|
||
}
|
||
|
||
/// Элемент остался на своём месте.
|
||
if (buf[place_value] == x)
|
||
continue;
|
||
|
||
buf[place_value] = x;
|
||
buf[i] = 0;
|
||
}
|
||
}
|
||
|
||
/// Вставить значение.
|
||
void insertImpl(HashValue_t x)
|
||
{
|
||
if (x == 0)
|
||
{
|
||
m_size += !has_zero;
|
||
has_zero = true;
|
||
return;
|
||
}
|
||
|
||
size_t place_value = place(x);
|
||
while (buf[place_value] && buf[place_value] != x)
|
||
{
|
||
++place_value;
|
||
place_value &= mask();
|
||
|
||
#ifdef UNIQUES_HASH_SET_COUNT_COLLISIONS
|
||
++collisions;
|
||
#endif
|
||
}
|
||
|
||
if (buf[place_value] == x)
|
||
return;
|
||
|
||
buf[place_value] = x;
|
||
++m_size;
|
||
}
|
||
|
||
/** Вставить в новый буфер значение, которое было в старом буфере.
|
||
* Используется при увеличении размера буфера, а также при чтении из файла.
|
||
*/
|
||
void reinsertImpl(HashValue_t x)
|
||
{
|
||
size_t place_value = place(x);
|
||
while (buf[place_value])
|
||
{
|
||
++place_value;
|
||
place_value &= mask();
|
||
|
||
#ifdef UNIQUES_HASH_SET_COUNT_COLLISIONS
|
||
++collisions;
|
||
#endif
|
||
}
|
||
|
||
buf[place_value] = x;
|
||
}
|
||
|
||
/** Если хэш-таблица достаточно заполнена, то сделать resize.
|
||
* Если элементов слишком много - то выкидывать половину, пока их не станет достаточно мало.
|
||
*/
|
||
void shrinkIfNeed()
|
||
{
|
||
if (unlikely(m_size > max_fill()))
|
||
{
|
||
if (m_size > UNIQUES_HASH_MAX_SIZE)
|
||
{
|
||
while (m_size > UNIQUES_HASH_MAX_SIZE)
|
||
{
|
||
++skip_degree;
|
||
rehash();
|
||
}
|
||
}
|
||
else
|
||
resize();
|
||
}
|
||
}
|
||
|
||
|
||
public:
|
||
UniquesHashSet() :
|
||
m_size(0),
|
||
skip_degree(0),
|
||
has_zero(false)
|
||
{
|
||
alloc(UNIQUES_HASH_SET_INITIAL_SIZE_DEGREE);
|
||
#ifdef UNIQUES_HASH_SET_COUNT_COLLISIONS
|
||
collisions = 0;
|
||
#endif
|
||
}
|
||
|
||
UniquesHashSet(const UniquesHashSet & rhs)
|
||
: m_size(rhs.m_size), skip_degree(rhs.skip_degree), has_zero(rhs.has_zero)
|
||
{
|
||
alloc(rhs.size_degree);
|
||
memcpy(buf, rhs.buf, buf_size() * sizeof(buf[0]));
|
||
}
|
||
|
||
UniquesHashSet & operator= (const UniquesHashSet & rhs)
|
||
{
|
||
if (size_degree != rhs.size_degree)
|
||
{
|
||
free();
|
||
alloc(rhs.size_degree);
|
||
}
|
||
|
||
m_size = rhs.m_size;
|
||
skip_degree = rhs.skip_degree;
|
||
has_zero = rhs.has_zero;
|
||
|
||
memcpy(buf, rhs.buf, buf_size() * sizeof(buf[0]));
|
||
|
||
return *this;
|
||
}
|
||
|
||
~UniquesHashSet()
|
||
{
|
||
free();
|
||
}
|
||
|
||
void insert(Value_t x)
|
||
{
|
||
HashValue_t hash_value = hash(x);
|
||
if (!good(hash_value))
|
||
return;
|
||
|
||
insertImpl(hash_value);
|
||
shrinkIfNeed();
|
||
}
|
||
|
||
size_t size() const
|
||
{
|
||
if (0 == skip_degree)
|
||
return m_size;
|
||
|
||
size_t res = m_size * (1 << skip_degree);
|
||
|
||
/** Псевдослучайный остаток - для того, чтобы не было видно,
|
||
* что количество делится на степень двух.
|
||
*/
|
||
res += (intHashCRC32(m_size) & ((1 << skip_degree) - 1));
|
||
|
||
/** Коррекция систематической погрешности из-за коллизий при хэшировании в UInt32.
|
||
* Формула fixed_res(res)
|
||
* - при каком количестве разных элементов fixed_res,
|
||
* при их случайном разбрасывании по 2^32 корзинам,
|
||
* получается в среднем res заполненных корзин.
|
||
*/
|
||
size_t p32 = 1ULL << 32;
|
||
size_t fixed_res = round(p32 * (log(p32) - log(p32 - res)));
|
||
return fixed_res;
|
||
}
|
||
|
||
void merge(const UniquesHashSet & rhs)
|
||
{
|
||
if (rhs.skip_degree > skip_degree)
|
||
{
|
||
skip_degree = rhs.skip_degree;
|
||
rehash();
|
||
}
|
||
|
||
if (!has_zero && rhs.has_zero)
|
||
{
|
||
has_zero = true;
|
||
++m_size;
|
||
shrinkIfNeed();
|
||
}
|
||
|
||
for (size_t i = 0; i < rhs.buf_size(); ++i)
|
||
{
|
||
if (rhs.buf[i] && good(rhs.buf[i]))
|
||
{
|
||
insertImpl(rhs.buf[i]);
|
||
shrinkIfNeed();
|
||
}
|
||
}
|
||
}
|
||
|
||
void write(DB::WriteBuffer & wb) const
|
||
{
|
||
if (m_size > UNIQUES_HASH_MAX_SIZE)
|
||
throw Poco::Exception("Cannot write UniquesHashSet: too large size_degree.");
|
||
|
||
DB::writeIntBinary(skip_degree, wb);
|
||
DB::writeVarUInt(m_size, wb);
|
||
|
||
if (has_zero)
|
||
{
|
||
HashValue_t x = 0;
|
||
DB::writeIntBinary(x, wb);
|
||
}
|
||
|
||
for (size_t i = 0; i < buf_size(); ++i)
|
||
if (buf[i])
|
||
DB::writeIntBinary(buf[i], wb);
|
||
}
|
||
|
||
void read(DB::ReadBuffer & rb)
|
||
{
|
||
has_zero = false;
|
||
|
||
DB::readIntBinary(skip_degree, rb);
|
||
DB::readVarUInt(m_size, rb);
|
||
|
||
if (m_size > UNIQUES_HASH_MAX_SIZE)
|
||
throw Poco::Exception("Cannot read UniquesHashSet: too large size_degree.");
|
||
|
||
free();
|
||
|
||
UInt8 new_size_degree = m_size <= 1
|
||
? UNIQUES_HASH_SET_INITIAL_SIZE_DEGREE
|
||
: std::max(UNIQUES_HASH_SET_INITIAL_SIZE_DEGREE, static_cast<int>(log2(m_size - 1)) + 2);
|
||
|
||
alloc(new_size_degree);
|
||
|
||
for (size_t i = 0; i < m_size; ++i)
|
||
{
|
||
HashValue_t x = 0;
|
||
DB::readIntBinary(x, rb);
|
||
if (x == 0)
|
||
has_zero = true;
|
||
else
|
||
reinsertImpl(x);
|
||
}
|
||
}
|
||
|
||
void readAndMerge(DB::ReadBuffer & rb)
|
||
{
|
||
UInt8 rhs_skip_degree = 0;
|
||
DB::readIntBinary(rhs_skip_degree, rb);
|
||
|
||
if (rhs_skip_degree > skip_degree)
|
||
{
|
||
skip_degree = rhs_skip_degree;
|
||
rehash();
|
||
}
|
||
|
||
size_t rhs_size = 0;
|
||
DB::readVarUInt(rhs_size, rb);
|
||
|
||
if (rhs_size > UNIQUES_HASH_MAX_SIZE)
|
||
throw Poco::Exception("Cannot read UniquesHashSet: too large size_degree.");
|
||
|
||
if ((1U << size_degree) < rhs_size)
|
||
{
|
||
UInt8 new_size_degree = std::max(UNIQUES_HASH_SET_INITIAL_SIZE_DEGREE, static_cast<int>(log2(rhs_size - 1)) + 2);
|
||
resize(new_size_degree);
|
||
}
|
||
|
||
for (size_t i = 0; i < rhs_size; ++i)
|
||
{
|
||
HashValue_t x = 0;
|
||
DB::readIntBinary(x, rb);
|
||
insertHash(x);
|
||
}
|
||
}
|
||
|
||
static void skip(DB::ReadBuffer & rb)
|
||
{
|
||
size_t size = 0;
|
||
|
||
rb.ignore();
|
||
DB::readVarUInt(size, rb);
|
||
|
||
if (size > UNIQUES_HASH_MAX_SIZE)
|
||
throw Poco::Exception("Cannot read UniquesHashSet: too large size_degree.");
|
||
|
||
rb.ignore(sizeof(HashValue_t) * size);
|
||
}
|
||
|
||
void writeText(DB::WriteBuffer & wb) const
|
||
{
|
||
if (m_size > UNIQUES_HASH_MAX_SIZE)
|
||
throw Poco::Exception("Cannot write UniquesHashSet: too large size_degree.");
|
||
|
||
DB::writeIntText(skip_degree, wb);
|
||
wb.write(",", 1);
|
||
DB::writeIntText(m_size, wb);
|
||
|
||
if (has_zero)
|
||
wb.write(",0", 2);
|
||
|
||
for (size_t i = 0; i < buf_size(); ++i)
|
||
{
|
||
if (buf[i])
|
||
{
|
||
wb.write(",", 1);
|
||
DB::writeIntText(buf[i], wb);
|
||
}
|
||
}
|
||
}
|
||
|
||
void readText(DB::ReadBuffer & rb)
|
||
{
|
||
has_zero = false;
|
||
|
||
DB::readIntText(skip_degree, rb);
|
||
DB::assertString(",", rb);
|
||
DB::readIntText(m_size, rb);
|
||
|
||
if (m_size > UNIQUES_HASH_MAX_SIZE)
|
||
throw Poco::Exception("Cannot read UniquesHashSet: too large size_degree.");
|
||
|
||
free();
|
||
|
||
UInt8 new_size_degree = m_size <= 1
|
||
? UNIQUES_HASH_SET_INITIAL_SIZE_DEGREE
|
||
: std::max(UNIQUES_HASH_SET_INITIAL_SIZE_DEGREE, static_cast<int>(log2(m_size - 1)) + 2);
|
||
|
||
alloc(new_size_degree);
|
||
|
||
for (size_t i = 0; i < m_size; ++i)
|
||
{
|
||
HashValue_t x = 0;
|
||
DB::assertString(",", rb);
|
||
DB::readIntText(x, rb);
|
||
if (x == 0)
|
||
has_zero = true;
|
||
else
|
||
reinsertImpl(x);
|
||
}
|
||
}
|
||
|
||
void insertHash(HashValue_t hash_value)
|
||
{
|
||
if (!good(hash_value))
|
||
return;
|
||
|
||
insertImpl(hash_value);
|
||
shrinkIfNeed();
|
||
}
|
||
|
||
#ifdef UNIQUES_HASH_SET_COUNT_COLLISIONS
|
||
size_t getCollisions() const
|
||
{
|
||
return collisions;
|
||
}
|
||
#endif
|
||
};
|
||
|
||
|
||
#undef UNIQUES_HASH_MAX_SIZE_DEGREE
|
||
#undef UNIQUES_HASH_MAX_SIZE
|
||
#undef UNIQUES_HASH_BITS_FOR_SKIP
|
||
#undef UNIQUES_HASH_SET_INITIAL_SIZE_DEGREE
|