Greenplum vs Citus. Часть 1

Citus работает с несколькими видами таблиц:

Пример записи в pg_dist_partition:

В данном случае имя логической таблицы (бывшей "обычной" локальной таблицы PostgreSQL) содержится в поле logicalrelid. Для данной таблицы распределение осуществляется по хешу (колонка partmethod), в колонке partkey содержится служебная информация по ключу распределения.

В Greenplum такого разделения не существует. Для таблиц append optimized существуют так называемые aux-таблицы, но они никак не скрываются и видны пользователям и/или администратору базы данных. Как следствие, нет необходимости и в явном преобразовании и неявном копировании исходных данных вызовом каких-либо специальных функций.

Таблицы, управляемые Citus, имеют следующие типы:

В Citus реализован так называемый многоуровневый подход к планированию запросов. Citus встраивается в процесс планирования за счёт использования доступных расширениям функций обратного вызова planner_hook, set_rel_pathlist_hook, get_relation_info_hook, set_join_pathlist_hook.

Входная точка планирования с точки зрения Citus — реализация distributed_planner. Для того, чтобы планировщик Citus включился в процесс, в запросе должна использоваться хотя бы одна таблица Citus. Является ли таблица соответствующей этому признаку определяется по внутренней системной таблице pg_dist_partition.

Планировщик разделяется на несколько уровней, которые задействуются в зависимости от структуры (можно даже сказать, от условной "сложности") запроса:

Каждый следующий уровень реализует более сложные либо высокоуровневые (группировки, сортировки и т.п.) подходы к планированию запросов.

Fast path planner

Fast path planner ограничен возможностями планирования запросов, по сути, затрагивающих один шард (либо использующих reference-таблицы).

Также запрос:

Этот тип планировщика ориентирован на OLTP-нагрузку, когда запрос можно перенаправить в один шард. В этом варианте планирования не происходит вызова standard_planner. Судя по документации, тем самым разработчики хотели избежать лишних накладных расходов, связанных с планированием стандартным планировщиком PostgreSQL. Такой вариант хорошо подходит для схем с использованием шардирования по некоторому ключу для так называемых Multi-tenant приложений. Например, когда все операции с таблицами ограничены каким-то определенным шардом, который соответствует id пользователя.

Router planner

Router planner дополняет fast path planner возможностью запрашивать несколько таблиц, но всё ещё находясь в границах одного шарда согласно условиям фильтрации по dist_key. Как было описано, Citus располагает реальные данные таблиц в отдельных таблицах шардов, которые в общем случае не видны пользователям. На этом этапе планировщик подменяет исходную, видимую пользователю распределённую таблицу на таблицы шардов с учётом их распределения по нодам. Данный тип планировщика уже умеет планировать подзапросы и CTE.

Например, для случая, когда таблицы a и b шардированы по одной колонке c1, следующий запрос может быть спланирован так:

На что можно обратить внимание:

В строке, отмеченной цифрой 3, можно увидеть Seq Scan по реальной физической таблице public.b_102332 (данные шарда). Как мы говорили ранее, эта таблица в общем случае не видна пользователям, но в плане она фигурирует как один из 32 шардов таблицы b.

Так как данные таблицы b шардированы по колонке c1 и в фильтре указано условие b.c1 = 1, то планировщик может задействовать router planner (fast path planner задействовать нельзя, так как в запросе присутствует подзапрос) и перенаправить запрос только в один шард.

В строке с номером 2 можно увидеть, что эта часть запроса планируется в виде подплана (SubPlan 1) и с точки зрения реального выполнения запроса может быть выполнена одним заданием (строка с номером 1 Task Count: 1). О заданиях (Task) подробнее поговорим в разделе про следующий тип планировщика recursive planning.

Если проводить сравнение схожего запроса в Greenplum, то можно выделить следующие сходства и различия:

