В 2021 году мы запустили Greenplum в нашем облаке. И очень скоро столкнулись с тем, что эластичность систем расчёта и хранения — это must have в облачных аналитических БД. А Greenplum — совсем не такой.

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

А что в других облаках?

Единого подхода в этом вопросе нет. Под спойлерами приведу краткие детали для разных облаков, а также ссылки на источники, где можно найти подробности.

Кроме коммерческих разработок есть научные, например, AnyBlob:

• Exploiting Cloud Object Storage for High‑Performance Analytics. Dominik Durner, Viktor Leis Technische, Thomas Neumann.

Общее описание нашей разработки

Наше решение архитектурно больше похоже на Snowflake+AnyBlob.

В Greenplum мы:

• пропатчили интерфейс smgr и сделали его расширяемым;

• расширили интерфейс smgr и добавили в него возможность работы с Object Storage S3;

• сделали extension yezzey для управления выгрузкой/загрузкой из Object Storage;

• написали специальный сервис YProxy для проксирования запросов от Greenplum к хранилищу S3.

Результат назвали Hybrid Storage в Greenplum (внутреннее название — yezzey). Суть — в возможности выгрузить данные в S3 без существенной деградации производительности запросов и изменений интерфейсов работы.

Интерфейс

Hybrid Storage сейчас работает только с Append Only (AO/AOCO) таблицами Greenplum. Предположим, у вас есть таблица:

postgres=# create table test(i int, j int, k int, kk int) with(appendonly=true, orientation=column) DISTRIBUTED RANDOMLY;
CREATE TABLE
postgres=# insert into test select * from generate_series(1, 100) a join generate_series(1, 100) b on true join generate_series(1, 100) c on true join generate_series(1, 100) d on true;
INSERT 0 100000000
postgres=# select count(1) from public.test;
   count
-----------
 100000000

Time: 5762.468 ms

Выгрузка данных в S3 производится вызовом метода yezzey_define_offload_policy(reloid OID, remove_locally BOOLEAN)  из extension yezzey:

postgres=# select yezzey_define_offload_policy('public', 'test');
NOTICE:  yezzey: relation virtual size calculated: 0  (seg0 slice1 10.129.0.12:6000 pid=706966)
NOTICE:  yezzey: relation virtual size calculated: 0  (seg1 slice1 10.129.0.30:6000 pid=707950)
INFO:  yezzey: relation segment reached external storage (blkno=1), up to logical eof 200242112  (seg0 slice1 10.129.0.12:6000 pid=706966)
INFO:  yezzey: relation segment reached external storage (blkno=1), up to logical eof 200295736  (seg1 slice1 10.129.0.30:6000 pid=707950)
NOTICE:  yezzey: relation virtual size calculated: 0  (seg1 slice1 10.129.0.30:6000 pid=707950)
NOTICE:  yezzey: relation virtual size calculated: 0  (seg0 slice1 10.129.0.12:6000 pid=706966)
INFO:  yezzey: relation segment reached external storage (blkno=129), up to logical eof 200295736  (seg1 slice1 10.129.0.30:6000 pid=707950)
NOTICE:  yezzey: relation virtual size calculated: 0  (seg1 slice1 10.129.0.30:6000 pid=707950)
INFO:  yezzey: relation segment reached external storage (blkno=129), up to logical eof 200242112  (seg0 slice1 10.129.0.12:6000 pid=706966)
NOTICE:  yezzey: relation virtual size calculated: 0  (seg0 slice1 10.129.0.12:6000 pid=706966)
INFO:  yezzey: relation segment reached external storage (blkno=257), up to logical eof 200295736  (seg1 slice1 10.129.0.30:6000 pid=707950)
NOTICE:  yezzey: relation virtual size calculated: 0  (seg1 slice1 10.129.0.30:6000 pid=707950)
INFO:  yezzey: relation segment reached external storage (blkno=257), up to logical eof 200242112  (seg0 slice1 10.129.0.12:6000 pid=706966)
NOTICE:  yezzey: relation virtual size calculated: 0  (seg0 slice1 10.129.0.12:6000 pid=706966)
INFO:  yezzey: relation segment reached external storage (blkno=385), up to logical eof 200295736  (seg1 slice1 10.129.0.30:6000 pid=707950)
INFO:  yezzey: relation segment reached external storage (blkno=385), up to logical eof 200242112  (seg0 slice1 10.129.0.12:6000 pid=706966)
 yezzey_define_offload_policy
------------------------------

(1 row)

Time: 63464.499 ms

postgres=# select count(1) from public.test;
   count
-----------
 100000000
(1 row)

Time: 6331.992 ms

postgres=# select offload_reloid, segindex, segfileindex, external_storage_filepath, external_bytes from yezzey_relation_describe_external_storage_structure('test');
 offload_reloid | segindex | segfileindex |                                                   external_storage_filepath                                                    | external_bytes
