Обзор архитектуры ClickHouse

ClickHouse - полноценная колоночная СУБД. Данные хранятся в колонках, а в процессе обработки - в массивах (векторах или фрагментах (chunk’ах) колонок). По возможности операции выполняются на массивах, а не на индивидуальных значениях. Это называется “векторизованное выполнения запросов” (vectorized query execution), и помогает снизить стоимость фактической обработки данных.

Эта идея не нова. Такой подход использовался в APL (A programming language, 1957) и его потомках: A + (диалект APL), J (1990), K (1993) и Q (язык программирования Kx Systems, 2003). Программирование на массивах (Array programming) используется в научных вычислительных системах. Эта идея не является чем-то новым и для реляционных баз данных: например, она используется в системе VectorWise (так же известной как Actian Vector Analytic Database от Actian Corporation).

Существует два различных подхода для увеличения скорости обработки запросов: выполнение векторизованного запроса и генерация кода во время выполнения (runtime code generation). В последнем случае код генерируется на лету для каждого типа запроса, удаляя все косвенные и динамические обращения. Ни один из этих подходов не имеет явного преимущества. Генерация кода во время выполнения выигрывает, если объединяет большое число операций, таким образом полностью используя вычислительные блоки и конвейер CPU. Выполнение векторизованного запроса может быть менее практично потому, что задействует временные векторы данных, которые должны быть записаны и прочитаны из кэша. Если временные данные не помещаются в L2 кэш, будут проблемы. С другой стороны выполнение векторизованного запроса упрощает использование SIMD инструкций CPU. Научная работа наших друзей показывает преимущества сочетания обоих подходов. ClickHouse использует выполнение векторизованного запроса и имеет ограниченную начальную поддержку генерации кода во время выполнения.

Колонки

Для представления столбцов в памяти (фактически, фрагментов столбцов) используется интерфейс IColumn. Интерфейс предоставляет вспомогательные методы для реализации различных реляционных операторов. Почти все операции иммутабельные: они не изменяют оригинальных колонок, а создают новую с измененными значениями. Например, метод IColumn :: filter принимает фильтр - набор байт. Он используется для реляционных операторов WHERE и HAVING. Другой пример: метод IColumn :: permute используется для поддержки ORDER BY, метод IColumn :: cut - LIMIT и т. д.

Различные реализации IColumn (ColumnUInt8, ColumnString и т. д.) отвечают за распределение данных колонки в памяти. Для колонок целочисленного типа это один смежный массив, такой как std :: vector. Для колонок типа String и Array это два вектора: один для всех элементов массивов, располагающихся смежно, второй для хранения смещения до начала каждого массива. Также существует реализация ColumnConst, в которой хранится только одно значение в памяти, но выглядит как колонка.

Поля

Тем не менее, можно работать и с индивидуальными значениями. Для представления индивидуальных значений используется Поле (Field). Field - размеченное объединение UInt64, Int64, Float64, String и Array. IColumn имеет метод оператор [] для получения значения по индексу n как Field, а также метод insert для добавления Field в конец колонки. Эти методы не очень эффективны, так как требуют временных объектов Field, представляющих индивидуальное значение. Есть более эффективные методы, такие как insertFrom, insertRangeFrom и т.д.

Field не несет в себе достаточно информации о конкретном типе данных в таблице. Например, UInt8, UInt16, UInt32 и UInt64 в Field представлены как UInt64.

Дырявые абстракции (Leaky Abstractions)

IColumn предоставляет методы для общих реляционных преобразований данных, но они не отвечают всем потребностям. Например, ColumnUInt64 не имеет метода для вычисления суммы двух столбцов, а ColumnString не имеет метода для запуска поиска по подстроке. Эти бесчисленные процедуры реализованы вне IColumn.

Различные функции на колонках могут быть реализованы обобщенным, неэффективным путем, используя IColumn методы для извлечения значений Field, или специальным путем, используя знания о внутреннем распределение данных в памяти в конкретной реализации IColumn. Для этого функции приводятся к конкретному типу IColumn и работают напрямую с его внутренним представлением. Например, в ColumnUInt64 есть метод getData, который возвращает ссылку на внутренний массив, чтение и заполнение которого, выполняется отдельной процедурой напрямую. Фактически, мы имеем "дырявые абстракции", обеспечивающие эффективные специализации различных процедур.

Типы данных (Data Types)