Greenplum не выделяет этот сценарий с точки зрения перенаправления в один сегмент — в плане присутствует Gather Motion 32:1 (строка, отмеченная цифрой 1), который собирает данные с сегментов от нижележащих узлов. За счёт применения фильтра (строка с номером 6) объём данных в данном случае может быть сокращен, но формально план не будет отличаться для любого количества строк. Таблицы в Greenplum для этого запроса распределены по a.c1 и b.c1, и это позволяет планировщику, собрав данные от узла Seq Scan (строка 5), сразу же вычислить и агрегирующую функцию count(), b.c1 (строка 4), далее собрать по этим данным хеш-таблицу (строка 3) и осуществить локальное (в рамках сегмента) соединение Hash Left Join (строка 2).

По факту, план Citus выглядит более оптимальным для данного сценария, так как в его случае запрос сразу же перенаправляется в нужный шард (а это по умолчанию 1/32 от общего объёма данных всей таблицы), никакого ожидания остальных нод не происходит. Разумеется, это отражается на времени выполнения:

Однако картина сильно меняется в случае такого запроса (в данном случае уже используется более продвинутый планировщик recursive planner, который мы рассмотрим в следующем разделе):

Greenplum строит более продвинутый план, стоит только обратить внимание на строки, отмеченные цифрами 1 и 3:

Первое значимое отличие, что Greenplum рассчитал агрегат COUNT и сгруппировал результаты по колонке b.c1 (строки с номерами 2 и 3). Второе отличие, что GPORCA пробросила условие фильтрации (b.c1 > 19990000) вглубь плана (строка 4), что значимо сократило выборку данных из таблицы b. При этом для соединения был использован Hash Left Join (строка 1) — фактически проходя по таблице а с условием (a.c1 > 19990000) из хеш-таблицы (строка 3), получаем искомый результат.

Что делает Citus: фактически узел последовательного чтения Seq Scan по таблице a для каждой строки вызывает подплан SubPlan 1 (строка c номером 1), при этом условие фильтра (c1 > 100) чтения по таблице b (строка 3) приводит к полному чтению таблицы b (отброс первых 100 строк не в счёт). И так происходит 321 раз исходя из значения счетчика loops=321 (строка 2) для каждого из 32 шардов.

Нетрудно себе представить, что это приведёт к разительной разнице по времени выполнения. Даже для малого количества строк в таблицах a и b (по 21 млн. строк в каждой) разница во времени составила:

Если таблицы не шардированы по одной и той же колонке (в моём случае я пересоздал таблицы и шардировал одну по a.c1, вторую — по b.c2), то попытка выполнения такого запроса завершится неудачей:

Greenplum в свою очередь не испытывает тут каких-либо трудностей, как мы говорили ранее, план будет в целом похож на исходный (для случая распределения по одной колонке), но добавляется узел Redistribute Motion (строка, отмеченная цифрой 1):

Подводя итоги по планировщикам fast path planner и router planner:

Эти ограничения снимаются в следующем, более продвинутом типе планировщика recursive planner.

Recursive planner

Recursive planner реализует:

Данный тип планировщика пытается рассмотреть каждую возможную часть запроса на предмет её изолированного выполнения в pushdown-варианте. Суть этого действия в следующем: условно, на вход планировщика попадает дерево запроса, которое анализируется вглубь. В ходе планирования все подзапросы и CTE проверяются на их возможность исполнения непосредственно на рабочих нодах (worker). Если для данного подзапроса/CTE это невозможно, то для него осуществляется вызов обычного планировщика PostgreSQL (planner). Результат добавляется в список планов подзапросов. Важный момент состоит в том, что эти подпланы запускаются до начала обработки распределённого плана выполнения (distributed plan). Результаты отработки этих подпланов записываются во временные файлы на диске. Мы рассмотрим этот процесс в подробностях чуть позднее.

Таким подходом разработчики решают основную задачу — возможность поддержки выполнения большинства видов запросов SQL (да, исключения есть: среди них отсутствие поддержки GROUPING SETS, CUBE, ROLLUP).

Получив и, по сути, материализовав промежуточные результаты подзапроса на каждой ноде/шарде, Citus может соединять эти данные с распределёнными таблицами по любому ключу.

Стоит отметить, что в отличие от Greenplum рекурсивные и модифицирующие CTE, если запрос не ограничен одним шардом, не поддерживаются:

Предлагаю рассмотреть пример работы данного планировщика на паре запросов:

Recursive planner принимает решение спланировать подзапрос (строка, отмеченная цифрой 1) через подплан (строка с номером 2), собрать результат выполнения подплана на одной ноде и подсчитать итоговую агрегирующую функцию на основе промежуточных данных (строка 7).

Предлагаю разобрать, почему планировщик принимает такие решения и как осуществляется промежуточное хранение результатов.

Согласно правилам pushdown для подзапросов наличие LIMIT требует оформления подзапроса в виде отдельного подплана. Если рассматривать план снизу вверх, то строки от Seq Scan (строка 6) подаются на вход узлу сортировки Sort (строка 5), далее накладывается условие ограничение строк LIMIT. Эти действия выполняются параллельно на всех шардах таблицы (по умолчанию равных 32). Таким образом, на выходе этих узлов плана от каждого шарда приходят LIMIT N строк, отсортированных по a.c3. Однако, чтобы предоставить пользователю именно LIMIT N строк согласно ожидаемой пользователем сортировке, планировщику приходится упорядочить данные от всех шардов и только потом применить итоговое условие по LIMIT (строки 3 и 4). В противном случае пользователь может получить совсем не то, что он ожидает. Эти результаты кешируются на ноде-координаторе запроса и становятся доступны остальной части плана через функцию pg_catalog.read_intermediate_result. Механизм работы функции read_intermediate_result описан в разделе Особенности выполнения запросов.

Планировщик Greenplum планирует запрос так:

Тут стоит обратить внимание на шаг Gather Motion 32:1 и его параметризацию в виде Merge Key: c3 (строки 2 и 3). Планировщик GPORCA, зная, что результаты с каждого сегмента будут приходить упорядоченно согласно a.c3 (строка 4), имеет возможность сделать, по сути, соединение слиянием и применить LIMIT уже на результатах этого набора строк (строка 1). Этот шаг формально соответствует узлу Limit и Sort под ним (строки 3 и 4 в плане Citus).

Ключевое различие планов состоит в том, что Citus для каждого шарда выполняет распределённый подплан Distributed Subplan (строка 2), в рамках этого плана общий результат, полученный от шардов, сортируется (строка 4) и ограничивается узлом Limit (строка 3). Результаты собираются на координаторе запроса и материализуются в виде файлов, которые на координаторе вычитываются функцией read_intermediate_result (строка 8).

Greenplum в рамках Gather Motion делает это на лету и на практике показывает лучший результат:

Этот же тип планировщика позволяет спланировать запросы с HAVING в подзапросах, однако связанные (correlated) запросы ему в отличие от Greenplum уже не по зубам:

Greenplum же справляется с этой задаче без проблем:

Даже на довольно простых примерах Citus заставляет задумываться, а что мне нужно сделать со схемой и распределением данным, чтобы запрос всё-таки выполнялся. Планировщик GPORCA на практике выглядит более продвинутым.

Например, для коррелированных подзапросов:

Для первого запроса большие таблицы могут быть в дополнение к шардированию партицированы по каким-либо полям.

Logical Planner & Optimizer

Подробное рассмотрение этого уровня планирования выходит за рамки статьи. Задачи этого уровня отчасти сходны с решаемыми задачами groupping_planner в PostgreSQL: это операции GROUP BY, ORDER BY, LIMIT, агрегирующие функции.

Также на этом уровне осуществляется оптимизация операций соединений. Основная цель этой оптимизации состоит в сокращении числа соединений, ведущих к перераспределению данных. Судя по всему, ввиду специфики перераспределения на основе записи в файл промежуточных результатов и COPY-протокола (об этом далее), оценки cost-based не производится.

Citus строит свой план выполнения на основе реализации узла CustomScan.

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

Для начала сам запрос и его план:

Мы уже знаем, что такой запрос приводит к созданию и выполнению подплана (строка, отмеченная цифрой 1), использованию результатов этого подплана в Hash Join (строка с номером 6) и чтению промежуточных результатов в inner-части соединения (строка 7).

Citus встраивается в процесс выполнения запросов в таких точках расширения: ExecutorStart_hook, ExecutorRun_hook, ExplainOneQuery_hook, prev_ExecutorEnd, ExecutorEnd_hook.