----------------+----------+--------------+--------------------------------------------------------------------------------------------------------------------------------+----------------
          32176 |        1 |            0 | wal-e/6/segments_005/seg1/basebackups_005/yezzey/1663_12813_81b5e60c711c42d92d6115c2140f6be4_27082_1__DY_1_xlog_279307134824   | 200296702
          32176 |        1 |            0 | wal-e/6/segments_005/seg1/basebackups_005/yezzey/1663_12813_81b5e60c711c42d92d6115c2140f6be4_27082_129__DY_1_xlog_279307157504 | 200296702
          32176 |        1 |            0 | wal-e/6/segments_005/seg1/basebackups_005/yezzey/1663_12813_81b5e60c711c42d92d6115c2140f6be4_27082_257__DY_1_xlog_279307179720 | 200296702
          32176 |        1 |            0 | wal-e/6/segments_005/seg1/basebackups_005/yezzey/1663_12813_81b5e60c711c42d92d6115c2140f6be4_27082_385__DY_1_xlog_279307180720 | 200296702
          32176 |        0 |            0 | wal-e/6/segments_005/seg0/basebackups_005/yezzey/1663_12813_81b5e60c711c42d92d6115c2140f6be4_27082_1__DY_1_xlog_279307134824   | 200243079
          32176 |        0 |            0 | wal-e/6/segments_005/seg0/basebackups_005/yezzey/1663_12813_81b5e60c711c42d92d6115c2140f6be4_27082_129__DY_1_xlog_279307157504 | 200243079
          32176 |        0 |            0 | wal-e/6/segments_005/seg0/basebackups_005/yezzey/1663_12813_81b5e60c711c42d92d6115c2140f6be4_27082_257__DY_1_xlog_279307179720 | 200243079
          32176 |        0 |            0 | wal-e/6/segments_005/seg0/basebackups_005/yezzey/1663_12813_81b5e60c711c42d92d6115c2140f6be4_27082_385__DY_1_xlog_279307180720 | 200243079
(8 rows)

Сам объект остаётся неизменным, дальнейшая работа с ним продолжается так же, как если бы данные находились на локальном диске.

Для выгрузки данных:

1. Берётся эксклюзивная блокировка на соответствующий AO/AOCS‑объект из pg_class (как при DDL‑операциях, например, exchange partition).

2. Файлы данных таблицы на всех сегментах выгружаются в S3, последовательно один за другим.

3. Файлы в процессе выгрузки пакуются и шифруются PGP.

4. В метаданных таблицы изменяется табличное пространство на Yezzey (виртуальное табличное пространство с oid=8555).

5. Выполняется commit.

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

Сообщение вида yezzey: relation virtual size calculated <число> показывает размер выгруженных ранее в S3 объектов (0 означает, что данные выгружаются впервые).

Сообщение вида yezzey: relation segment reached external storage (blkno=385), up to logical eof 200242112 (seg0 slice1 10.129.0.12:6000 pid=706966) говорит об успешной выгрузке данных в S3 сегмента и дополнительно сообщает размер выгруженного файла — 200 242 112 байта (на самом деле, это eof).

Метод yezzey_relation_describe_external_storage_structure(i_relname TEXT) выводит справочную информацию о выгруженных файлах с сегментов в S3 в виде:

  offload_reloid — oid таблицы
  segindex — id сегмента
  segfileindex — id файла данных
  external_storage_filepath — путь файла в S3
  external_bytes — размер файла

В примере у нас таблица из четырёх колонок хранится поколоночно на двух сегментах, получаем 2×4 = 8 файлов. Название файла генерится Greenplum, включает в себя oid таблицы, колонки и последовательный индекс файла.

Исходный код и алгоритм работы

Наш код выложен на github под лицензией PostgreSQL (очень похожа на BSD). А также доступен в Yandex Cloud во всех версиях Greenplum, начиная с 6.25.

Упрощённая схема работы

БД вместо чтения/записи блока данных с локального диска выполняет чтение/запись блока с хранилища S3.

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

Как видно на схеме, основной новый компонент системы — YProxy. Это специальный процесс для проксирования запросов от Greenplum к S3. Прокси не нужна, если запросов к S3 мало. Проблемы возникают, если параллельных потоков обмена данными становится много. Greenplum выполняет чтение блоков по мере необходимости, не приоретизируя свои процессы. Теперь предположим, что мы хотим прочитать партиционированную таблицу из X партиций, хранимую поколоночно из Y колонок. В хранилище будет отправлено X*Y запросов.