IDataType отвечает за сериализацию и десериализацию - чтение и запись фрагментов колонок или индивидуальных значений в бинарном или текстовом формате. IDataType точно соответствует типам данных в таблицах - DataTypeUInt32, DataTypeDateTime, DataTypeString и т. д.

IDataType и IColumn слабо связаны друг с другом. Различные типы данных могут быть представлены в памяти с помощью одной реализации IColumn. Например, и DataTypeUInt32, и DataTypeDateTime в памяти представлены как ColumnUInt32 или ColumnConstUInt32. В добавок к этому, один тип данных может быть представлен различными реализациями IColumn. Например, DataTypeUInt8 может быть представлен как ColumnUInt8 и ColumnConstUInt8.

IDataType хранит только метаданные. Например, DataTypeUInt8 не хранит ничего (кроме скрытого указателя vptr), а DataTypeFixedString хранит только N (фиксированный размер строки).

В IDataType есть вспомогательные методы для данных различного формата. Среди них методы сериализации значений, допускающих использование кавычек, сериализации значения в JSON или XML. Среди них нет прямого соответствия форматам данных. Например, различные форматы Pretty и TabSeparated могут использовать один вспомогательный метод serializeTextEscaped интерфейса IDataType.

Блоки (Block)

Block это контейнер, который представляет фрагмент (chunk) таблицы в памяти. Это набор троек - (IColumn, IDataType, имя колонки). В процессе выполнения запроса, данные обрабатываются Block-ами. Если у нас есть Block, значит у нас есть данные (в объекте IColumn), информация о типе (в IDataType), которая говорит нам, как работать с колонкой, и имя колонки (оригинальное имя колонки таблицы или служебное имя, присвоенное для получения промежуточных результатов вычислений).

При вычислении некоторой функции на колонках в блоке мы добавляем еще одну колонку с результатами в блок, не трогая колонки аргументов функции, потому что операции иммутабельные. Позже ненужные колонки могут быть удалены из блока, но не модифицированы. Это удобно для устранения общих подвыражений.

Блоки создаются для всех обработанных фрагментов данных. Напоминаем, что одни и те же типы вычислений, имена колонок и типы переиспользуются в разных блоках и только данные колонок изменяются. Лучше разделить данные и заголовок блока потому, что в блоках маленького размера мы имеем большой оверхэд по временным строкам при копировании умных указателей (shared_ptrs) и имен колонок.

Потоки блоков (Block Streams)

Потоки блоков обрабатывают данные. Мы используем потоки блоков для чтения данных, трансформации или записи данных куда-либо. IBlockInputStream предоставляет метод read для получения следующего блока, пока это возможно, и метод write, чтобы продвигать (push) блок куда-либо.

Потоки отвечают за:

1. Чтение и запись в таблицу. Таблица лишь возвращает поток для чтения или записи блоков.
2. Реализацию форматов данных. Например, при выводе данных в терминал в формате Pretty, вы создаете выходной поток блоков, который форматирует поступающие в него блоки.
3. Трансформацию данных. Допустим, у вас есть IBlockInputStream и вы хотите создать отфильтрованный поток. Вы создаете FilterBlockInputStream и инициализируете его вашим потоком. Затем вы тянете (pull) блоки из FilterBlockInputStream, а он тянет блоки исходного потока, фильтрует их и возвращает отфильтрованные блоки вам. Таким образом построены конвейеры выполнения запросов.

Имеются и более сложные трансформации. Например, когда вы тянете блоки из AggregatingBlockInputStream, он считывает все данные из своего источника, агрегирует их, и возвращает поток агрегированных данных вам. Другой пример: конструктор UnionBlockInputStream принимает множество источников входных данных и число потоков. Такой Stream работает в несколько потоков и читает данные источников параллельно.

Потоки блоков используют «втягивающий» (pull) подход к управлению потоком выполнения: когда вы вытягиваете блок из первого потока, он, следовательно, вытягивает необходимые блоки из вложенных потоков, так и работает весь конвейер выполнения. Ни «pull» ни «push» не имеют явного преимущества, потому что поток управления неявный, и это ограничивает в реализации различных функций, таких как одновременное выполнение нескольких запросов (слияние нескольких конвейеров вместе). Это ограничение можно преодолеть с помощью сопрограмм (coroutines) или просто запуском дополнительных потоков, которые ждут друг друга. У нас может быть больше возможностей, если мы сделаем поток управления явным: если мы локализуем логику для передачи данных из одной расчетной единицы в другую вне этих расчетных единиц. Читайте эту статью для углубленного изучения.