Начнём наш отсчет от реализации ExecutorStart_hook функции обратного вызова CitusExecutorRun. Для такого запроса как наш, ядро PostgreSQL вызывает функцию CitusExecutorRun в ExecutorRun.

CitusExecutorRun помимо вспомогательных, специфичных для Citus задач вроде пропуска проверок ограничений (имеется в виду constraints check) на координаторе для запросов ALTER TABLE (это делегируется на ноды), решает основную задачу — поиск узлов "состояние сканирования" (CitusScanState), которые ранее были добавлены одним из планировщиков.

CitusScanState представляет собой структуру-"контейнер", которая включает в себя:

План (если быть точнее, то узлы PlanState) обходится вглубь, и найденные экземпляры CitusScanState добавляются в список.

Для нашего случая в этом списке будет один узел Custom Scan (Citus Adaptive) (строка 1) и ассоциированный с ним экземпляр CustomScanState. Для каждого из найденных узлов и ассоциированного с ним распределённого плана выполнения для распределённых таблиц берутся блокировки всех партиций, а также выполняются все подпланы, начиная с самого верхнего. В нашем случае это будет подплан Distributed Subplan 24_1.

Получение результатов на нодах осуществляется через экземпляр DestReceiver. Нода-координатор создаёт так называемый RemoteFileDestReceiver.

В его задачи входит:

Как только RemoteFileDestReceiver инициализирован, Citus начинает выполнение подплана (строка 2). Для этого используется механизм порталов (Portal). В портал отправляется спланированная ранее часть запроса HashAggregate (строка 3). Сам портал принимает в качестве одного из своих аргументов экземпляр DestReceiver, которым и выступает RemoteFileDestReceiver.

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

Далее план попадает на вход функции ядра ExecutePlan. Так как верхнеуровневым узлом плана является узел агрегатной функции HashAggregate (строка 3, наш COUNT(DISTINCT)), то управление передается ядру (модуль nodeAgg.c), а оно в свою очередь вызывает нижележащий узел, которым является узел CustomScan Citus.

Для выполнения задач сканирования (то есть непосредственного получения строк) реализация CustomScan использует так называемый AdaptiveExecutor. Полноценное рассмотрение этого модуля займёт ещё одну такую же статью, рассмотрим только основные задачи, касающиеся нашего запроса.

Верхнеуровнево, основная задача AdaptiveExecutor — выполнение набора заданий (list of tasks). Примером задания может служить запрос к шарду (по сути, к таблице). В нашем случае планировщик определил 32 задания (по числу шардов таблицы).

Также AdaptiveExecutor создает TupleStore для временного хранения полученных результатов.

Как работает функция read_intermediate_result:

Задача read_intermediate_result — представить промежуточный результат в формате COPY, полученный в ходе broadcast-выполнения подплана либо шафлинга (Repartitioning, об этом в деталях далее) в виде набора строк. Файл с промежуточным результатами парсится и преобразуется в те форматы колонок, которые вызывающая сторона передала в качестве параметров. Например, в случае нашего запроса (строка 5) read_intermediate_result('24_1'::text, 'binary'::citus_copy_format) intermediate_result(c2 integer) первый аргумент 24_1 представляет собой идентификатор результирующего набора строк, второй аргумент задаёт формат ('binary'). Идентификатор далее используется для получения имени файла в каталоге pgsql_job_cache.

Далее функция ReadFileIntoTupleStore перегоняет строки из формата COPY (используется PostgreSQL-реализация модуля copy.c) в Tuplestorestate. Далее результаты уже представлены в виде кортежей.

Где хранятся промежуточные результаты:

Промежуточные результаты выполнения запросов хранятся в каталоге base/pgsql_job_cache на каждой ноде. Если эти результаты используются в распределённой транзакции, то это каталог base/pgsql_job_cache/<user id>_<coordinator node id>_<transaction number>/. Если распределённая транзакция не используется, то этот путь base/pgsql_job_cache/<user id>_<process id>/.

Пример размещения:

Как Citus делает соединение не по ключу шардирования в том случае, когда таблицы не распределены по этому ключу?

Рассмотрим следующий запрос для случая, когда таблица a распределена по ключу a.c1, а таблица b по ключу b.c3 (для упрощения разбора механизма предварительно распределим каждую таблицу по 2 шардам):

Для выполнения подобного соединения задействуется repartitioning-механизм, основанный на разбиении на задания (Jobs) с одноуровневой иерархией, где некоторый набор заданий зависим от основного. По сути, этот набор заданий представляет собой вариацию подпланов, результаты выполнения которых перераспределяются между нодами согласно ключам. Отличие от обычных подпланов в том, что в заданиях отсутствует этап объединения результатов (имеется в виду этап merge), в подпланах результат в свою очередь всегда полностью перераспределяется по нодам (broadcast). Задания включаются в распределённый план выполнения и выполняются в 2 этапа.

На первом этапе на каждый из 4 шардов обеих таблиц (для обслуживания используется отдельное libpq-соединение) приходит запрос вида:

Самое важное в этих запросах:

Внутри это приводит к вызову функции worker_partition_query_result. Задача этой функции — получить результаты выполнения запроса и записать их в локальные (для ноды) файлы, согласно схеме шардирования и ключу шардирования. Для получения и записи результатов в промежуточные файлы используются экземпляры DestReceiver. Для каждого файла с результатами создаётся свой получатель. При этом общее число файлов для одного шарда задаётся значением параметра Merge Task Count, в нашем случае равным 12. По готовности получателей запускается портал (PortalRun), который и исполняет запрос.

После выполнения этой операции для 1 из 4 шардов содержимое каталога pgsql_job_cache для соответствующей распределённой транзакции будет примерно таким:

Таким образом, для одной таблицы, участвующей в процессе repartitioning, число промежуточных файлов для данного этапа будет равно Map Task Count (Table Shard Count) * Merge Task Count.

При этом обратите внимание, что для одной и той же таблицы можно увидеть такое содержимое каталогов для 2 узлов:

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

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

содержание промежуточного каталога для таблицы выше будет таким:

Для соединения с другими узлами используется libpq-протокол и команда COPY. На этом этапе, собственно, и происходит соединение по уже ставшим локальными данным.

Далее эти результаты используются уже знакомой нам функцией read_intermediate_result.

Схематично описанный процесс выглядит следующим образом.

Мы подошли к интересной и сложной теме видимости изменений разными транзакциями, блокировкам и самим распределённым транзакциям. Тема обширная, с массой нюансов. В рамках этой статьи я хотел бы показать основные отличия на примере пары запросов.

Для начала общая архитектура Citus в этой части:

Citus для соединений между узлами использует стандартный libpq-протокол и, соответственно, libpq-соединения. Соединения кешируются, переиспользуются; для достижения лучшей степени параллелизации при исполнении команд, при возможности, используются несколько соединений с одним узлом.

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

В противном случае другие команды в рамках этой же транзакции не смогли бы увидеть ещё не закоммиченные данные. Схожее требование возникает и при записи в справочную (reference) таблицу, после которой выполняются команды чтения из этой таблицы, в том числе и в операциях соединения с распределёнными таблицами (distributed). Это не всегда возможно в принципе. В документации приведен пример такого случая, если GUC citus.multi_shard_modify_mode не задан как sequential:

Операция DELETE (строка, отмеченная цифрой 2) в рамках общей транзакции с несколькими командами (строка 1) затрагивает несколько шардов. Если эта операция будет выполняться параллельно, то будут задействованы несколько соединений. Таким образом, с учетом требования по reference-таблицам становится невозможным использовать одно соединение для команды join (строка 3).

В целом, Citus вынужден аккуратно обращаться с соединениями по многим аспектам: с одной стороны — контролировать число параллельных соединений, чтобы не перегрузить ноды; с другой стороны — соблюдать правила видимости данных и не допускать взаимных блокировок. За последние два аспекта отвечает отдельный модуль Placement connection tracking. В его задачи входит отслеживание того, какие группы шардов затрагивались операциями DML/DDL/SELECT данной транзакцией, и принятие решения, какие соединения использовать для дальнейших операций.

