ggrebalance: Часть 1. Shrink

Для понимания, что ggrebalance делает во время шринка, полезно будет в общих чертах рассмотреть действия gpexpand по добавлению новых сегментов (Рис. 7). На первом этапе gpexpand подготавливает новые сегменты: берет блокировку на возможность обновления каталога, тем самым предотвращая создание таблиц на старом числе сегментов, переносит бэкап координатора в места расположения новых сегментов (каталог координатора становится "шаблоном" для новых сегментов), поднимает их, обновляет gp_segment_configuration (на число сегментов из этого отношения ориентируются все подсистемы, реализующие распределенную работу), отпускает блокировку каталога, и сохраняет в служебной таблице список существующих отношений на основе снимка pg_class всех баз данных.

Наконец, синхронизирует зеркала вновь добавленных сегментов.

На втором этапе происходит перераспределение данных через ALTER TABLE …​ EXPAND: для каждой таблицы из подготовленного ранее списка создается временное отношение с обновленным распределением, на старом наборе сегментов запускается сканирование исходной таблицы и ее перераспределение на все сегменты со вставкой во временное отношение. Чтобы не менять свойства старого отношения и просто подвязать к нему перераспределенные данные, ALTER TABLE меняет местами relfilenode нового отношения с relfilenode старого в pg_class, и в конце временное отношение удаляется. Такой метод перераспределения данных называется CTAS-подход, и является достаточно надежным способом реорганизации данных в Greengage (пользователи могут перераспределять свои таблицы через команду ALTER TABLE …​ SET WITH (reorganize=true)). Однако у CTAS-метода есть некоторые недостатки. Во-первых, в данном случае происходит дублирование данных отношения — для больших кластеров это означает кратное увеличение требований к дисковому пространству. Во-вторых, создание нового отношения всякий раз увеличивает счетчик OID, тем самым приближаясь к oid wraparound, что не особо критично, но хотелось бы этого избежать. Условно продемонстрировать создание промежуточного отношения в CTAS-подходе можно, вручную создав таблицу на меньшем числе сегментов и выполнив CTAS во временную таблицу:

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

И третий недостаток: в конце ALTER TABLE …​ EXPAND происходит полная перестройка индексов. Это дополнительная нагрузка на I/O и время операции, пропорциональное числу и размерам индексов.

Для шринка, в свою очередь, также используются принципы перераспределения: целевое число сегментов ниже, чем текущее, и перераспределение должно закончиться до обновления gp_segment_configuration. Это означает, что во время перераспределения кластер формально все еще располагает старым числом сегментов, и все механизмы маршрутизации запросов работают по-прежнему.

В ggrebalance для шринка таблиц применяется альтернативный подход, которым было доработано ядро СУБД — перераспределение данных через целевой INSERT, реализованный во введенной команде ALTER TABLE <name> REBALANCE <target_segment_count>. Она покрывает как случай расширения (с CTAS в релизе 1.0), так и шринк, о котором идет речь в данной статье. В случае сокращения числа сегментов вместо создания промежуточного отношения, INSERT-подход просто переливает данные с выводимых сегментов на оставшиеся посредством выполнения запроса (планировщик был также доработан для поддержки случая шринка, вне ALTER TABLE …​ REBALANCE план будет отличаться):

При реализации команды ALTER TABLE …​ REBALANCE добавлены изменения в планировщике для случая шринка. Во-первых, изменено количество сегментов распределения при вставке на целевое, чтобы создать конфликт распределений отношения вставки и подплана INSERT. Планировщик, анализируя конфликт, добавляет Redistribute Motion. Во-вторых, чтобы напрямую диспетчеризировать сканирование t1 на сегменты шринка, добавляется предикат gp_segment_id IN ({set of shrunk content IDs}), указывающий на место выполнения сканирования. Тогда первый слайс выполняется на выводимых сегментах, а корневой слайс, где выполняется вставка, порождается уже на целевых сегментах. Таким образом, решение проблемы перераспределения со стороны INSERT позволяет избежать ненужного промежуточного дублирования данных (кроме, очевидно, кортежей с удаляемых сегментов) и перестроения индексов.

