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

Некоторые библиотеки, перечисленные в следующем разделе, могут помочь в переносе схемы базы данных путём добавления столбца распределения в ключи. Однако ниже представлен пример базовых SQL-команд для преобразования простых ключей в составные в БД разработки:

BEGIN;

-- Удаление простых первичных ключей (с каскадным удалением внешних ключей)

ALTER TABLE products   DROP CONSTRAINT products_pkey CASCADE;
ALTER TABLE orders     DROP CONSTRAINT orders_pkey CASCADE;
ALTER TABLE line_items DROP CONSTRAINT line_items_pkey CASCADE;

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

ALTER TABLE products   ADD PRIMARY KEY (store_id, product_id);
ALTER TABLE orders     ADD PRIMARY KEY (store_id, order_id);
ALTER TABLE line_items ADD PRIMARY KEY (store_id, line_item_id);

-- Пересоздание внешних ключей с будущим столбцом распределения

ALTER TABLE line_items ADD CONSTRAINT line_items_store_fkey
  FOREIGN KEY (store_id) REFERENCES stores (store_id);
ALTER TABLE line_items ADD CONSTRAINT line_items_product_fkey
  FOREIGN KEY (store_id, product_id) REFERENCES products (store_id, product_id);
ALTER TABLE line_items ADD CONSTRAINT line_items_order_fkey
  FOREIGN KEY (store_id, order_id) REFERENCES orders (store_id, order_id);

COMMIT;

После завершения схема из предыдущего раздела будет выглядеть так (подчёркнутые элементы — первичные ключи, выделенные курсивом элементы — внешние ключи):

Обязательно измените потоки данных, чтобы добавить к входящим данным ключи.

Если используется объектно-реляционное отображение (ORM), отличное от приведённых выше, или выполняются многоарендные запросы непосредственно в SQL, следуйте этим общим принципам. Далее будет использоваться предыдущий пример приложения для электронной торговли.

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

-- До
SELECT *
  FROM orders
 WHERE order_id = 123;

-- После
SELECT *
  FROM orders
 WHERE order_id = 123
   AND store_id = 42; -- <== added

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

Наконец, при соединении таблиц обязательно используйте фильтр по tenant_id. Например, можно узнать, сколько пар «отличных шерстяных штанов» было продано в данном магазине:

-- Можно включить в соединение store_id и
-- добавить фильтр по этому значению в один из запросов

SELECT sum(l.quantity)
  FROM line_items l
 INNER JOIN products p
    ON l.product_id = p.product_id
   AND l.store_id = p.store_id
 WHERE p.name='Awesome Wool Pants'
   AND l.store_id='8c69aa0d-3f13-4440-86ca-443566c1fc75'

-- Либо можно не добавлять store_id в условие соединения,
-- но отфильтровать по нему обе таблицы. Это может быть полезно
-- при построении запросов в ORM

SELECT sum(l.quantity)
  FROM line_items l
 INNER JOIN products p ON l.product_id = p.product_id
 WHERE p.name='Awesome Wool Pants'
   AND l.store_id='8c69aa0d-3f13-4440-86ca-443566c1fc75'
   AND p.store_id='8c69aa0d-3f13-4440-86ca-443566c1fc75'

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

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

Во время тестирования можно создать следующую конфигурацию:

-- Задайте нужное имя базы данных

ALTER DATABASE citus SET citus.multi_task_query_log_level = 'error';

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

Во время запуска в производство можно настроить тот же параметр для внесения записей в журнал вместо вывода ошибок:

ALTER DATABASE citus SET citus.multi_task_query_log_level = 'log';

Для получения дополнительной информации о допустимых значениях обратитесь к описанию citus.multi_task_query_log_level.

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

Следующие таблицы хорошо подходят для преобразования в таблицы-справочники:

Например, предположим, что многоарендному сайту для электронной торговли необходимо рассчитать налог с продаж для транзакций в любом из магазинов. Налоговая информация не является уникальной для какого-либо арендатора, поэтому разумно объединить её в общую таблицу. Таблица-справочник, ориентированная на США, может выглядеть так:

-- Таблица-справочник

CREATE TABLE states (
  code char(2) PRIMARY KEY,
  full_name text NOT NULL,
  general_sales_tax numeric(4,3)
);

-- Распределение этой таблицы на все рабочие узлы

SELECT create_reference_table('states');

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

Помимо распределения таблицы как единого реплицированного сегмента, функция create_reference_table помечает её как таблицу-справочник в таблицах метаданных citus. В расширении автоматически выполняется двухфазная фиксация изменений в помеченных таким образом таблицах, что гарантирует согласованность данных.

Существующую распределённую таблицу можно преобразовать в таблицу-справочник так:

SELECT undistribute_table('имя_таблицы');
SELECT create_reference_table('имя_таблицы');

Если существующую базу данных Postgres Pro преобразовать в узел-координатор для кластера citus, данные в таблицах можно распределить эффективно с минимальными помехами для приложения.

Описанная ранее функция create_distributed_table работает как с пустыми, так и с непустыми таблицами, причём в последних она автоматически распределяет строки таблицы по кластеру. Показателем проведённого распределения служит следующее сообщение: NOTICE: Copying data from local table... (ВНИМАНИЕ: Идёт копирование из локальной таблицы). Например:

CREATE TABLE series AS SELECT i FROM generate_series(1,1000000) i;
SELECT create_distributed_table('series', 'i');
NOTICE:  Copying data from local table...
NOTICE:  copying the data has completed
DETAIL:  The local data in the table is no longer visible, but is still on disk.
HINT:  To remove the local data, run: SELECT truncate_local_data_after_distributing_table($$public.series$$)
 create_distributed_table
 --------------------------

 (1 row)

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

При распределении таблиц A и B, где A ссылается на B по внешнему ключу, сначала распределите таблицу B. Если сделать это в неправильном порядке, возникнет ошибка:

ERROR:  cannot create foreign key constraint
DETAIL:  Referenced table must be a distributed table or a reference table.

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

После распределения таблиц используйте функцию truncate_local_data_after_distributing_table для удаления локальных данных. Оставшиеся локальные данные в распределённых таблицах недоступны для запросов citus и могут вызывать нарушения ограничений на узле-координаторе.