Модуль руководствуется такими правилами:

В Greenplum для этих целей реализован механизм читающих и пишущих процессов в рамках группы связанных процессов (gangs). При создании соединений между QE и QD последний задаёт параметр, который определяет, будет ли процесс читающим или пишущим. Допускается только один пишущий процесс на группу связанных процессов, он же и будет выполнять задачи записи в рамках распределённой транзакции. Таким образом, только этот процесс может менять состояние БД и, как правило, он же отвечает и за управление блокировками в рамках транзакции (LockHolder). Мне такое чёткое разделение представляется более прозрачным подходом.

Касательно протоколов взаимодействия между нодами. В Greenplum протокол libpq пропатчен, в частности, под распределённую обработку (в основном изменения касаются взаимодействия QD и QE). Интерфейс обработки команд, полученных по libpq, расширен, например, такими командами как:

Для интерконнекта используется несколько вариантов: tcp, ufpifc и ic proxy. Центральным элементом такого взаимодействия являются узлы Motion, которые и отвечают за передачу данных между сегментами с учётом их вариантов в виде Gather Motion, Redistribute Motion и Broadcast Motion.

Таким образом, Citus полагается полностью на общепринятый libpq, в то время как Greenplum использует свой вариант протокола libpq для отправки запросов на сегменты, а за перемещение данных отвечает уже один из вариантов интерконнекта. Своя реализация протокола означает сложности с вливанием изменений апстрим-версии, но в целом это касается общего недостатка Greenplum, как существенно "допиленного" под MPP PostgreSQL.

Если сравнивать узлы Motion с их ближайшими концептуальными аналогами в виде описанных выше реализаций fetch_intermediate_results, read_intermediate_result и worker_partition_query_result, то в Greenplum отсутствует перекладывание данных между файлами с узла на узел, использование команды COPY и т.п. Мотивация такого решения создателями Citus понятна — максимально воспользовались имеющимся в PostgreSQL инструментарием. Насколько эффективна такая реализация — вопрос открытый. Мне представляется, что передача узлами Motion данных напрямую "в сеть" является более эффективным подходом.

С точки зрения аналогов параллелизации исполнения запросов, ближайшим аналогом слайсов Greenplum — независимо выполняемых частей плана, "мостиком" между которыми как раз выступают разного рода узлы Motion — как мне представляется, являются задачи (Tasks) и задания (Jobs).

Для пишущих транзакций, которые затрагивают шарды на разных нодах Citus использует PostgreSQL-реализацию двухфазного коммита (2PC) с некоторыми доработками под распределённую среду (связанными с HA и авто-восстановлением 2PC-транзакций после сбоя).

Сам подход и частности его реализации в Citus описаны в массе статей. В его основе выражения PREPARE TRANSACTION, COMMIT PREPARED и ROLLBACK PREPARED, которые PostgreSQL предлагает для внешних менеджеров транзакций. Я же предлагаю сравнить и подсветить преимущества реализации схожего, но, по факту, более продвинутого механизма в Greenplum.

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

Чтобы продемонстрировать возможные последствия подобной ситуации, я добавил задержку в блок кода, который отправляет команды COMMIT PREPARED на рабочие ноды согласно реализации протокола 2PC. Последовательность запросов такая:

Далее отправляем запросы INSERT:

Но, так как в рамках COMMIT (строке, отмеченной цифрой 2) между этими двумя транзакциями у нас добавлена искусственная (но возможная в реальности) задержка, то в параллельной сессии становится возможным получение таких данных:

Проблема в строке, отмеченной цифрой 1, где мы видим, что SELECT-запрос в параллельной сессии выдал строку (3, 'foo', 3), но не выдал (4, 'foo', 4).

Таким образом, отсюда следует очевидный вывод — Citus в рамках распределённой СУБД согласно модели консистентности распределённых систем попадает в классификацию Read Uncommitted.

Причина такого поведения данной СУБД в том, что в Citus отсутствует понятие единого снимка (имеется ввиду Snapshot для реализации MVCC в PostgresSQL) для распределённых транзакций.

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

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