Следует отметить, что конвейер выполнения запроса создает временные данные на каждом шаге. Мы стараемся сохранить размер блока достаточно маленьким, чтобы временные данные помещались в кэш процессора. При таком допущении запись и чтение временных данных практически бесплатны по сравнению с другими расчетами. Мы могли бы рассмотреть альтернативу, которая заключается в том, чтобы объединить многие операции в конвейере вместе. Это может сделать конвейер как можно короче и удалить большую часть временных данных, что может быть преимуществом, но у такого подхода также есть недостатки. Например, разделенный конвейер позволяет легко реализовать кэширование промежуточных данных, использование промежуточных данных из аналогичных запросов, выполняемых одновременно, и объединение конвейеров для аналогичных запросов.

Форматы

Форматы данных реализуются с помощью потоков блоков. Есть форматы представления (presentational), пригодные только для вывода данных клиенту, такие как Pretty формат, который предоставляет только IBlockOutputStream. И есть форматы ввода/вывода, такие как TabSeparated или JSONEachRow.

Существуют также потоки строк: IRowInputStream и IRowOutputStream. Они позволяют вытягивать/выталкивать данные отдельными строками, а не блоками. Они нужны только для упрощения реализации ориентированных на строки форматов. Обертка BlockInputStreamFromRowInputStream и BlockOutputStreamFromRowOutputStream позволяет конвертировать потоки, ориентированные на строки, в обычные потоки, ориентированные на блоки.

I/O

Для байт-ориентированных ввода/вывода существуют абстрактные классы ReadBuffer и WriteBuffer. Они используются вместо C++ iostream. Не волнуйтесь: каждый зрелый проект C++ использует что-то другое вместо iostream по уважительным причинам.

ReadBuffer и WriteBuffer это просто непрерывный буфер и курсор, указывающий на позицию в этом буфере. Реализации могут как владеть так и не владеть памятью буфера. Существует виртуальный метод заполнения буфера следующими данными (для ReadBuffer) или сброса буфера куда-нибудь (например WriteBuffer). Виртуальные методы редко вызываются.

Реализации ReadBuffer/WriteBuffer используются для работы с файлами и файловыми дескрипторами, а также сетевыми сокетами, для реализации сжатия (CompressedWriteBuffer инициализируется вместе с другим WriteBuffer и осуществляет сжатие данных перед записью в него), и для других целей – названия ConcatReadBuffer, LimitReadBuffer, и HashingWriteBuffer говорят сами за себя.

Буферы чтения/записи имеют дело только с байтами. В заголовочных файлах ReadHelpers и WriteHelpers объявлены некоторые функции, чтобы помочь с форматированием ввода/вывода. Например, есть помощники для записи числа в десятичном формате.

Давайте посмотрим, что происходит, когда вы хотите вывести результат в JSON формате в стандартный вывод (stdout). У вас есть результирующий набор данных, готовый к извлечению из IBlockInputStream. Вы создаете WriteBufferFromFileDescriptor(STDOUT_FILENO) чтобы записать байты в stdout. Вы создаете JSONRowOutputStream, инициализируете с этим WriteBuffer'ом, чтобы записать строки JSON в stdout. Кроме того вы создаете BlockOutputStreamFromRowOutputStream, реализуя IBlockOutputStream. Затем вызывается copyData для передачи данных из IBlockInputStream в IBlockOutputStream и все работает. Внутренний JSONRowOutputStream будет писать в формате JSON различные разделители и вызвать IDataType::serializeTextJSON метод со ссылкой на IColumn и номер строки в качестве аргументов. Следовательно, IDataType::serializeTextJSON вызовет метод из WriteHelpers.h: например, writeText для числовых типов и writeJSONString для DataTypeString.

Таблицы

Интерфейс IStorage служит для отображения таблицы. Различные движки таблиц являются реализациями этого интерфейса. Примеры StorageMergeTree, StorageMemory и так далее. Экземпляры этих классов являются просто таблицами.

Ключевые методы IStorage это read и write. Есть и другие варианты - alter, rename, drop и так далее. Метод read принимает следующие аргументы: набор столбцов для чтения из таблицы, AST запрос и желаемое количество потоков для вывода. Он возвращает один или несколько объектов IBlockInputStream и информацию о стадии обработки данных, которая была завершена внутри табличного движка во время выполнения запроса.