В citus большинство команд ALTER TABLE транслируется автоматически. Добавление столбцов или изменение их значений по умолчанию работает так же, как и в базе данных Postgres Pro на одном компьютере:

-- Добавление столбца

ALTER TABLE products ADD COLUMN description text;

-- Изменение значения по умолчанию

ALTER TABLE products ALTER COLUMN price SET DEFAULT 7.77;

Существенные изменения в существующем столбце, например переименование или изменение типа данных, также допустимы. Однако нельзя изменять тип данных столбца распределения. Этот столбец определяет, как данные таблицы распределяются по кластеру citus, и такое изменение потребует перемещения данных.

Попытка изменить тип данных столбца вызовет ошибку:

-- Предположим, что store_id типа integer является столбцом
-- распределения для таблицы products

ALTER TABLE products
ALTER COLUMN store_id TYPE text;

/*
ERROR:  cannot execute ALTER TABLE command involving partition column
*/

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

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

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

Внешние ключи могут создаваться в следующих случаях:

Внешние ключи от таблиц-справочников к распределённым таблицам не поддерживаются.

В citus поддерживаются все ссылочные действия с внешними ключами от локальных таблиц к таблицам-справочникам, но не поддерживаются ON DELETE/ON UPDATE CASCADE в обратном направлении (ссылки на локальные таблицы).

В примере ниже показано, как создавать первичные и внешние ключи в распределённых таблицах:

--
-- Добавление первичного ключа
-- --------------------

-- Распределите эти таблицы по account_id. Таблицы ads и clicks
-- должны использовать составные ключи, содержащие account_id.

ALTER TABLE accounts ADD PRIMARY KEY (id);
ALTER TABLE ads ADD PRIMARY KEY (account_id, id);
ALTER TABLE clicks ADD PRIMARY KEY (account_id, id);

-- Затем распределите таблицы

SELECT create_distributed_table('accounts', 'id');
SELECT create_distributed_table('ads',      'account_id');
SELECT create_distributed_table('clicks',   'account_id');

--
-- Добавьте внешние ключи
-- -------------------

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

ALTER TABLE ads ADD CONSTRAINT ads_account_fk
  FOREIGN KEY (account_id) REFERENCES accounts (id);
ALTER TABLE clicks ADD CONSTRAINT clicks_ad_fk
  FOREIGN KEY (account_id, ad_id) REFERENCES ads (account_id, id);

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

-- Допустим, в каждой рекламе должно использоваться уникальное изображение. Обратите внимание,
-- что при распределении по account_id ограничение можно установить ограничение только по учётной записи.

ALTER TABLE ads ADD CONSTRAINT ads_unique_image
  UNIQUE (account_id, image_url);

Ограничения NOT NULL можно применять к любому столбцу (распределения или нет), поскольку для них не требуется поиск между рабочими узлами.

ALTER TABLE ads ALTER COLUMN image_url SET NOT NULL;

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

Рассмотрим приложение, которое хранит профили пользователей в таблице-справочнике.

-- Здесь используется столбец типа text, но в настоящем приложении
-- может использоваться citext, который доступен в
-- дополнительном модуле Postgres Pro (contrib)

CREATE TABLE users ( email text PRIMARY KEY );
SELECT create_reference_table('users');

Представим, что в таблицу попадает несколько некорректных адресов.

INSERT INTO users VALUES
   ('foo@example.com'), ('hacker12@aol.com'), ('lol');

Нужно проверить адреса, но Postgres Pro обычно не позволяет добавлять ограничение CHECK, которому не удовлетворяют существующие строки. Однако допускается ограничение NOT VALID:

ALTER TABLE users
ADD CONSTRAINT syntactic_email
CHECK (email ~
   '^[a-zA-Z0-9.!#$%&''*+/=?^_`{|}~-]+@[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$'
) NOT VALID;

Эти команды выполнятся успешно, и новые строки будут проверяться.

INSERT INTO users VALUES ('fake');

/*
ERROR:  new row for relation "users_102010" violates
        check constraint "syntactic_email_102010"
DETAIL:  Failing row contains (fake).
*/

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

-- Попытка проверить все строки позднее
ALTER TABLE users
VALIDATE CONSTRAINT syntactic_email;

В документации Postgres Pro подробно описаны ограничения NOT VALID и VALIDATE CONSTRAINT в ALTER TABLE.

В citus поддерживается добавление и удаление индексов:

-- Добавление индекса

CREATE INDEX clicked_at_idx ON clicks USING BRIN (clicked_at);

-- Удаление индекса

DROP INDEX clicked_at_idx;

Для добавления индекса требуется блокировка операций записи, что может быть нежелательно в многоарендной «системе записи». Чтобы свести к минимуму время простоя приложения, создавайте индекс параллельно. Этот метод более трудоёмок, чем стандартное построение индекса, и его выполнение занимает значительно больше времени. Однако, поскольку такой метод позволяет продолжать выполнение обычных операций во время построения индекса, он более предпочтителен для добавления новых индексов в производственной среде.

-- Добавление индекса без блокировки операций записи в таблицу

CREATE INDEX CONCURRENTLY clicked_at_idx ON clicks USING BRIN (clicked_at);

В citus также поддерживаются операторы INSERT… SELECT, вставляющие строки на основе результатов запроса SELECT. Это удобный способ заполнения таблиц, который также позволяет выполнять команду UPSERT с помощью предложения ON CONFLICT, что является самым простым примером выполнения распределённой свёртки.

В citus существует три способа вставки с помощью оператора SELECT:

Чтобы узнать, какой способ использует citus в данный момент, используйте команду EXPLAIN. Если в целевой таблице очень большое количество сегментов, возможно, стоит отключить пересекционирование, см. citus.enable_repartitioned_insert_select.

Для массовой загрузки данных из файла можно напрямую использовать команду \copy.

Сначала загрузите пример с набором данных github_events:

wget http://examples.citusdata.com/github_archive/github_events-2015-01-01-{0..5}.csv.gz
gzip -d github_events-2015-01-01-*.gz

Затем скопируйте данные с помощью psql. Обратите внимание, что для этих данных требуется база данных с кодировкой UTF-8:

\COPY github_events FROM 'github_events-2015-01-01-0.csv' WITH (format CSV)