В Greenplum такая ситуация не допускается. Ядро системы отслеживает информацию об активных распределённых транзакциях. Для этого в общеиспользуемой памяти процессов каждого из бэкендов хранится идентификатор распределённой транзакции (gxid) по аналогии с обычными идентификаторами локальных транзакций PostgreSQL (xid). Этот идентификатор присваивается, если транзакции требуется 2PC (т.е. транзакция является распределённой). За генерацию и присваивание значения gxid отвечает координатор запроса (QD).

Как уже говорилось ранее, в Greenplum расширен набор команд, которые обрабатываются бэкендами. В частности, командами 'M' и 'T'. Обработчики этих команд в том числе принимают сериализованную информацию для распределённой обработки транзакций от QD. Эти данные включают в себя и глобальный (распределённый) снимок.

Greenplum расширяет PostgreSQL-структуру для хранения снимков с целью хранения информации по глобальному снимку. Далее этот снимок используется для проверки видимости строк. Как говорилось ранее, только один процесс в gang может быть пишущим (writer) — таким образом, читающие (reader) процессы полагаются на то, что снимок получил пишущий процесс. Это в том числе решает проблемы видимости строк, затронутых пишущей частью транзакции (т.е. то, что Citus приходится решать манипуляциями с тем, какие libpq-соединения можно использовать).

Упрощённая модель проверки видимости строки с использованием глобальных снимков следующая.

Если идентификатор xmin данной строки соответствует транзакции, которая является применённой (правила видимости для ещё не применённых и frozen-транзакций оставим в данном случае за скобками), то далее Greenplum проверяет видимость с учётом снимка видимости для распределённой транзакции.

На примере запроса, который запускался ранее, Greenplum производит проверку, является ли данная транзакция распределённой и нужно ли проверять по глобальному снимку видимости. В нашем случае распределённая транзакция ещё не имела статуса завершённой, поэтому данная строка не является видимой для запроса в другой сессии.

Citus

Citus следует таким правилам исполнения и получения результирующего значения функций:

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

Citus строит такой план:

Согласно правилам вызова функций, он вычисляет условие в WHERE (строка, отмеченная цифрой 4) для каждого шарда (строка с номером 1). Таким образом, теоретически возможна следующая ситуация:

Допустим, в распределённой таблице d хранится 2 строки. На первом шарде Shard 1 значение атрибута равно 0, на втором шарде Shard 2 значение равно 1. В справочной таблице хранятся такие же 2 строки.

Согласно плану выполнения выборка из таблицы r производится на каждом шарде согласно совпадению вычисленного значения round(random()), т.е. это либо 0, либо 1. С некоторой вероятностью возможно получение таких кортежей на выходе узла Seq Scan (строка 3): на первом Shard 1 получена только строка с ключом 1, на втором с ключом 0.

Однако планировщик сделал pushdown соединения Hash Join на каждый шард (строка 2). Очевидно, что в данной комбинации ключей пользователь с некоторой долей вероятности получит результат, который бы он никогда не получил, например, для обычного PostgreSQL!

Так как "суперпозицией" проекций r для обоих шардов является множество { 1, 0 }, которое бы сматчилось по ключам со строками 0 и 1 распределённой таблицы d, в нашем случае дало пустое множество после объединения результатов с шардов:

Greenplum

В свою очередь Greenplum планирует запрос таким образом, чтобы соответствовать критериям "идентичности результатов" нераспределённому хранилищу. Важный момент, что выборку с волатильной функцией random() в рамках запроса к реплицируемой (аналог таблицы Citus reference) он строит единожды и распределяет по сегментам через Redistribute Motion (строка, отмеченная цифрой 2).

Таким образом, Greenplum может обеспечить корректность соединения Hash Join (строка 1), тем самым соответствуя возможным вариантам выборки нераспределённой СУБД:

Так как результаты выборки Seq Scan (строка 3) будут идентичны для всех сегментов, то каждый сегмент сможет сматчить каждый по своему ключу и дать корректный результат:

По итогу, в Citus в такой комбинации используемых таблиц есть некоторая вероятность получить некорректные с точки зрения соответствия нераспределённой СУБД результаты выборки.