Если данные хранятся локально, управлением вводом‑выводом занимается планировщик ОС. Он контролирует запросы и ставит им приоритеты. Без планировщика параллельные запросы быстро забьют storage запросами так, что общая его производительность очень сильно деградирует. Вместо очереди на обслуживание получим хаос, когда часть запросов успела быстро отработать, а часть — очень долго ждет данных. Производительность SQL‑запросов деградирует, потребление памяти растет.

YProxy здесь выступает в качестве планировщика. Все запросы от Greenplum поступают к YProxy. YProxy держит пул соединений с S3 и управляет приоритетами запросов, организует очередь, выделяет каждому запросу для работы определённый квант времени/полосы пропускания. Так что даже очень большой поток запросов от Greenplum гарантированно не приведёт к исчерпанию CPU или сети на хосте кластера. Запросы будут обрабатываться поочерёдно, и те, кто раньше начал работу, успеют сделать больше.

Алгоритм записи данных

Здесь показан пример метаданных AO‑таблицы. Таблица состоит из трёх файлов, все файлы — разной длины, каждый файл был модифицирован несколько раз. Стрелка от «1» к файлу «32 776.1» символизирует бегущую транзакцию: текущий Logical EOF указывает на середину файла. Новая транзакция дописывает данные в конец файла, но пока что не зафиксировала свои изменения, так что Logical EOF меньше реального размера файла. По успешному завершению транзакции Modcount и Logical EOF увеличат свои значения.

Алгоритмы записи данных в yezzey:

1. Пусть есть табличка, в которой есть транзакция как на рисунке выше, и мы стартуем ещё одну.

2. Выбираем себе блок‑файл, в который будем писать (стандартным для Greenplum образом).

3. Вставляем/обновляем запись в таблицу метаданных pg_aoseg, тем самым блокируем слот.

4. Пишем данные в файл.

5. Для yezzey данные пишутся не локально, а сразу в S3 потоковой записью через put.

6. В отличие от локального хранения, данные в S3 не дописываются в файл, а пишутся в новый файл (через multipart upload).

7. Метаданные об этом новом файле сохраняются кроме pg_aoseg ещё и в таблице yezzey vitrual index (пример содержимого таблицы приведён на рисунке ниже).

8. Во время завершения statement мы дожидаемся завершения multipart upload.

9. По завершении upload обновляются метаданные об Logical EOF в pg_aoseg и yezzey vitrual index.

10. По commit метаданные становятся видимыми всем транзакциям: можно читать до нового Logical EOF.

Алгоритм чтения данных

Это пример содержимого yezzey virtual index для файла 32 776.2. Локальный файл — один, но был сформирован двумя транзакциями. Этим двум транзакциям соответствует два файла в S3, и две записи про них в таблице yezzey_virtual_index. Сумма EOF по двум этим файлам равна Logical EOF в таблице pg_aoseg.

Алгоритм чтения данных в yezzey:

1. Читаем текущий снапшот yezzey_virtual_index.

2. В снапшоте упорядочиваем файлы по логическим EOF.

3. Начинаем последовательное чтение: отдаём Executor по 32 Кб (blocksize) из текущего файла, по окончании текущего начинаем стримить из следующего файла.

4. Чтение выполняем через YProxy.

5. Читаем с ретраями, чтобы переживать недоступность YProxy/S3.

6. Каждый AO‑файл имеет logical EOF (pg_aoseg), но хранится в нескольких файлах в S3 (yezzey vitrual index). Мы читаем эти файлы из S3 последовательно пока не прочитаем logical EOF байт.

7. В последнем файле после EOF может хранится ещё что‑то — мусор от незавершившейся транзакции.

8. MVCC обеспечивается блокировками записей в pg_aoseg и версионностью чтений таблицы с метаданными (таблица метаданных — обычная heap‑таблица PostgreSQL). Файлы в S3 не меняются, а всегда дописываются новые, с изменением метаданных в yezzey virtual index.

Алгоритм восстановления и удаления данных

1. Копируем весь bucket в S3 в новый кластер.

2. Пропускаем при recovery транзакций всё, что относится к S3.

3. Восстановиться можно только на точку консистентности до момента 1, так как добавление данных в S3 пропускается.

4. Все остальные метаданные восстанавливает сам Greenplum, когда проигрывает WAL для heap‑таблиц.

5. Стриминг WAL на disaster recovery кластер поддержан с ограничением: S3 должен быть доступен также и с disaster recovery кластера.

Данные в yezzey не удаляются, но мусор нужно чистить (да, когда‑то эти данные были очень нужны, но теперь это мусор). Для чистки мусора используются структуры yezzey_expire_index, вот пример данных:

postgres=# select * from gp_dist_random('yezzey.yezzey_expire_index');
 reloid | relfileoid | last_use_lsn | expire_lsn |              fqnmd5