В citus агрегатные функции count(distinct) поддерживаются несколькими способами. Если агрегатная функция count(distinct) выполняет агрегирование по столбцу распределения, citus может напрямую передавать запрос рабочим узлам. В противном случае citus запускает отдельные операторы SELECT на каждом рабочем узле и список возвращается координатору, где производится окончательный подсчёт.

Обратите внимание, что передача этих данных замедляется с увеличением количества отдельных элементов на рабочих узлах. Это особенно актуально для запросов, содержащих несколько агрегатных функций count(distinct), например:

-- Несколько различных функций count в одном запросе обычно выполняются медленно
SELECT count(distinct a), count(distinct b), count(distinct c)
FROM table_abc;

Для запросов такого типа отдельные операторы SELECT на рабочих узлах в результате по сути создают перекрёстное произведение строк, которые должны быть переданы координатору.

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

Столбец HyperLogLog. Некоторые пользователи уже хранят данные в виде столбцов hll. В таких случаях они могут динамически группировать эти данные с помощью функции hll_union_agg(hll_column).

Вычислить первые n элементов множества можно с помощью функций count, sort и limit. Однако по мере увеличения объёма данных этот метод становится медленным и ресурсоёмким. В таком случае эффективнее использовать приближение.

Расширение с открытым исходным кодом topn для Postgres Pro позволяет быстро получать приблизительные результаты для запросов типа «top-n». Расширение материализует первые значения в тип данных json. Расширение topn может постепенно обновлять эти значения или объединять их по мере необходимости по различным временным интервалам.

Прежде чем перейти к реалистичному примеру использования topn, рассмотрим, как работают некоторые его примитивные операции. Сначала topn_add изменяет объект JSON, подсчитывая количество просмотров ключа:

-- Сначала ничего нет, фиксируем, что увидели букву «а»
SELECT topn_add('{}', 'a');
-- => {"a": 1}

-- Фиксируем появление ещё одной буквы «а»
SELECT topn_add(topn_add('{}', 'a'), 'a');
-- => {"a": 2}

Расширение также предоставляет агрегатные функции для сканирования нескольких значений:

-- Для normal_rand
CREATE EXTENSION tablefunc;

-- Подсчёт значений из нормального распределения
SELECT topn_add_agg(floor(abs(i))::text)
  FROM normal_rand(1000, 5, 0.7) i;
-- => {"2": 1, "3": 74, "4": 420, "5": 425, "6": 77, "7": 3}

Если количество уникальных значений превышает предельную величину, агрегирование удаляет информацию о тех значениях, которые встречаются реже всего. Это позволяет контролировать использование пространства. Предельную величину можно установить с помощью параметра конфигурации topn.number_of_counters, значение по умолчанию — 1000.

Теперь рассмотрим более реалистичный пример работы topn. Для этого возьмём обзоры продуктов Amazon за 2000 год и воспользуемся topn для быстрой генерации запросов. Сначала загрузите набор данных:

curl -L https://examples.citusdata.com/customer_reviews_2000.csv.gz | \
  gunzip > reviews.csv

Затем заполните распределённую таблицу этими данными:

CREATE TABLE customer_reviews
(
    customer_id TEXT,
    review_date DATE,
    review_rating INTEGER,
    review_votes INTEGER,
    review_helpful_votes INTEGER,
    product_id CHAR(10),
    product_title TEXT,
    product_sales_rank BIGINT,
    product_group TEXT,
    product_category TEXT,
    product_subcategory TEXT,
    similar_product_ids CHAR(10)[]
);

SELECT create_distributed_table('customer_reviews', 'product_id');

\COPY customer_reviews FROM 'reviews.csv' WITH CSV

После этого добавьте расширение, создайте целевую таблицу для хранения данных JSON, сгенерированных topn, и вызовите ранее упоминавшуюся функцию topn_add_agg.

-- Выполните указанную команду на узле-кординаторе, она будет транслирована и на рабочие узлы
CREATE EXTENSION topn;

-- Таблица для материализации ежедневно агрегируемых данных
CREATE TABLE reviews_by_day
(
  review_date date unique,
  agg_data jsonb
);

SELECT create_reference_table('reviews_by_day');

-- Материализовать количество отзывов по каждому продукту за день для каждого покупателя
INSERT INTO reviews_by_day
  SELECT review_date, topn_add_agg(product_id)
  FROM customer_reviews
  GROUP BY review_date;

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

SELECT review_date, (topn(agg_data, 1)).*
FROM reviews_by_day
ORDER BY review_date
LIMIT 5;

┌─────────────┬────────────┬───────────┐
│ review_date │    item    │ frequency │
├─────────────┼────────────┼───────────┤
│ 2000-01-01  │ 0939173344 │        12 │
│ 2000-01-02  │ B000050XY8 │        11 │
│ 2000-01-03  │ 0375404368 │        12 │
│ 2000-01-04  │ 0375408738 │        14 │
│ 2000-01-05  │ B00000J7J4 │        17 │
└─────────────┴────────────┴───────────┘

Поля JSON, созданные topn, можно объединить с помощью функций topn_union и topn_union_agg. Последняя может использоваться, чтобы объединить данные за весь первый месяц и составить список из пяти продуктов, получивших наибольшее количество отзывов за этот период.

SELECT (topn(topn_union_agg(agg_data), 5)).*
FROM reviews_by_day
WHERE review_date >= '2000-01-01' AND review_date < '2000-02-01'
ORDER BY 2 DESC;

┌────────────┬───────────┐
│    item    │ frequency │
├────────────┼───────────┤
│ 0375404368 │       217 │
│ 0345417623 │       217 │
│ 0375404376 │       217 │
│ 0375408738 │       217 │
│ 043936213X │       204 │
└────────────┴───────────┘

За подробным описанием и примерами обратитесь к файлу readme topn.

Поиск точного процентиля по большому количеству строк может быть чрезвычайно затратным, поскольку все строки необходимо передать узлу-координатору для окончательной сортировки и обработки. С другой стороны, поиск приближённых значений можно выполнять параллельно на рабочих узлах, используя так называемый потоковый алгоритм приближения. Затем узел-координатор объединяет сжатые сводки в окончательный результат, а не читает полные строки.

Популярный алгоритм для процентилей использует сжатую структуру данных под названием t-digest, и доступен для Postgres Pro в расширении tdigest. В citus есть встроенная поддержка этого расширения.

Пример использования tdigest в citus:

Когда в запросах используются какие-либо определённые в расширении агрегатные функции, в citus переписываются запросы, чтобы по возможности передавать частичные вычисления tdigest рабочим узлам.

Точностью tdigest можно управлять с помощью аргумента compression, передаваемого в агрегатных функциях. Компромисс — это точность и объём данных, которыми обмениваются рабочие узлы и координатор. Подробное описание использования агрегатных функций в tdigest можно найти в документации расширения.

Когда две таблицы совмещены, их можно эффективно объединить в общих столбцах распределения. Совмещённое соединение — наиболее эффективный способ соединения двух больших распределённых таблиц.

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

Таблицы-справочники можно использовать как таблицы «размерностей» для эффективного объединения с большими таблицами «фактов». Поскольку таблицы-справочники полностью реплицируются на все рабочие узлы, справочное соединение можно разложить на локальные соединения на каждом рабочем узле и выполнять параллельно. Справочное соединение похоже на более гибкую версию совмещённого соединения, поскольку таблицы-справочники не распределяются по какому-либо конкретному столбцу и могут свободно соединяться по любому из своих столбцов.

Таблицы-справочники также можно соединять с локальными таблицами на узле-координаторе.

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

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

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

При необходимости citus может собирать результаты подзапросов и CTE на узле-координаторе, а затем передавать их обратно через рабочие узлы для использования во внешнем запросе. Это позволяет citus поддерживать большее разнообразие SQL-конструкций.

Например, подзапросы в предложении WHERE иногда не могут выполняться одновременно с основным запросом, и должны выполняться отдельно. Допустим, что в приложении для веб-аналитики поддерживается таблица page_views, секционированная по page_id. Чтобы запросить количество посетителей на двадцати самых посещаемых страницах, можно использовать подзапрос для формирования списка страниц, а затем внешний запрос для подсчета посетителей.

SELECT page_id, count(distinct host_ip)
FROM page_views
WHERE page_id IN (
  SELECT page_id
  FROM page_views
  GROUP BY page_id
  ORDER BY count(*) DESC
  LIMIT 20
)
GROUP BY page_id;

Исполнитель выполнил бы фрагмент этого запроса для каждого сегмента по page_id, подсчитав отдельные host_ips и объединив результаты на узле-координаторе. Однако LIMIT в подзапросе означает, что он не может быть выполнен как часть фрагмента. При рекурсивном планировании запроса citus может запускать подзапрос отдельно, передавать результаты всем рабочим узлам, выполнять основной фрагмент запроса и возвращать результаты узлу-координатору. Механизм «push-pull» поддерживает подзапросы, подобные описанному выше.

Рассмотрим на рабочем примере выходные данные EXPLAIN для этого запроса:

GroupAggregate  (cost=0.00..0.00 rows=0 width=0)
  Group Key: remote_scan.page_id
  ->  Sort  (cost=0.00..0.00 rows=0 width=0)
    Sort Key: remote_scan.page_id
    ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
      ->  Distributed Subplan 6_1
        ->  Limit  (cost=0.00..0.00 rows=0 width=0)
          ->  Sort  (cost=0.00..0.00 rows=0 width=0)
            Sort Key: COALESCE((pg_catalog.sum((COALESCE((pg_catalog.sum(remote_scan.worker_column_2))::bigint, '0'::bigint))))::bigint, '0'::bigint) DESC
            ->  HashAggregate  (cost=0.00..0.00 rows=0 width=0)
              Group Key: remote_scan.page_id
              ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
                Task Count: 32
                Tasks Shown: One of 32
                ->  Task
                  Node: host=localhost port=9701 dbname=postgres
                  ->  HashAggregate  (cost=54.70..56.70 rows=200 width=12)
                    Group Key: page_id
                    ->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=4)
      Task Count: 32
      Tasks Shown: One of 32
      ->  Task
        Node: host=localhost port=9701 dbname=postgres
        ->  HashAggregate  (cost=84.50..86.75 rows=225 width=36)
          Group Key: page_views.page_id, page_views.host_ip
          ->  Hash Join  (cost=17.00..78.88 rows=1124 width=36)
            Hash Cond: (page_views.page_id = intermediate_result.page_id)
            ->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=36)
            ->  Hash  (cost=14.50..14.50 rows=200 width=4)
              ->  HashAggregate  (cost=12.50..14.50 rows=200 width=4)
                Group Key: intermediate_result.page_id
                ->  Function Scan on read_intermediate_result intermediate_result  (cost=0.00..10.00 rows=1000 width=4)

Разобьём план на части и рассмотрим каждую отдельно.

GroupAggregate  (cost=0.00..0.00 rows=0 width=0)
  Group Key: remote_scan.page_id
  ->  Sort  (cost=0.00..0.00 rows=0 width=0)
    Sort Key: remote_scan.page_id

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

->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
  ->  Distributed Subplan 6_1
.

У выборочное сканирования есть два больших поддерева, начинающихся с «распределённого подплана».

->  Limit  (cost=0.00..0.00 rows=0 width=0)
  ->  Sort  (cost=0.00..0.00 rows=0 width=0)
    Sort Key: COALESCE((pg_catalog.sum((COALESCE((pg_catalog.sum(remote_scan.worker_column_2))::bigint, '0'::bigint))))::bigint, '0'::bigint) DESC
    ->  HashAggregate  (cost=0.00..0.00 rows=0 width=0)
      Group Key: remote_scan.page_id
      ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
        Task Count: 32
        Tasks Shown: One of 32
        ->  Task
          Node: host=localhost port=9701 dbname=postgres
          ->  HashAggregate  (cost=54.70..56.70 rows=200 width=12)
            Group Key: page_id
            ->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=4)
.

Рабочие узлы выполняют вышеуказанные операции для каждого из тридцати двух сегментов (citus выбирает для отображения один из них). Все части подзапроса IN (…) легко узнаваемы: сортировка, группирование и ограничение. Когда этот запрос завершается на всех рабочих узлах, они отправляют результаты обратно узлу-координатору, который объединяет их как «промежуточные результаты».

Task Count: 32
Tasks Shown: One of 32
->  Task
  Node: host=localhost port=9701 dbname=postgres
  ->  HashAggregate  (cost=84.50..86.75 rows=225 width=36)
    Group Key: page_views.page_id, page_views.host_ip
    ->  Hash Join  (cost=17.00..78.88 rows=1124 width=36)
      Hash Cond: (page_views.page_id = intermediate_result.page_id)