В большинстве случаев метод read отвечает только за чтение указанных столбцов из таблицы, а не за дальнейшую обработку данных. Вся дальнейшая обработка данных осуществляется интерпретатором запросов и не входит в сферу ответственности IStorage.

Но есть и заметные исключения:

• AST запрос, передающийся в метод read, может использоваться движком таблицы для получения информации о возможности использования индекса и считывания меньшего количества данных из таблицы.
• Иногда движок таблиц может сам обрабатывать данные до определенного этапа. Например, StorageDistributed можно отправить запрос на удаленные серверы, попросить их обработать данные до этапа, когда данные с разных удаленных серверов могут быть объединены, и вернуть эти предварительно обработанные данные. Затем интерпретатор запросов завершает обработку данных.

Метод read может возвращать несколько объектов IBlockInputStream, позволяя осуществлять параллельную обработку данных. Эти несколько блочных входных потоков могут считываться из таблицы параллельно. Затем вы можете обернуть эти потоки различными преобразованиями (такими как вычисление выражений или фильтрация), которые могут быть вычислены независимо, и создать UnionBlockInputStream поверх них, чтобы читать из нескольких потоков параллельно.

Есть и другие варианты. Например, TableFunction возвращает временный объект IStorage, который можно подставить во FROM.

Чтобы получить быстрое представление о том, как реализовать свой движок таблиц, посмотрите на что-то простое, например StorageMemory или StorageTinyLog.

В качестве результата выполнения метода read, IStorage возвращает QueryProcessingStage – информацию о том, какие части запроса были обработаны внутри хранилища.

Парсеры (Parsers)

Написанный от руки парсер, анализирующий запрос, работает по методу рекурсивного спуска. Например, ParserSelectQuery просто рекурсивно вызывает нижестоящие парсеры для различных частей запроса. Парсеры создают абстрактное синтаксическое дерево (AST). AST представлен узлами, которые являются экземплярами IAST.

Генераторы парсеров не используются по историческим причинам.

Интерпретаторы

Интерпретаторы отвечают за создание конвейера выполнения запроса из AST. Есть простые интерпретаторы, такие как InterpreterExistsQuery и InterpreterDropQuery или более сложный InterpreterSelectQuery. Конвейер выполнения запроса представляет собой комбинацию входных и выходных потоков блоков. Например, результатом интерпретации SELECT запроса является IBlockInputStream для чтения результирующего набора данных; результат интерпретации INSERT запроса - это IBlockOutputStream, для записи данных, предназначенных для вставки; результат интерпретации INSERT SELECT запроса - это IBlockInputStream, который возвращает пустой результирующий набор при первом чтении, но копирует данные из SELECT к INSERT.

InterpreterSelectQuery использует ExpressionAnalyzer и ExpressionActions механизмы для анализа запросов и преобразований. Именно здесь выполняется большинство оптимизаций запросов на основе правил. ExpressionAnalyzer написан довольно грязно и должен быть переписан: различные преобразования запросов и оптимизации должны быть извлечены в отдельные классы, чтобы позволить модульные преобразования или запросы.

Функции

Существуют обычные функции и агрегатные функции. Агрегатные функции смотрите в следующем разделе.

Обычные функции не изменяют число строк и работают так, как если бы обрабатывали каждую строку независимо. В действительности же, функции вызываются не к отдельным строкам, а блокам данных для реализации векторизованного выполнения запросов.

Некоторые функции, такие как blockSize, rowNumberInBlock, и runningAccumulate, эксплуатируют блочную обработку и нарушают независимость строк.

ClickHouse имеет сильную типизацию, поэтому нет никакого неявного преобразования типов. Если функция не поддерживает определенную комбинацию типов, она создает исключение. Но функции могут работать (перегружаться) для многих различных комбинаций типов. Например, функция plus (для реализации + оператор) работает для любой комбинации числовых типов: UInt8 + Float32, UInt16 + Int8 и так далее. Кроме того, некоторые вариадические функции, такие как concat, могут принимать любое количество аргументов.

Реализация функции может быть немного неудобной, поскольку функция явно определяет поддерживаемые типы данных и поддерживается IColumns. Например, в plus функция имеет код, генерируемый экземпляром шаблона C++ для каждой комбинации числовых типов, а также постоянные или непостоянные левые и правые аргументы.