Если сравнивать два подхода, то разница видна уже на концептуальном примере. Пусть имеется кластер с 6 сегментами, и происходит шринк до 4. Для таблицы размером в 600 ГБ, равномерно распределенной по 6 сегментам (~100 ГБ на сегмент), при шринке CTAS-подход создал бы временную копию всей таблицы (600 ГБ новых данных на диске). INSERT-подход переместит лишь строки с сегментов 4 и 5, то есть около 200 ГБ. Временно (до вывода сегмента из кластера) эти данные будут присутствовать в двух экземплярах, но расходуя гораздо меньше ресурсов.

Упомянутый ранее gpshrink применяет простой, императивный подход к реализации (написанный одним Python-файлом по аналогии с gpexpand). Используется простой if/elif/else control flow, прогресс по сути нумерованный список шагов, без формальной валидации допустимых переходов между ними. ggrebalance, в свою очередь, вводит больше контроля над процессом шринка. Операция шринка в ggrebalance реализована в виде детерминированного конечного автомата (finite-state machine, FSM) на основе Python-библиотеки transitions с различными типами состояний: эфемерными (нигде не сохраняются, нужны для общей реализации переходов, но не важны для самой логики шринка), основными состояниями операции (сохраняются в базе, упорядочены), состояниями отката (rollback). Библиотека позволяет определить правила смены состояний и соблюдать их при выполнении. Также библиотека предоставляет гибкий набор хуков и коллбэков, которые можно вызывать перед сменой состояний или после; они могут быть привязаны к определенному состоянию, определять дополнительные условия перехода, сценарий при возникновении исключений и т.д. То есть такого инструментария более чем достаточно для описания сложных процессов.

Частичная архитектура без перечисления всех состояний и переходов изображена на Рис. 8. ggrebalance построен вокруг нескольких вложенных конечных автоматов, такая архитектура позволяет четко разделить ответственность между уровнями и хранить состояние каждого из них в базе данных для возобновления после прерывания.

Классы GGRebalanceMainSM и GGShrink описывают явные графы состояний и переходов между ними. Каждый переход сохраняется в служебной схеме (ggrebalance, Рис. 9) внутри базы данных postgres. Именно это обеспечивает реентерабельность: при повторном запуске после любого прерывания утилита считывает последнее зафиксированное состояние и возобновляет выполнение со следующего шага.

Например, для сценария шринка фиксация статусов может выглядеть так:

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

Рассмотрим ключевые этапы потока шринка подробнее.

Рассмотрим практический пример из рисунка 6: кластер из 3 хостов с 12 primary-сегментами (по 4 на хост), и нам требуется уменьшить их количество до 9 (шринк) и сбалансировать до 3 сегментов на хост. Стратегию зеркалирования не меняем (grouped).

Важное замечание: шринк кластера подразумевает вывод сегментов, чьи content_id образуют хвост последовательности всех content_id после отбрасывания из нее <target_segment_count>. Произвольный сегмент, ввиду специфики хеширования Greengage, выкинуть не получится — только с конца. Команда запуска:

Перед стартом ggrebalance убеждается в отсутствии конкурирующих утилит (gpexpand, gprecoverseg, gpbackup и т.д.), проверяет, что все primary-сегменты доступны, и создает PID-файл для предотвращения параллельного запуска. Далее выполняются следующие шаги:

Примерный вывод следующий:

Теоретическое преимущество INSERT-подхода перед CTAS заключается в том, что объем перемещаемых данных пропорционален доле удаляемых сегментов, а не полному размеру таблиц. Для проверки этого утверждения был проведен ряд экспериментов по оценке скорости шринка.

Первая серия измерений посвящена сравнению скорости выполнения ALTER TABLE …​ REBALANCE, основанной на логике INSERT, и скорости выполнения этой же команды, реализованной через CTAS. В облаке был развернут кластер Greengage, состоящий из 8 хостов и 64 primary-сегментов. Характеристики сегмент-хостов и настройки кластера вкратце приведены в таблице 1.

Рассматривались следующие сценарии:

Кластер заполнялся адаптированным для Greengage DB датасетом TPC-DS размером около 2 ТБ (генерация с scale factor = 3000).

Для целостной картины измерялось время выполнения шринка и степень разрастания дискового пространства, или дисковая амплификация (увеличение занимаемого места относительно логического объема самих данных), для следующих таблиц (все партицированные):