.

Расширение citus запускает другое задание для исполнителя во втором поддереве. Он считает отдельных посетителей в page_views и использует JOIN для соединения с промежуточными результатами. Промежуточные результаты помогут ограничиться двадцатью страницами с наибольшими показателями.

->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=36)
->  Hash  (cost=14.50..14.50 rows=200 width=4)
  ->  HashAggregate  (cost=12.50..14.50 rows=200 width=4)
    Group Key: intermediate_result.page_id
    ->  Function Scan on read_intermediate_result intermediate_result  (cost=0.00..10.00 rows=1000 width=4)
.

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

В примере выше показано, как в citus запрос выполняется в несколько этапов с распределённым подпланом и как можно использовать EXPLAIN, чтобы изучить выполнение распределённых запросов.

Указанные ограничения применяются ко всем моделям операций:

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

Подробную информацию о диалекте SQL-команд Postgres Pro (который может применяться пользователями citus в исходном виде) можно найти в Справке: «Команды SQL».

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

Если SQL-запрос не поддерживается, один из способов обойти его — использовать CTE вместе с так называемым двухэтапным выполнением.

SELECT * FROM dist WHERE EXISTS (SELECT 1 FROM local WHERE local.a = dist.a);
/*
ERROR:  direct joins between distributed and local tables are not supported
HINT:  Use CTEs or subqueries to select from local tables and use them in joins
*/

Чтобы обойти это ограничение, можно превратить запрос в запрос маршрутизатора, обернув распределённую часть в CTE.

WITH cte AS (SELECT * FROM dist)
SELECT * FROM cte WHERE EXISTS (SELECT 1 FROM local WHERE local.a = cte.a);

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

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

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

-- Такой запрос не работает

  SELECT repo_id, event_type, event_public,
         grouping(event_type, event_public),
         min(created_at)
    FROM github_events
   WHERE repo_id IN (8514, 15435, 19438, 21692)
GROUP BY repo_id, ROLLUP(event_type, event_public);

ERROR:  could not run distributed query with GROUPING
HINT:  Consider using an equality filter on the distributed table's partition column.

Но есть одна хитрость. Соответствующую информацию можно передать координатору в виде временной таблицы:

-- Размещение данных без агрегирования в локальную таблицу

CREATE TEMP TABLE results AS (
  SELECT repo_id, event_type, event_public, created_at
    FROM github_events
       WHERE repo_id IN (8514, 15435, 19438, 21692)
    );

-- Запуск агрегирования локально

  SELECT repo_id, event_type, event_public,
         grouping(event_type, event_public),
         min(created_at)
    FROM results
GROUP BY repo_id, ROLLUP(event_type, event_public);

 repo_id |    event_type     | event_public | grouping |         min
---------+-------------------+--------------+----------+---------------------
    8514 | PullRequestEvent  | t            |        0 | 2016-12-01 05:32:54
    8514 | IssueCommentEvent | t            |        0 | 2016-12-01 05:32:57
   19438 | IssueCommentEvent | t            |        0 | 2016-12-01 05:48:56
   21692 | WatchEvent        | t            |        0 | 2016-12-01 06:01:23
   15435 | WatchEvent        | t            |        0 | 2016-12-01 05:40:24
   21692 | WatchEvent        |              |        1 | 2016-12-01 06:01:23
   15435 | WatchEvent        |              |        1 | 2016-12-01 05:40:24
    8514 | PullRequestEvent  |              |        1 | 2016-12-01 05:32:54
    8514 | IssueCommentEvent |              |        1 | 2016-12-01 05:32:57
   19438 | IssueCommentEvent |              |        1 | 2016-12-01 05:48:56
   15435 |                   |              |        3 | 2016-12-01 05:40:24
   21692 |                   |              |        3 | 2016-12-01 06:01:23
   19438 |                   |              |        3 | 2016-12-01 05:48:56
    8514 |                   |              |        3 | 2016-12-01 05:32:54

Создание временной таблицы на координаторе — это крайняя мера. Она ограничена размером диска и мощностью ЦП узла.

Попробуйте переписать свои запросы с использованием синтаксиса INSERT INTO ... SELECT.

Следующий SQL-код:

INSERT INTO a.widgets (map_id, widget_name)
VALUES (
    (SELECT mt.map_id FROM a.map_tags mt WHERE mt.map_license = '12345'),
    'Test'
);

Станет таким:

INSERT INTO a.widgets (map_id, widget_name)
SELECT mt.map_id, 'Test'
  FROM a.map_tags mt
 WHERE mt.map_license = '12345';

В таблице pg_dist_partition хранятся метаданные о распределённых таблицах. Для каждой распределённой таблицы также хранится информация о методе распределения и подробная информация о столбце распределения.

SELECT * FROM pg_dist_partition;
 logicalrelid  | partmethod |                                                        partkey                                                         | colocationid | repmodel 
---------------+------------+------------------------------------------------------------------------------------------------------------------------+--------------+----------
 github_events | h          | {VAR :varno 1 :varattno 4 :vartype 20 :vartypmod -1 :varcollid 0 :varlevelsup 0 :varnoold 1 :varoattno 4 :location -1} |            2 | s
 (1 row)

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

SELECT * FROM pg_dist_shard;
 logicalrelid  | shardid | shardstorage | shardminvalue | shardmaxvalue
---------------+---------+--------------+---------------+---------------
 github_events |  102026 | t            | 268435456     | 402653183
 github_events |  102027 | t            | 402653184     | 536870911
 github_events |  102028 | t            | 536870912     | 671088639
 github_events |  102029 | t            | 671088640     | 805306367
 (4 rows)

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

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

Это представление помогает исследовать сегменты, например, чтобы найти неравенство размеров узлов.

SELECT * FROM citus_shards;

.
 table_name | shardid | shard_name   | citus_table_type | colocation_id | nodename  | nodeport | shard_size