Это отличное место для реализации генерации кода во время выполнения, чтобы избежать раздувания кода шаблона. Кроме того, он позволяет добавлять слитые функции, такие как fused multiply-add или выполнять несколько сравнений в одной итерации цикла.

Из-за векторизованного выполнения запроса функции не закорачиваются. Например, если вы пишете WHERE f(x) AND g(y), обе части вычисляются, даже для строк, когда f(x) равно нулю (за исключением тех случаев, когда f(x) является нулевым постоянным выражением). Но если избирательность условия f(x) высока, и расчет f(x) обходится гораздо дешевле, чем g(y), лучше всего разделить вычисление на этапы. На первом этапе вычислить f(x), отфильтровать результирующие столбцы, а затем вычислять g(y) только для меньших, отфильтрованных фрагментов данных.

Агрегатные функции

Агрегатные функции - это функции с состоянием (stateful). Они накапливают переданные значения в некотором состоянии и позволяют получать результаты из этого состояния. Работа с ними осуществляется с помощью интерфейса IAggregateFunction. Состояния могут быть как простыми (состояние для AggregateFunctionCount это всего лишь одна переменная типа UInt64) так и довольно сложными (состояние AggregateFunctionUniqCombined представляет собой комбинацию линейного массива, хэш-таблицы и вероятностной структуры данных HyperLogLog).

Состояния распределяются в Arena (пул памяти) для работы с несколькими состояниями при выполнении запроса GROUP BY высокой кардинальности (большим числом уникальных данных). Состояния могут иметь нетривиальный конструктор и деструктор: например, сложные агрегатные состояния могут сами аллоцировать дополнительную память. Потому к созданию и уничтожению состояний, правильной передаче владения и порядку уничтожения следует уделять больше внимание.

Агрегатные состояния могут быть сериализованы и десериализованы для передачи их по сети во время выполнения распределенного запроса или для записи их на диск при дефиците оперативной памяти. Они даже могут храниться в таблице с DataTypeAggregateFunction, чтобы позволяет выполнять инкрементное агрегирование данных.

Формат сериализации данных для состояний агрегатных функций в настоящее время не версионируется. Это нормально, если агрегатные состояния хранятся только временно. Но у нас есть такая возможность AggregatingMergeTree механизм таблиц для инкрементной агрегации, и люди уже используют его в эксплуатации. Именно по этой причине требуется помнить об обратная совместимости при изменении формата сериализации для любой агрегатной функции.

Сервер

Сервер предоставляет несколько различных интерфейсов.

• HTTP интерфейс для любых сторонних клиентов.
• TCP интерфейс для родного ClickHouse клиента и межсерверной взаимодействия при выполнении распределенных запросов.
• Интерфейс для передачи данных при репликации.

Внутри простой многопоточный сервер без корутин (coroutines), файберов (fibers) и т.д. Поскольку сервер не предназначен для обработки большого количества простых запросов, а ориентирован на обработку сложных запросов относительно низкой интенсивности, каждый из потоков может обрабатывать огромное количество аналитических запросов.

Сервер инициализирует класс Context, где хранит необходимое для выполнения запроса окружение: список доступных баз данных, пользователей и прав доступа, настройки, кластеры, список процессов, журнал запросов и т.д. Это окружение используется интерпретаторами.

Мы поддерживаем полную обратную и прямую совместимость для TCP интерфейса: старые клиенты могут общаться с новыми серверами, а новые клиенты могут общаться со старыми серверами. Но мы не хотим поддерживать его вечно и прекращаем поддержку старых версий примерно через год.

Примечание

Для всех сторонних приложений мы рекомендуем использовать HTTP интерфейс, потому что он прост и удобен в использовании. TCP интерфейс тесно связан с внутренними структурами данных: он использует внутренний формат для передачи блоков данных и использует специальное кадрирование для сжатых данных. Мы не выпустили библиотеку C для этого протокола, потому что потребовала бы линковки большей части кодовой базы ClickHouse, что непрактично.

Выполнение распределенных запросов (Distributed Query Execution)

Сервера в кластере в основном независимы. Вы можете создать Распределенную (Distributed) таблицу на одном или всех серверах в кластере. Такая таблица сама по себе не хранит данные - она только предоставляет возможность "просмотра" всех локальных таблиц на нескольких узлах кластера. При выполнении SELECT распределенная таблица переписывает запрос, выбирает удаленные узлы в соответствии с настройками балансировки нагрузки и отправляет им запрос. Распределенная таблица просит удаленные сервера обработать запрос до той стадии, когда промежуточные результаты с разных серверов могут быть объединены. Затем он получает промежуточные результаты и объединяет их. Распределенная таблица пытается возложить как можно больше работы на удаленные серверы и сократить объем промежуточных данных, передаваемых по сети.