Для каждой из 5 таблиц трижды выполнялся ее шринк CTAS и INSERT-методами — всего 30 прогонов. После каждого шринка таблица восстанавливалась на исходное число сегментов также через ALTER TABLE …​ REBALANCE <nsegs_origin>, вызывающий CTAS перераспределение, то есть шаг восстановления после шринка вносит одинаковый шум в возможные результаты прогонов. Измерялись величины:

Теоретическая оценка степени разрастания основана на предположении, что данные таблиц равномерно распределены по всему кластеру. Для CTAS-метода ожидается, что в какой-то момент появится полный дубликат отношения. А для INSERT — лишь доля объема, пропорциональная количеству сегментов, на которых происходит сканирование.

Стоит отметить, что это не полноценное нагрузочное тестирование, а частичная оценка производительности с использованием инструментов и ресурсов, доступных рядовым разработчикам. Подробный анализ производительности на околопродовых данных (~600 первичных сегментов) будет опубликован в будущем.

Ниже приведены результаты измерений.

На основе полученных результатов можно сделать следующие выводы. Во-первых, теория CTAS "полная копия таблицы" подтвердилась экспериментально. Степень разрастания можно вычислить как по колонкам размеров отношений, так и по значениям размеров всего кластера. Совпадение теории и практики почти полное. Небольшие отклонения объясняются частотой опрашивания хостов (получения размеров первичных сегментов, каждые 10 секунд), пиковый момент мог попасть между двумя опросами.

Во-вторых, можно наблюдать ускорение INSERT по сравнению с CTAS (таблица 5).

CTAS: Рост пропорционален целевому числу сегментов: 56, 32, 16. Чем оно меньше, тем дольше выполняется перераспределение, время складывается из двух независимых I/O-фаз примерно равного веса. Для INSERT картина аналогичная: время масштабируется линейно с объемом перемещаемых данных, здесь же большее влияние вносит нагрузка на сетевое перераспределение кортежей.

Таким образом, по дисковой амплификации INSERT выигрывает при любом сценарии без исключений. Разница наиболее показательна при 64 → 56, где INSERT создает в 7—​8 раз меньше пикового дискового давления, чем CTAS. По времени INSERT выигрывает при 64 → 56 и 64 → 32 для всех таблиц с большим отрывом. При 64 → 16 преимущество INSERT сокращается до 20—​30% на больших таблицах и исчезает на маленьких AO-таблицах без компрессии. Предположительно, с увеличением перераспределяемых данных растет объем генерируемых WAL, так как INSERT вставляет данные построчно через стандартный путь heap_insert / appendonly_insert, который при wal_level >= replica пишет WAL для каждого блока. При перемещении 50% данных store_sales (232 ГБ) это колоссальный объем WAL. Также при INSERT в существующую AO-таблицу каждый сегмент должен обновлять строки в pg_aoseg — системный каталог, который хранит метаданные AO-сегментных файлов. При множестве параллельных вставок возникает конкуренция за доступ к этой системной таблице.

Также можно грубо оценить пропускную способность (throughput) в ГБ/с по формуле:

CTAS при 64 → 56 имеет наивысший throughput 0.154 ГБ/с — потому что все 56 целевых сегментов пишут параллельно и объем записи велик. При уменьшении целевого числа сегментов throughput деградирует. Причина в том, что формула учитывает только объем данных, записываемых в новую таблицу, тогда как реальное время определяется еще и чтением полного объема исходной таблицы, которое при агрессивном шринке становится доминирующей фазой. При 64 → 16 CTAS читает 100% данных, но пишет лишь 25% — оставшиеся 75% прочитанных данных просто перераспределяются на меньшее число сегментов, и именно этот дисбаланс между объемом чтения и записи объясняет столь низкий эффективный throughput CTAS при агрессивном шринке.

Проведенные эксперименты дают количественную основу для сравнения двух фундаментально различных стратегий перераспределения данных в Greengage-кластере. У нас есть измеренные точки: INSERT-подход выигрывает по дисковой амплификации при любом сценарии и по времени при умеренном шринке, но теряет временно́е преимущество при агрессивном сжатии из-за роста WAL-нагрузки и contention на метаданных AO-хранилища. CTAS-подход проигрывает по пиковому росту занимаемой памяти, но имеет меньше накладных расходов при агрессивном шринке. В следующих релизах определенно планируется оптимизация этих моментов.

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