------------+---------+--------------+------------------+---------------+-----------+----------+------------
 dist       |  102170 | dist_102170  | distributed      |            34 | localhost |     9701 |   90677248
 dist       |  102171 | dist_102171  | distributed      |            34 | localhost |     9702 |   90619904
 dist       |  102172 | dist_102172  | distributed      |            34 | localhost |     9701 |   90701824
 dist       |  102173 | dist_102173  | distributed      |            34 | localhost |     9702 |   90693632
 ref        |  102174 | ref_102174   | reference        |             2 | localhost |     9701 |       8192
 ref        |  102174 | ref_102174   | reference        |             2 | localhost |     9702 |       8192
 dist2      |  102175 | dist2_102175 | distributed      |            34 | localhost |     9701 |     933888
 dist2      |  102176 | dist2_102176 | distributed      |            34 | localhost |     9702 |     950272
 dist2      |  102177 | dist2_102177 | distributed      |            34 | localhost |     9701 |     942080
 dist2      |  102178 | dist2_102178 | distributed      |            34 | localhost |     9702 |     933888

Параметр colocation_id относится к группе совмещения. За дополнительной информацией о citus_table_type обратитесь к разделу Типы таблиц.

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

SELECT * FROM pg_dist_placement;
  placementid | shardid | shardstate | shardlength | groupid
 -------------+---------+------------+-------------+---------
            1 |  102008 |          1 |           0 |       1
            2 |  102008 |          1 |           0 |       2
            3 |  102009 |          1 |           0 |       2
            4 |  102009 |          1 |           0 |       3
            5 |  102010 |          1 |           0 |       3
            6 |  102010 |          1 |           0 |       4
            7 |  102011 |          1 |           0 |       4

В таблице pg_dist_node содержится информация о рабочих узлах кластера.

SELECT * FROM pg_dist_node;
 nodeid | groupid | nodename  | nodeport | noderack | hasmetadata | isactive | noderole | nodecluster | metadatasynced | shouldhaveshards
--------+---------+-----------+----------+----------+-------------+----------+----------+-------------+----------------+------------------
      1 |       1 | localhost |    12345 | default  | f           | t        | primary  | default     | f              | t
      2 |       2 | localhost |    12346 | default  | f           | t        | primary  | default     | f              | t
      3 |       3 | localhost |    12347 | default  | f           | t        | primary  | default     | f              | t
(3 rows)

В таблице citus.pg_dist_object содержится список объектов, например типов и функций, которые были созданы на узле-координаторе и распространены на рабочие узлы. При добавлении новых рабочих узлов в кластер citus автоматически создаёт копии распределённых объектов на новых узлах (в правильном порядке, чтобы соответствовать зависимостям объектов).

«Стабильные адреса» однозначно идентифицируют объекты независимо от конкретного сервера. Во время обновления Postgres Pro в citus объекты отслеживаются с помощью стабильных адресов, созданных функцией pg_identify_object_as_address.

Ниже представлен пример добавления записи в таблицу citus.pg_dist_object функцией create_distributed_function:

CREATE TYPE stoplight AS enum ('green', 'yellow', 'red');

CREATE OR REPLACE FUNCTION intersection()
RETURNS stoplight AS $$
DECLARE
        color stoplight;
BEGIN
        SELECT *
          FROM unnest(enum_range(NULL::stoplight)) INTO color
         ORDER BY random() LIMIT 1;
        RETURN color;
END;
$$ LANGUAGE plpgsql VOLATILE;

SELECT create_distributed_function('intersection()');

-- will have two rows, one for the TYPE and one for the FUNCTION
TABLE citus.pg_dist_object;

-[ RECORD 1 ]---------------+------
classid                     | 1247
objid                       | 16780
objsubid                    | 0
type                        |
object_names                |
object_args                 |
distribution_argument_index |
colocationid                |
-[ RECORD 2 ]---------------+------
classid                     | 1255
objid                       | 16788
objsubid                    | 0
type                        |
object_names                |
object_args                 |
distribution_argument_index |
colocationid                |

В citus поддерживается сегментирование на основе схем и есть представление citus_schemas, содержащее информацию о том, какие схемы в системе были распределены. В представлении показываются только распределённые схемы, но не локальные.

Например:

schema_name  | colocation_id | schema_size | schema_owner
--------------+---------------+-------------+--------------
user_service |             1 | 0 bytes     | user_service
time_service |             2 | 0 bytes     | time_service
ping_service |             3 | 632 kB      | ping_service

В представлении citus_tables содержится сводная информация обо всех таблицах, управляемых citus (распределённые и таблицы-справочники). В представлении объединена информация из таблиц метаданных citus, чтобы можно было проще исследовать параметры этих таблиц:

Например:

SELECT * FROM citus_tables;

┌────────────┬──────────────────┬─────────────────────┬───────────────┬────────────┬─────────────┬─────────────┬───────────────┐
│ table_name │ citus_table_type │ distribution_column │ colocation_id │ table_size │ shard_count │ table_owner │ access_method │
├────────────┼──────────────────┼─────────────────────┼───────────────┼────────────┼─────────────┼─────────────┼───────────────┤
│ foo.test   │ distributed      │ test_column         │             1 │ 0 bytes    │          32 │ citus       │ heap          │
│ ref        │ reference        │ <none>              │             2 │ 24 GB      │           1 │ citus       │ heap          │
│ test       │ distributed      │ id                  │             1 │ 248 TB     │          32 │ citus       │ heap          │
└────────────┴──────────────────┴─────────────────────┴───────────────┴────────────┴─────────────┴─────────────┴───────────────┘

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

В представлении есть следующие столбцы:

SELECT * FROM time_partitions;

┌────────────────────────┬──────────────────┬─────────────────────────────────────────┬─────────────────────┬─────────────────────┬───────────────┐
│      parent_table      │ partition_column │                partition                │     from_value      │      to_value       │ access_method │
├────────────────────────┼──────────────────┼─────────────────────────────────────────┼─────────────────────┼─────────────────────┼───────────────┤
│ github_columnar_events │ created_at       │ github_columnar_events_p2015_01_01_0000 │ 2015-01-01 00:00:00 │ 2015-01-01 02:00:00 │ columnar      │
│ github_columnar_events │ created_at       │ github_columnar_events_p2015_01_01_0200 │ 2015-01-01 02:00:00 │ 2015-01-01 04:00:00 │ columnar      │
│ github_columnar_events │ created_at       │ github_columnar_events_p2015_01_01_0400 │ 2015-01-01 04:00:00 │ 2015-01-01 06:00:00 │ columnar      │
│ github_columnar_events │ created_at       │ github_columnar_events_p2015_01_01_0600 │ 2015-01-01 06:00:00 │ 2015-01-01 08:00:00 │ heap          │
└────────────────────────┴──────────────────┴─────────────────────────────────────────┴─────────────────────┴─────────────────────┴───────────────┘