--------+------------+--------------+------------+----------------------------------
  32176 |      27082 | 41/8015AB0   | 0/0        | 81b5e60c711c42d92d6115c2140f6be4
  32176 |      27082 | 41/8015AB0   | 0/0        | 81b5e60c711c42d92d6115c2140f6be4
(2 rows)

Структура yezzey_expire_index

В yezzey_expire_index для каждого файла в S3 указан expire_lsn. expire_lsn — минимальный lsn бекапа, которому нужен этот файл. 0/0 означает, что файл ни разу не бэкапился, он нужен всем бэкапам. Теперь если у нас есть список бэкапов кластера, и expire_lsn меньше, чем lsn начала самого старого бэкапа, то файл не нужен. Можно удалить его. Пробегаемся по всем файлам и удаляем те, которые больше не нужны.

Performance-тесты

Тестирование производительности мы решили сделать на базе открытого датасета поездок данных о поездках NY Yellow Taxi за период 2013–2022 гг (около миллиарда строк данных). Описание теста на GitHub.

Мы взяли Greenplum самой популярной в Yandex Cloud конфигурации в default‑настройках, загрузили в него данные. Также загрузили данные в csv‑файлы в S3, доступ к которым настроили через PXF. И сравнили производительность, если данные на локальных дисках, в Hybrid Storage и в PXF.

Таблицы фактов с поездками для всех движков были распределены по сегментам случайным образом (DISTRIBUTED RANDOMLY) и партиционированы по годам.

Тестовые запросы:

Q1. Количество поездок

select count(1) from ORDERS 

Q2. Количество поездок в разрезе таксопарков

select 
  vendorid, count(1) 
from 
  ORDERS 
group by 
  vendorid

Q3. Количество поездок в разрезе таксопарка, времени поездки, количества пассажиров и типа платежа

select 
  vendorid, pickup_date, passenger_count, payment_type, count(1) 
from 
  ORDERS 
group by 
  1, 2, 3, 4

Q4. Нарастающая максимальная стоимость поездки в разрезе места назначения за 2014-й год

select 
  distinct dolocationid, 
  max(total_amount) over (partition by dolocationid) 
from 
  ORDERS 
where 
  pickup_date between '2014-01-01' :: date 
  and '2014-12-31' :: date

Q5. То же, что и Q4, но за 2014-й, 2015-й и 2016-й 

select 
  distinct dolocationid, 
  max(total_amount) over (partition by dolocationid) 
from 
  ORDERS 
where 
  pickup_date between '2014-01-01' :: date 
  and '2016-12-31' :: date

Q6. Сумма поездок за 3 года в разрезе идентификатора локации из справочника локаций

select 
  z.locationid, 
  count(r.vendorid), 
  sum(r.total_amount) 
from 
  ORDERS r left join ZONES z on r.pulocationid = z.locationid 
where 
  r.pickup_date between '2014-01-01' :: date 
  and '2016-12-31' :: date 
group by 1

Q7. То же, что и Q6, но за всё время

select 
  z.locationid, 
  count(r.vendorid), 
  sum(r.total_amount) 
from 
  ORDERS r left join ZONES z on r.pulocationid = z.locationid 
group by 1

Каждый запрос выполнялся 3 раза. Результаты тестирования показаны в таблице.

Время выполнения тестовых запросов в различной конфигурации Greenplum

Когда использовать Hybrid Storage и что дальше

Тесты показали, что:

• Самый худший результат (Q1) — время выполнения запроса для Hybrid storage по сравнению с ванильным Greenplum увеличилось на 43%. Но если доступ к данным происходит по PXF, то время увеличивается в 20 раз.

• Самый лучший результат — время выполнения запроса не изменилось (Q5). При доступе по PXF для Q5 время увеличилось в 2,46 раза.

• Существенная разница во времени выполнения наблюдалась в простых запросах Q1–Q3. Они зависят в основном от производительности storage.

• Для сложных запросов Q4–Q7 увеличение времени выполнения было в пределах 10%–20%.

То есть в отличие от PXF Hybrid Storage не даёт существенного (в разы) ухудшения производительности.

Hybrid storage практично использовать для уменьшения размера кластера. Это позволяет:

• справиться с постоянным ростом размера кластера;

• снизить стоимость кластера.

Для этого можно:

1. Холодные данные, которые не используются постоянно, перенести в более дешёвое хранилище S3.

2. Данные для сложных запросов перенести в более дешёвое хранилище S3. И использовать в расчетах подход как в Snowflake, когда время выполнения запроса зависит больше от доступных ресурсов CPU и Network, чем от производительности storage.

Пока что всё. Но многое здесь ещё предстоит сделать: достаточно посмотреть, что уже сделано для аналитики в других облачных БД. Наши планы мы обсуждали на Greenplum® Community Meetup. Его запись доступна на сайте. Stay tuned.