Ситуация усложняется при использовании подзапросов в случае IN или JOIN, когда каждый из них использует таблицу Distributed. Есть разные стратегии для выполнения таких запросов.

Глобального плана выполнения распределенных запросов не существует. Каждый узел имеет собственный локальный план для своей части работы. У нас есть простое однонаправленное выполнение распределенных запросов: мы отправляем запросы на удаленные узлы и затем объединяем результаты. Но это невозможно для сложных запросов GROUP BY высокой кардинальности или запросов с большим числом временных данных в JOIN: в таких случаях нам необходимо перераспределить («reshuffle») данные между серверами, что требует дополнительной координации. ClickHouse не поддерживает выполнение запросов такого рода, и нам нужно работать над этим.

Merge Tree

MergeTree - это семейство движков хранения, поддерживающих индексацию по первичному ключу. Первичный ключ может быть произвольным набором (кортежем) столбцов или выражений. Данные в таблице MergeTree хранятся "частями" (“parts”). Каждая часть хранит данные отсортированные по первичному ключу (данные упорядочены лексикографически). Все столбцы таблицы хранятся в отдельных файлах column.bin в этих частях. Файлы состоят из сжатых блоков. Каждый блок обычно содержит от 64 КБ до 1 МБ несжатых данных, в зависимости от среднего значения размера данных. Блоки состоят из значений столбцов, расположенных последовательно один за другим. Значения столбцов находятся в одинаковом порядке для каждого столбца (порядок определяется первичным ключом), поэтому, когда вы выполняете итерацию по многим столбцам, вы получаете значения для соответствующих строк.

Сам первичный ключ является "разреженным" ("sparse"). Он не относится к каждой отдельной строке, а только к некоторым диапазонам данных. Отдельный файл «primary.idx» имеет значение первичного ключа для каждой N-й строки, где N называется гранулярностью индекса ("index_granularity", обычно N = 8192). Также для каждого столбца у нас есть файлы column.mrk с "метками" ("marks"), которые обозначают смещение для каждой N-й строки в файле данных. Каждая метка представляет собой пару: смещение начала сжатого блока от начала файла и смещение к началу данных в распакованном блоке. Обычно сжатые блоки выравниваются по меткам, а смещение в распакованном блоке равно нулю. Данные для primary.idx всегда находятся в памяти, а данные для файлов column.mrk кэшируются.

Когда мы собираемся читать что-то из части данных MergeTree, мы смотрим содержимое primary.idx и определяем диапазоны, которые могут содержать запрошенные данные, затем просматриваем содержимое column.mrk и вычисляем смещение, чтобы начать чтение этих диапазонов. Из-за разреженности могут быть прочитаны лишние данные. ClickHouse не подходит для простых точечных запросов высокой интенсивности, потому что весь диапазон строк размером index_granularity должен быть прочитан для каждого ключа, а сжатый блок должен быть полностью распакован для каждого столбца. Мы сделали индекс разреженным, потому что мы должны иметь возможность поддерживать триллионы строк на один сервер без существенных расходов памяти на индексацию. Кроме того, поскольку первичный ключ является разреженным, он не уникален: он не может проверить наличие ключа в таблице во время INSERT. Вы можете иметь множество строк с одним и тем же ключом в таблице.

При выполнении INSERT для группы данных в MergeTree, элементы группы сортируются по первичному ключу и образует новую “часть”. Фоновые потоки периодически выбирают некоторые части и объединяют их в одну отсортированную часть, чтобы сохранить относительно небольшое количество частей. Вот почему он называется MergeTree. Конечно, объединение приводит к повышению интенсивности записи. Все части иммутабельные: они только создаются и удаляются, но не изменяются. Когда выполняется SELECT, он содержит снимок таблицы (набор частей). После объединения старые части также сохраняются в течение некоторого времени, чтобы упростить восстановление после сбоя, поэтому, если мы видим, что какая-то объединенная часть, вероятно, повреждена, мы можем заменить ее исходными частями.