В таблице pg_dist_colocation содержится информация о том, какие сегменты таблиц должны быть размещены вместе, или совмещены. Если две таблицы находятся в одной группе совмещения, в citus гарантируется, что сегменты с одинаковыми значениями секций будут размещены на одних и тех же рабочих узлах. Таким образом можно оптимизировать соединения, выполнять некоторые распределённые свёртки и поддерживать внешние ключи. Совмещение сегментов предполагается, когда количество сегментов и типы столбцов секционирования совпадают в двух таблицах; однако при создании распределённой таблицы можно указать пользовательскую группу совмещения, если необходимо.

SELECT * FROM pg_dist_colocation;
  colocationid | shardcount | replicationfactor | distributioncolumntype | distributioncolumncollation
 --------------+------------+-------------------+------------------------+-----------------------------
             2 |         32 |                 1 |                     20 |                           0
  (1 row)

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

Расширение citus поставляется со следующими стратегиями, указанными в таблице:

SELECT * FROM pg_dist_rebalance_strategy;

-[ RECORD 1 ]------------------+---------------------------------
name                           | by_shard_count
default_strategy               | f
shard_cost_function            | citus_shard_cost_1
node_capacity_function         | citus_node_capacity_1
shard_allowed_on_node_function | citus_shard_allowed_on_node_true
default_threshold              | 0
minimum_threshold              | 0
improvement_threshold          | 0
-[ RECORD 2 ]------------------+---------------------------------
name                           | by_disk_size
default_strategy               | t
shard_cost_function            | citus_shard_cost_by_disk_size
node_capacity_function         | citus_node_capacity_1
shard_allowed_on_node_function | citus_shard_allowed_on_node_true
default_threshold              | 0.1
minimum_threshold              | 0.01
improvement_threshold          | 0.5

При использовании стратегии by_shard_count каждый сегмент имеет одинаковую стоимость. Она применяется для выравнивании количества сегментов на всех узлах. Если используется стратегия по умолчанию, by_disk_size, стоимость каждого сегмента соответствует его размеру на диска в байтах с прибавлением стоимости совмещённых с ним сегментов. Размер диска рассчитывается с помощью функции pg_total_relation_size, поэтому учитывается размер индексов. Цель этой стратегии — равномерное использование дискового пространства на всех узлах. Обратите внимание, что установка порогового значения 0.1 предотвращает ненужное перемещение сегментов, вызванное незначительными отличиями используемого дискового пространства.

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

В citus реализована таблица citus_stat_statements для сбора статистики выполнения запросов. Она аналогична представлению pg_stat_statements в Postgres Pro, которое отслеживает статистику скорости выполнения запросов, и может быть соединена с ним.

-- Создание и заполнение распределённой таблицы
create table foo ( id int );
select create_distributed_table('foo', 'id');
insert into foo select generate_series(1,100);

-- Включение сбора статистики
-- Представление pg_stat_statements должно быть указано в shared_preload_libraries
create extension pg_stat_statements;

SELECT count(*) from foo;
SELECT * FROM foo where id = 42;

SELECT * FROM citus_stat_statements;

Результат:

-[ RECORD 1 ]-+----------------------------------------------
queryid       | -909556869173432820
userid        | 10
dbid          | 13340
query         | insert into foo select generate_series($1,$2)
executor      | insert-select
partition_key |
calls         | 1
-[ RECORD 2 ]-+----------------------------------------------
queryid       | 3919808845681956665
userid        | 10
dbid          | 13340
query         | select count(*) from foo;
executor      | adaptive
partition_key |
calls         | 1
-[ RECORD 3 ]-+----------------------------------------------
queryid       | 5351346905785208738
userid        | 10
dbid          | 13340
query         | select * from foo where id = $1
executor      | adaptive
partition_key | 42
calls         | 1

Ограничения:

Представление citus_stat_tenants дополняет таблицу citus_stat_statements, предоставляя информацию о том, сколько запросов выполняется каждым арендатором. Отслеживание запросов по арендаторам может использоваться для принятия решения о выполнении изоляции арендаторов.

В этом представлении подсчитываются последние одноарендные запросы, выполненные в течение настраиваемого периода времени. Количество запросов только для чтения и общее количество запросов за период увеличивается до конца текущего периода. Затем это количество переносятся в статистику за последний период, которая хранится до истечения срока действия. Период подсчёта запросов можно задать в секундах с помощью citus.stats_tenants_ period, значение по умолчанию — 60 секунд.

В представлении отображается до citus.stat_tenants_limit строк (по умолчанию — 100). Учитываются только запросы к одному арендатору, многоарендные запросы игнорируются.

Отслеживание статистики на уровне арендатора увеличивает издержки и по умолчанию отключено. Чтобы включить его, установите для citus.stat_tenants_track значение 'all'.

В качестве примера предположим, что есть распределённая таблица с именем dist_table со столбцом распределения tenant_id. Затем выполним несколько запросов:

INSERT INTO dist_table(tenant_id) VALUES (1);
INSERT INTO dist_table(tenant_id) VALUES (1);
INSERT INTO dist_table(tenant_id) VALUES (2);

SELECT count(*) FROM dist_table WHERE tenant_id = 1;

В статистике на уровне арендатора будут отражены только что выполненные запросы:

SELECT tenant_attribute, read_count_in_this_period,
       query_count_in_this_period, cpu_usage_in_this_period
  FROM citus_stat_tenants;

tenant_attribute | read_count_in_this_period | query_count_in_this_period | cpu_usage_in_this_period
------------------+---------------------------+----------------------------+--------------------------
1                |                         1 |                          3 |                 0.000883
2                |                         0 |                          1 |                 0.000144

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

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

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

Рассмотрим в качестве примера подсчёт строк в распределённой таблице:

-- Выполнение за один сеанс
-- (с pg_sleep, чтобы всё было видно)

SELECT count(*), pg_sleep(3) FROM users_table;

Запрос появляется в citus_dist_stat_activity:

-- Запуск в другом сеансе

SELECT * FROM citus_dist_stat_activity;