Благодаря персистентному конечному автомату повторный запуск ggrebalance после любого прерывания (сбой сети, нехватка дискового пространства, сигнал SIGINT и т.д.) безопасен. Утилита считывает последнее зафиксированное состояние FSM и сравнивает его с реальным состоянием кластера:

На уровне таблиц каждая запись в служебной очереди имеет статус: none — не обработана, done — обработана. При возобновлении пул воркеров забирает только необработанные таблицы, уже перераспределенные пропускаются. Это исключает двойную работу даже при многократных прерываниях в середине шага SHRINK_TABLES.

Раскроем необходимость обеспечения реентерабельности на примере. Допустим, был запущен процесс шринка (все тот же Рис. 6), который начал перераспределять данные. И в это время произошел некритичный инцидент, затрагивающий работу кластера Greengage, после которого появилась необходимость перезапустить систему и прервать шринк. ggrebalance может быть сам по себе прерван через:

Пусть шринк был прерван, находясь в следующем состоянии:

Из лога видно, что таблица t1 была успешно обработана, после чего ggrebalance приступил к t2 и не смог завершить ее — произошло прерывание. Это подтверждается и таблицей table_rebalance_status_detail: t1 имеет статус done с временной меткой завершения, тогда как таблица t2 осталась в статусе none, то есть ее перераспределение так и не было зафиксировано как начатое. Таблица rebalance_status показывает, что последним сохраненным состоянием машины является STATE_SHRINK_TABLES_STARTED — машина вошла в цикл обработки таблиц, но не успела из него выйти. Казалось бы, достаточно просто возобновить работу с того же состояния и перераспределить оставшуюся t2. Однако здесь кроется первая особенность реализации операции шринка, когда в состоянии STATE_BACKUP_CATALOG_AND_UPDATE_TARGET_SEGMENT_COUNT_STARTED, которое предшествовало шринку таблиц, ggrebalance выполнил вызов gp_toolkit.gp_set_rebalance_numsegments(target_count) — установил в каталоге целевое количество сегментов. Этот параметр является сессионно-независимым и глобальным. После перезагрузки кластера этот параметр сбрасывается в значение по умолчанию — полный набор сегментов. Это означает, что если в промежутке между прерыванием шринка и его возобновлением в кластере были созданы новые таблицы, они окажутся распределены по полному набору сегментов, а не по целевому. Такой сценарий достаточно опасен для целостности шринка, чреват потерей данных с выводимых сегментов.

ggrebalance учитывает такую ситуацию наряду с другими краевыми случаями, гарантируя, что сегменты не будут выведены, пока все таблицы не перераспределятся. ggrebalance не продолжает с точки прерывания, а делает шаг назад и восстанавливает rebalance_numsegments для параллельных DDL-запросов, затем заново обходит все базы данных и формирует актуальную очередь, отбирая таблицы по условию распределения. Без описанной логики любой перезапуск кластера в ходе шринка превращался бы в инцидент: новые таблицы оставались бы "за бортом" операции, а следующий запуск ggrebalance либо завершал бы шринк, игнорируя их, либо требовал бы ручного анализа расхождений. В этом отношении gpshrink подвержен риску потери данных при параллельных операциях с кластером.

Операция отката шринка (ggrebalance --rollback) доступна только до момента обновления gp_segment_configuration — то есть не позднее состояния STATE_SHRINK_TABLES_DONE. Если каталог уже обновлен, откат невозможен по определению — кластер работает с новым числом сегментов. При допустимом откате запускается поток отката, который:

Откат сам по себе полностью реентерабелен: каждый шаг персистируется в том же потоке состояний states_rollback_flow, и повторный запуск ggrebalance --rollback корректно продолжит прерванный откат. При этом уже обработанные при откате таблицы (их статус обновлен обратно на none) не затрагиваются повторно. Таким образом, оба крайних сценария — "операция прервана, хочу продолжить" и "операция прервана, хочу вернуть все как было" — обрабатываются ggrebalance детерминировано и без ручного вмешательства в состояние кластера.

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