MergeTree не является деревом LSM (Log-structured merge-tree — журнально-структурированное дерево со слиянием), потому что оно не содержит «memtable» и «log»: вставленные данные записываются непосредственно в файловую систему. Это делает его пригодным только для вставки данных в пакетах, а не по отдельным строкам и не очень часто - примерно раз в секунду это нормально, а тысячу раз в секунду - нет. Мы сделали это для простоты и потому, что мы уже вставляем данные в пакеты в наших приложениях.

Таблицы MergeTree могут иметь только один (первичный) индекс: вторичных индексов нет. Было бы неплохо разрешить несколько физических представлениям в одной логической таблице, например, хранить данные в более чем одном физическом порядке или даже разрешить представления с предварительно агрегированными данными вместе с исходными данными.

Существуют движки MergeTree, которые выполняют дополнительную работу во время фоновых слияний. Примерами являются CollapsingMergeTree и AggregatingMergeTree. Это можно рассматривать как специальную поддержку обновления. Помните, что это не настоящие обновления, поскольку пользователи обычно не контролируют время выполнения фоновых слияний, а данные в таблице MergeTree почти всегда хранятся в нескольких частях, а не в полностью объединенной форме.

Репликация

Репликация в ClickHouse может быть настроена для каждой таблицы отдельно. Вы можете иметь несколько реплицированных и несколько не реплицированных таблиц на одном сервере. Вы также можете реплицировать таблицы по-разному, например, одну с двухфакторной репликацией и другую с трехфакторной.

Репликация реализована в движке таблицы ReplicatedMergeTree. Путь в ZooKeeper указывается в качестве параметра движка. Все таблицы с одинаковым путем в ZooKeeper становятся репликами друг друга: они синхронизируют свои данные и поддерживают согласованность. Реплики можно добавлять и удалять динамически, просто создавая или удаляя таблицу.

Репликация использует асинхронную multi-master схему. Вы можете вставить данные в любую реплику, которая имеет открытую сессию в ZooKeeper, и данные реплицируются на все другие реплики асинхронно. Поскольку ClickHouse не поддерживает UPDATE, репликация исключает конфликты (conflict-free replication). Поскольку подтверждение вставок кворумом не реализовано, только что вставленные данные могут быть потеряны в случае сбоя одного узла.

Метаданные для репликации хранятся в ZooKeeper. Существует журнал репликации, в котором перечислены действия, которые необходимо выполнить. Среди этих действий: получить часть (get the part); объединить части (merge parts); удалить партицию (drop a partition) и так далее. Каждая реплика копирует журнал репликации в свою очередь, а затем выполняет действия из очереди. Например, при вставке в журнале создается действие «получить часть» (get the part), и каждая реплика загружает эту часть. Слияния координируются между репликами, чтобы получить идентичные до байта результаты. Все части объединяются одинаково на всех репликах. Одна из реплик-лидеров инициирует новое слияние кусков первой и записывает действия «слияния частей» в журнал. Несколько реплик (или все) могут быть лидерами одновременно. Реплике можно запретить быть лидером с помощью merge_tree настройки replicated_can_become_leader.

Репликация является физической: между узлами передаются только сжатые части, а не запросы. Слияния обрабатываются на каждой реплике независимо, в большинстве случаев, чтобы снизить затраты на сеть, во избежание усиления роли сети. Крупные объединенные части отправляются по сети только в случае значительной задержки репликации.

Кроме того, каждая реплика сохраняет свое состояние в ZooKeeper в виде набора частей и его контрольных сумм. Когда состояние в локальной файловой системе расходится с эталонным состоянием в ZooKeeper, реплика восстанавливает свою согласованность путем загрузки отсутствующих и поврежденных частей из других реплик. Когда в локальной файловой системе есть неожиданные или испорченные данные, ClickHouse не удаляет их, а перемещает в отдельный каталог и забывает об этом.

Примечание

Кластер ClickHouse состоит из независимых шардов, а каждый шард состоит из реплик. Кластер не является эластичным (not elastic), поэтому после добавления нового шарда данные не будут автоматически распределены между ними. Вместо этого нужно изменить настройки, чтобы выровнять нагрузку на кластер. Эта реализация дает вам больший контроль, и вполне приемлема для относительно небольших кластеров, таких как десятки узлов. Но для кластеров с сотнями узлов, которые мы используем в эксплуатации, такой подход становится существенным недостатком. Движки таблиц, которые охватывают весь кластер с динамически реплицируемыми областями, которые могут быть автоматически разделены и сбалансированы между кластерами, еще предстоит реализовать.

{## Original article ##}