-[ RECORD 1 ]----+-------------------------------------------
global_pid       | 10000012199
nodeid           | 1
is_worker_query  | f
datid            | 13724
datname          | postgres
pid              | 12199
leader_pid       |
usesysid         | 10
usename          | postgres
application_name | psql
client_addr      |
client_hostname  |
client_port      | -1
backend_start    | 2022-03-23 11:30:00.533991-05
xact_start       | 2022-03-23 19:35:28.095546-05
query_start      | 2022-03-23 19:35:28.095546-05
state_change     | 2022-03-23 19:35:28.09564-05
wait_event_type  | Timeout
wait_event       | PgSleep
state            | active
backend_xid      |
backend_xmin     | 777
query_id         |
query            | SELECT count(*), pg_sleep(3) FROM users_table;
backend_type     | client backend

В представлении citus_dist_stat_activity скрыты внутренние фрагменты запроса citus. Чтобы их увидеть, можно использовать более подробное представление citus_stat_activity. Например, предыдущий запрос count(*) обращается ко всем сегментам. Часть информации находится в сегменте users_table_102039, который виден в запросе ниже.

SELECT * FROM citus_stat_activity;

-[ RECORD 1 ]----+-----------------------------------------------------------------------
global_pid       | 10000012199
nodeid           | 1
is_worker_query  | f
datid            | 13724
datname          | postgres
pid              | 12199
leader_pid       |
usesysid         | 10
usename          | postgres
application_name | psql
client_addr      |
client_hostname  |
client_port      | -1
backend_start    | 2022-03-23 11:30:00.533991-05
xact_start       | 2022-03-23 19:32:18.260803-05
query_start      | 2022-03-23 19:32:18.260803-05
state_change     | 2022-03-23 19:32:18.260821-05
wait_event_type  | Timeout
wait_event       | PgSleep
state            | active
backend_xid      |
backend_xmin     | 777
query_id         |
query            | SELECT count(*), pg_sleep(3) FROM users_table;
backend_type     | client backend
-[ RECORD 2 ]----+-----------------------------------------------------------------------------------------
global_pid       | 10000012199
nodeid           | 1
is_worker_query  | t
datid            | 13724
datname          | postgres
pid              | 12725
leader_pid       |
usesysid         | 10
usename          | postgres
application_name | citus_internal gpid=10000012199
client_addr      | 127.0.0.1
client_hostname  |
client_port      | 44106
backend_start    | 2022-03-23 19:29:53.377573-05
xact_start       |
query_start      | 2022-03-23 19:32:18.278121-05
state_change     | 2022-03-23 19:32:18.278281-05
wait_event_type  | Client
wait_event       | ClientRead
state            | idle
backend_xid      |
backend_xmin     |
query_id         |
query            | SELECT count(*) AS count FROM public.users_table_102039 users WHERE true
backend_type     | client backend

Поле query показывает строки, подсчитываемые в сегменте 102039.

Ниже представлены примеры информативных запросов, которые можно сформулировать с помощью citus_stat_activity:

-- События ожидания активных запросов

SELECT query, wait_event_type, wait_event
  FROM citus_stat_activity
 WHERE state='active';

-- Первые события ожидания активных запросов

SELECT wait_event, wait_event_type, count(*)
  FROM citus_stat_activity
 WHERE state='active'
 GROUP BY wait_event, wait_event_type
 ORDER BY count(*) desc;

-- Общее количество внутренних подключений, созданных для каждого узла в citus

SELECT nodeid, count(*)
  FROM citus_stat_activity
 WHERE is_worker_query
 GROUP BY nodeid;

Следующее представление — citus_lock_waits. Чтобы увидеть его работу, можно вручную создать ситуацию с блокировкой. Сначала настройте тестовую таблицу узла-координатора:

CREATE TABLE numbers AS
  SELECT i, 0 AS j FROM generate_series(1,10) AS i;
SELECT create_distributed_table('numbers', 'i');

Затем с помощью двух сеансов на узле-координаторе запустите следующую последовательность операторов:

-- Сеанс 1                           -- Сеанс 2
-------------------------------------  -------------------------------------
BEGIN;
UPDATE numbers SET j = 2 WHERE i = 1;
                                       BEGIN;
                                       UPDATE numbers SET j = 3 WHERE i = 1;
                                       -- (это вызывает блокировку)

В представлении citus_lock_waits содержится информация о данной ситуации.

SELECT * FROM citus_lock_waits;

-[ RECORD 1 ]-------------------------+--------------------------------------
waiting_gpid                          | 10000011981
blocking_gpid                         | 10000011979
blocked_statement                     | UPDATE numbers SET j = 3 WHERE i = 1;
current_statement_in_blocking_process | UPDATE numbers SET j = 2 WHERE i = 1;
waiting_nodeid                        | 1
blocking_nodeid                       | 1

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

Таблица pg_dist_authinfo содержит параметры аутентификации, используемые узлами citus для подключения друг к другу.

При установке подключения узел проверяет, существует ли в таблице строка с nodeid и желаемым rolename. Если это так, узел включает соответствующую строку authinfo в строку подключения libpq. Типичным примером является сохранение пароля, например 'password=abc123', но можно ознакомиться с полным списком возможностей.

Параметры в authinfo разделяются пробелами и имеют форму key=val. Чтобы записать пустое значение или значение, содержащее пробелы, заключите его в одинарные кавычки, например, keyword='a value'. Одинарные кавычки и обратные косые черты внутри значения должны экранироваться обратной косой чертой, т. е. \' и \\.

Столбец nodeid также может принимать специальные значения 0 — все узлы и -1 — соединения локального замыкания. Если для данного узла существуют как специальные правила, так и правила на уровне узлов, специальные правила имеют приоритет.

SELECT * FROM pg_dist_authinfo;

 nodeid | rolename | authinfo
--------+----------+-----------------
    123 | jdoe     | password=abc123
(1 row)

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

Если есть информация о пуле, citus попытается использовать эти значения вместо установки прямого подключения. Информация pg_dist_poolinfo в этом случае заменяет собой pg_dist_node.

-- Подключение к узлу 1 (согласно идентификации в pg_dist_node)

INSERT INTO pg_dist_poolinfo (nodeid, poolinfo)
     VALUES (1, 'host=127.0.0.1 port=5433');