17.3. Миграция
Пред. НаверхГлава 17. Установка из двоичных пакетовНачало След.

17.3. Миграция #

17.3.1. Миграция на Postgres Pro Shardman
17.3.2. Миграция схемы БД Postgres Pro Shardman
17.3.3. Миграция данных
17.3.4. Проверка миграции Postgres Pro Shardman

17.3.1. Миграция на Postgres Pro Shardman #

Разные основные версии Postgres Pro Shardman, а также разные продукты на базе PostgreSQL, имеющие одинаковую основную версию, могут иметь несовместимые на двоичном уровне базы данных, так что просто заменить программу сервера и продолжить использовать его в общем случае нельзя.

Инструкции по обновлению для определённой версии вы найдёте в Замечаниях к выпуску для этой версии Postgres Pro Shardman.

17.3.2. Миграция схемы БД Postgres Pro Shardman #

В качестве примера для разработки воспользуемся демонстрационной базой данных «Авиакомпании». Подробное описание схемы базы данных доступно по ссылке https://postgrespro.ru/education/demodb. Эта демонстрационная схема используется в образовательных программах Postgres Professional, например в курсе «QPT. Оптимизация запросов».

Авторы схемы охарактеризовали её так: «Схема демонстрационной базы данных достаточно проста, чтобы быть понятной каждому, и в то же время достаточно сложна, чтобы строить нетривиальные запросы.»

Схема базы данных содержит несколько таблиц с осмысленным содержимым. Для примера возьмём версию демонстрационной БД от 13.10.2016. Ссылки для загрузки дампов базы данных и схемы (на русском языке) находятся на странице https://postgrespro.ru/education/courses/QPT. Примеры запросов, помимо приведённых ниже, можно взять из вышеупомянутого курса и книги «Postgres. Первое знакомство».

В этом разделе показаны два примера модификации схемы и адаптации запроса:

  • Наивный подход прост, требует минимальных преобразований схемы и призван внести ясность в работу запросов в распределённой схеме.

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

17.3.2.1. Исходная схема базы данных #

Рисунок 17.2. Исходная схема базы данных

Исходная схема базы данных


Авторы описывают базу данных «Airlines» следующим образом:

Основной сущностью является бронирование (bookings).

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

Билет включает один или несколько перелетов (ticket_flights). Несколько перелетов могут включаться в билет в случаях, когда нет прямого рейса, соединяющего пункты отправления и назначения (полет с пересадками), либо когда билет взят «туда и обратно». В схеме данных нет жёсткого ограничения, но предполагается, что все билеты в одном бронировании имеют одинаковый набор перелетов.

Каждый рейс (flights) следует из одного аэропорта (airports) в другой. Рейсы с одним номером имеют одинаковые пункты вылета и назначения, но будут отличаться датой отправления.

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

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

Давайте посмотрим на общие сущности и размеры таблиц в приведённой выше схеме. Можно заметить, что таблицы ticket_flights, boarding_passes и tickets связаны полем ticket_no. Кроме того, объём данных в этих таблицах составляет 95% от общего объёма БД.

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

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

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

17.3.2.2. Конфигурация кластера Postgres Pro Shardman #

Кластер Postgres Pro Shardman состоит из четырёх узлов — node1, node2, node3 и node4. Каждый узел кластера представляет собой отдельный сегмент.

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

  • shard-1 находится на узле кластера node1

  • shard-2 находится на узле кластера node2

  • shard-3 находится на узле кластера node3

  • shard-4 находится на узле кластера node4

Кластер намеренно представлен в упрощённой конфигурации. У узлов кластера нет реплик, а конфигурация не является отказоустойчивой.

17.3.2.3. Выбор ключа сегментирования #

17.3.2.3.1. Наивный [15] подход — ключ сегментирования ticket_no #

При таком подходе выбор ключа сегментирования довольно очевиден. Это номер билета ticket_no. Номер билета является первичным ключом таблицы tickets и внешним ключом таблиц ticket_flights и boarding_passes.

Первичный ключ таблиц ticket_flights и boarding_passes является составным. Это уникальный индекс, состоящий из ticket_no и flight_id.

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

Остальные таблицы — airports, flights, aircrafts и seats достаточно малы и редко изменяются. Эти свойства позволяют сделать их глобальными таблицами, или таблицами-справочниками.

Рисунок 17.3. Схема наивного подхода

Схема наивного подхода


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

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

[15]

17.3.2.3.1.1. Создание схемы, распределённой по ticket_no #

Сначала включите трансляцию DDL-операторов во все сегменты кластера: 

    SET shardman.broadcast_ddl TO on;

Создайте схему bookings для всех сегментов: 

    CREATE SCHEMA bookings;

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

база данных содержит «снимки» данных — аналогично резервной копии реальной системы, сделанной в определённый момент времени. Например, статус рейса Departed означает, что воздушное судно уже вылетело и находилось в воздухе на момент создания резервной копии. Время «снимка» сохраняется в функции bookings.now(). Этой функцией можно пользоваться в демонстрационных запросах там, где в обычной жизни использовалась бы функция now(). Кроме того, значение, возвращаемое этой функцией, определяет версию демонстрационной базы данных. Последняя доступная версия — от 13.10.2016: 

    SELECT bookings.now() as now;
              now
    -----------------------
    2016-10-13 17:00:00+03

Относительно именно этого момента времени все рейсы делятся на прошедшие и будущие.

Давайте создадим вспомогательную функцию bookings.now(): 

        CREATE FUNCTION bookings.now() RETURNS timestamp with time zone
    LANGUAGE sql IMMUTABLE COST 0.00999999978
    AS
    $sql$
    SELECT $qq$2016-10-13 17:00:00$qq$::TIMESTAMP AT TIME ZONE
$zz$Europe/Moscow$zz$;
    $sql$;

Кроме самих таблиц нам потребуется глобальная последовательность для генерирования идентификаторов (ID) при вставке данных в таблицу flights. В данном примере последовательность создаётся явно и далее связывается со столбцом этой таблицы, задавая значения, генерируемые последовательностью по умолчанию.

Создайте последовательность, используя следующий DDL-оператор: 

    CREATE SEQUENCE bookings.flights_flight_id_seq
        INCREMENT BY 1
        NO MINVALUE
        NO MAXVALUE
        CACHE 1 with(global);

Выражение with(global) создаёт единую распределённую последовательность, доступную на всех узлах кластера, которая выдаёт значения из определённого диапазона в каждом сегменте кластера, при этом диапазоны разных сегментов не пересекаются. За более подробным описанием работы глобальных последовательностей обратитесь к CREATE SEQUENCE и Справке: «Команды SQL».

На практике глобальные последовательности в каждом сегменте являются регулярные последовательности, которым выделяются последовательные блоки (по умолчанию — 65536 номеров). Когда все номера в блоке будут выделены, следующий блок выделяется локальной последовательности в сегменте. То есть числа из глобальных последовательностей уникальны, но строгой монотонности нет, и в значениях, заданных генератором последовательностей, могут быть «дыры» [16].

Последовательности могут иметь тип bigserial, smallserial или serial и применяться как для сегментированных, так и для глобальных таблиц.

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

[16]

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

Создайте глобальные таблицы, используя следующие DDL-операторы: 

    CREATE TABLE bookings.aircrafts (
        aircraft_code character(3) NOT NULL primary key,
        model text NOT NULL,
        range integer NOT NULL,
        CONSTRAINT aircrafts_range_check CHECK ((range > 0))
    ) with (global);

    CREATE TABLE bookings.seats (
        aircraft_code character(3) references bookings.aircrafts(aircraft_code),
        seat_no character varying(4) NOT NULL,
        fare_conditions character varying(10) NOT NULL,
        CONSTRAINT seats_fare_conditions_check CHECK (((fare_conditions)::text = ANY (ARRAY[('Economy'::character varying)::text, ('Comfort'::character varying)::text, ('Business'::character varying)::text]))),
        PRIMARY KEY (aircraft_code, seat_no)
    ) with (global);

    CREATE TABLE bookings.airports (
        airport_code character(3) NOT NULL primary key,
        airport_name text NOT NULL,
        city text NOT NULL,
        longitude double precision NOT NULL,
        latitude double precision NOT NULL,
        timezone text NOT NULL
    )  with (global);

    CREATE TABLE bookings.bookings (
        book_ref character(6) NOT NULL,
        book_date timestamp with time zone NOT NULL,
        total_amount numeric(10,2) NOT NULL,
        PRIMARY KEY (book_ref)
    ) with (global);

    CREATE TABLE bookings.flights (
        flight_id bigint NOT NULL PRIMARY KEY,-- <= a sequence will be assigned
        flight_no character(6) NOT NULL,
        scheduled_departure timestamp with time zone NOT NULL,
        scheduled_arrival timestamp with time zone NOT NULL,
        departure_airport character(3) REFERENCES bookings.airports(airport_code),
        arrival_airport character(3) REFERENCES bookings.airports(airport_code),
        status character varying(20) NOT NULL,
        aircraft_code character(3) references bookings.aircrafts(aircraft_code),
        actual_departure timestamp with time zone,
        actual_arrival timestamp with time zone,
        CONSTRAINT flights_check CHECK ((scheduled_arrival > scheduled_departure)),
        CONSTRAINT flights_check1 CHECK (((actual_arrival IS NULL) OR ((actual_departure IS NOT NULL) AND (actual_arrival IS NOT NULL) AND (actual_arrival > actual_departure)))),
        CONSTRAINT flights_status_check CHECK (((status)::text = ANY (ARRAY[('On Time'::character varying)::text, ('Delayed'::character varying)::text, ('Departed'::character varying)::text, ('Arrived'::character varying)::text, ('Scheduled'::character varying)::text, ('Cancelled'::character varying)::text])))
    ) with (global);

    -- associate the sequence with table column
    ALTER SEQUENCE bookings.flights_flight_id_seq OWNED BY bookings.flights.flight_id;


    -- assign the default value to the column
    ALTER TABLE bookings.flights ALTER COLUMN flight_id SET DEFAULT nextval('bookings.flights_flight_id_seq');

    ALTER TABLE bookings.flights ADD CONSTRAINT flights_flight_no_scheduled_departure_key UNIQUE (flight_no, scheduled_departure);

Далее создайте сегментированные таблицы tickets, ticket_flights и boarding_passes в схеме bookings: 

    CREATE TABLE bookings.tickets (
    ticket_no character(13) PRIMARY KEY,
    book_ref character(6) REFERENCES bookings.bookings(book_ref),
    passenger_id character varying(20) NOT NULL,
    passenger_name text NOT NULL,
    contact_data jsonb
) with (distributed_by='ticket_no', num_parts=4);

CREATE TABLE bookings.ticket_flights (
    ticket_no character(13) NOT NULL,
    flight_id bigint references bookings.flights(flight_id),
    fare_conditions character varying(10) NOT NULL,
    amount numeric(10,2) NOT NULL,
    CONSTRAINT ticket_flights_amount_check CHECK ((amount >= (0)::numeric)),
    CONSTRAINT ticket_flights_fare_conditions_check CHECK (((fare_conditions)::text = ANY (ARRAY[('Economy'::character varying)::text, ('Comfort'::character varying)::text, ('Business'::character varying)::text]))),
    PRIMARY KEY (ticket_no, flight_id)
) with (distributed_by='ticket_no', colocate_with='bookings.tickets');

CREATE TABLE bookings.boarding_passes (
    ticket_no character(13) NOT NULL,
    flight_id bigint NOT NULL,
    boarding_no integer NOT NULL,
    seat_no character varying(4) NOT NULL,
    FOREIGN KEY (ticket_no, flight_id) REFERENCES bookings.ticket_flights(ticket_no, flight_id),
    PRIMARY KEY (ticket_no, flight_id)
) with (distributed_by='ticket_no', colocate_with='bookings.tickets');

-- ограничения обязательно должны содержать ключ сегментирования
ALTER TABLE bookings.boarding_passes ADD CONSTRAINT boarding_passes_flight_id_boarding_no_key UNIQUE (ticket_no, flight_id, boarding_no);

ALTER TABLE bookings.boarding_passes ADD CONSTRAINT boarding_passes_flight_id_seat_no_key UNIQUE (ticket_no, flight_id, seat_no);

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

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

Последнее, что осталось сделать – создать представление, которое понадобится при выполнении некоторых запросов: 

    CREATE VIEW bookings.flights_v AS
    SELECT f.flight_id,
        f.flight_no,
        f.scheduled_departure,
        timezone(dep.timezone, f.scheduled_departure) AS scheduled_departure_local,
        f.scheduled_arrival,
        timezone(arr.timezone, f.scheduled_arrival) AS scheduled_arrival_local,
        (f.scheduled_arrival - f.scheduled_departure) AS scheduled_duration,
        f.departure_airport,
        dep.airport_name AS departure_airport_name,
        dep.city AS departure_city,
        f.arrival_airport,
        arr.airport_name AS arrival_airport_name,
        arr.city AS arrival_city,
        f.status,
        f.aircraft_code,
        f.actual_departure,
        timezone(dep.timezone, f.actual_departure) AS actual_departure_local,
        f.actual_arrival,
        timezone(arr.timezone, f.actual_arrival) AS actual_arrival_local,
        (f.actual_arrival - f.actual_departure) AS actual_duration
      FROM bookings.flights f,
        bookings.airports dep,
        bookings.airports arr
      WHERE ((f.departure_airport = dep.airport_code) AND (f.arrival_airport = arr.airport_code));

Теперь создание распределённой схемы завершено. Отключите трансляцию DDL-операторов: 

    SET shardman.broadcast_ddl TO off;
17.3.2.3.2. Комплексный подход — ключ сегментирования book_ref #

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

Допустим, авиакомпания работает на рынке более 10 лет, и таблица bookings в этом случае вырастет в объёме до такого размера, что сделать её глобальной не представляется возможным. Но и распределить данные этой таблицы нельзя, так как она не содержит общих полей с другими таблицами, по которым её можно распределить (как в варианте 1).

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

Если посмотреть на таблицу bookings, можно заметить, что поле book_ref — уникальное и является первичным ключом для данной таблицы, а также служит внешним ключом к таблице tickets. Поэтому это поле кажется подходящим для выбора в качестве ключа сегментирования. Однако поле book_ref отсутствует в таблицах ticket_flights и boarding_passes.

Если добавить поле book_ref в таблицы ticket_flights и boarding_passes, то станет возможным распределение всех таблиц: bookings, tickets, ticket_flights и boarding_passes по ключу сегментирования book_ref.

Операцию по добавлению ключа book_ref в таблицы ticket_flights и boarding_passes следует производить в исходной схеме, при этом, поле book_ref должно быть заполнено данными из таблицы bookings.

Рисунок 17.4. Изменение исходной схемы

Изменение исходной схемы


17.3.2.3.2.1. Изменение исходной схемы #

Для корректной передачи данных из исходной схемы в распределённую, исходную схему необходимо изменить следующим образом:

  1. Добавьте поле book_ref в таблицы ticket_flights и boarding_passes: 

        ALTER TABLE ticket_flights
      ADD COLUMN book_ref char(6);

    ALTER TABLE boarding_passes
      ADD COLUMN book_ref char(6);

  2. В этих таблицах заполните добавленное поле book_ref данными: 

        WITH batch AS (SELECT book_ref,
                          ticket_no
                    FROM tickets)
    UPDATE ticket_flights
      SET book_ref = batch.book_ref
      FROM batch
    WHERE ticket_flights.ticket_no = batch.ticket_no
      AND ticket_flights.book_ref IS NULL;


    WITH batch AS (SELECT book_ref,
                          ticket_no
                    FROM tickets)
    UPDATE boarding_passes
      SET book_ref = batch.book_ref
      FROM batch
    WHERE boarding_passes.ticket_no = batch.ticket_no
      AND boarding_passes.book_ref IS NULL;

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

Теперь схема БД готова к переносу.

17.3.2.3.2.2. Создание схемы, распределённой по ключу book_ref #

Здесь используется функция Postgres Pro Shardman shardman.broadcast_all_sql для трансляции DDL-операторов на все узлы кластера. Создайте схему bookings для всех сегментов: 

    SELECT shardman.broadcast_all_sql('CREATE SCHEMA bookings');

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

Сначала создайте вспомогательную функцию bookings.now(): 

    SELECT shardman.broadcast_all_sql(
        $sql$
        CREATE FUNCTION bookings.now() RETURNS timestamp with time zone
        LANGUAGE sql IMMUTABLE COST 0.00999999978
        AS 
        $q$
        SELECT $qq$2016-10-13 17:00:00$qq$::TIMESTAMP 
                        AT TIME ZONE $zz$Europe/Moscow$zz$;
        $q$;
        $sql$
    );

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

В данном примере глобальная последовательность не создаётся явно, так как Postgres Pro Shardman при указании типа поля bigserial создаёт глобальную последовательность автоматически.

Теперь создайте глобальные таблицы, используя следующие DDL-операторы: 

    CREATE TABLE bookings.aircrafts (
        aircraft_code character(3) NOT NULL PRIMARY KEY,
        model text NOT NULL,
        range integer NOT NULL,
        CONSTRAINT aircrafts_range_check CHECK ((range > 0))
    ) WITH (global);

    CREATE TABLE bookings.seats (
        aircraft_code character(3) REFERENCES bookings.aircrafts(aircraft_code),
        seat_no character varying(4) NOT NULL,
        fare_conditions character varying(10) NOT NULL,
        CONSTRAINT seats_fare_conditions_check CHECK ((
              (fare_conditions)::text = ANY (ARRAY[
                ('Economy'::character varying)::text,
                ('Comfort'::character varying)::text,
                ('Business'::character varying)::text])
              )),
        PRIMARY KEY (aircraft_code, seat_no)
    ) WITH (global);

    CREATE TABLE bookings.airports (
        airport_code character(3) NOT NULL PRIMARY KEY,
        airport_name text NOT NULL,
        city text NOT NULL,
        longitude double precision NOT NULL,
        latitude double precision NOT NULL,
        timezone text NOT NULL
    )  WITH (global);

    CREATE TABLE bookings.flights (
    -- the global sequence will be created automatically
    -- the default value will be assigned
        flight_id bigserial NOT NULL PRIMARY KEY,
        flight_no character(6) NOT NULL,
        scheduled_departure timestamp with time zone NOT NULL,
        scheduled_arrival timestamp with time zone NOT NULL,
        departure_airport character(3) REFERENCES bookings.airports(airport_code),
        arrival_airport character(3) REFERENCES bookings.airports(airport_code),
        status character varying(20) NOT NULL,
        aircraft_code character(3) REFERENCES bookings.aircrafts(aircraft_code),
        actual_departure timestamp with time zone,
        actual_arrival timestamp with time zone,
        CONSTRAINT flights_check CHECK ((scheduled_arrival > scheduled_departure)),
        CONSTRAINT flights_check1 CHECK ((
                      (actual_arrival IS NULL) 
                  OR ((actual_departure IS NOT NULL) 
                  AND (actual_arrival IS NOT NULL) 
                  AND (actual_arrival > actual_departure)))),
        CONSTRAINT flights_status_check CHECK (
              ((status)::text = ANY (
          ARRAY[('On Time'::character varying)::text,
                ('Delayed'::character varying)::text,
                ('Departed'::character varying)::text,
                ('Arrived'::character varying)::text,
                ('Scheduled'::character varying)::text,
                ('Cancelled'::character varying)::text])))
    ) WITH (global);

    ALTER TABLE bookings.flights 
      ADD CONSTRAINT flights_flight_no_scheduled_departure_key 
      UNIQUE (flight_no, scheduled_departure);

Теперь создайте сегментированные таблицы bookings, tickets, ticket_flights и boarding_passes в схеме bookings, как в предыдущем примере: 

    -- в данных таблицах не предполагается изменений, кроме их распределения
CREATE TABLE bookings.bookings (
   book_ref character(6) NOT NULL PRIMARY KEY,
   book_date timestamp with time zone NOT NULL,
   total_amount numeric(10,2) NOT NULL
) WITH (distributed_by='book_ref', num_parts=4);


CREATE TABLE bookings.tickets (
   ticket_no character(13) NOT NULL,
   book_ref character(6) REFERENCES bookings.bookings(book_ref),
   passenger_id character varying(20) NOT NULL,
   passenger_name text NOT NULL,
   contact_data jsonb,
   PRIMARY KEY (book_ref, ticket_no)
) WITH (distributed_by='book_ref', colocate_with='bookings.bookings');

-- в данных таблицах добавляется дополнительный внешний ключ book_ref
CREATE TABLE bookings.ticket_flights (
   ticket_no character(13) NOT NULL,
   flight_id bigint NOT NULL,
   fare_conditions character varying(10) NOT NULL,
   amount numeric(10,2) NOT NULL,
   book_ref character(6) NOT NULL, -- <= added book_ref
   CONSTRAINT ticket_flights_amount_check 
        CHECK ((amount >= (0)::numeric)),
   CONSTRAINT ticket_flights_fare_conditions_check 
        CHECK (((fare_conditions)::text = ANY (
        ARRAY[('Economy'::character varying)::text,
              ('Comfort'::character varying)::text,
              ('Business'::character varying)::text]))),
   FOREIGN KEY (book_ref, ticket_no) 
        REFERENCES bookings.tickets(book_ref, ticket_no),
   PRIMARY KEY (book_ref, ticket_no, flight_id) -- <= changed the primary key
) with (distributed_by='book_ref', colocate_with='bookings.bookings');


CREATE TABLE bookings.boarding_passes (
    ticket_no character(13) NOT NULL,
    flight_id bigint NOT NULL,
    boarding_no integer NOT NULL,
    seat_no character varying(4) NOT NULL,
    FOREIGN KEY (book_ref, ticket_no, flight_id) 
         REFERENCES bookings.ticket_flights(book_ref, ticket_no, flight_id),
    PRIMARY KEY (book_ref, ticket_no, flight_id)
) WITH (distributed_by='book_ref', colocate_with='bookings.bookings');

CREATE TABLE bookings.boarding_passes (
  ticket_no character(13) NOT NULL,
  flight_id bigint NOT NULL,
  boarding_no integer NOT NULL,
  seat_no character varying(4) NOT NULL,
  book_ref character(6) NOT NULL, – <= added book_ref
FOREIGN KEY (book_ref, ticket_no, flight_id)
REFERENCES bookings.ticket_flights(book_ref, ticket_no, flight_id),
PRIMARY KEY (book_ref, ticket_no, flight_id)
) WITH (distributed_by='book_ref', colocate_with='bookings.bookings');

-- ограничения должны обязательно содержать ключ сегментирования
ALTER TABLE bookings.boarding_passes 
  ADD CONSTRAINT boarding_passes_flight_id_boarding_no_key 
  UNIQUE (book_ref, ticket_no, flight_id, boarding_no);

ALTER TABLE bookings.boarding_passes 
  ADD CONSTRAINT boarding_passes_flight_id_seat_no_key 
  UNIQUE (book_ref, ticket_no, flight_id, seat_no);

Создайте представление bookings.flights: 

    SELECT shardman.broadcast_all_sql($$
CREATE VIEW bookings.flights_v AS
SELECT f.flight_id,
      f.flight_no,
      f.scheduled_departure,
      timezone(dep.timezone, f.scheduled_departure) AS scheduled_departure_local,
      f.scheduled_arrival,
      timezone(arr.timezone, f.scheduled_arrival)   AS scheduled_arrival_local,
      (f.scheduled_arrival - f.scheduled_departure) AS scheduled_duration,
      f.departure_airport,
      dep.airport_name                              AS departure_airport_name,
      dep.city                                      AS departure_city,
      f.arrival_airport,
      arr.airport_name                              AS arrival_airport_name,
      arr.city                                      AS arrival_city,
      f.status,
      f.aircraft_code,
      f.actual_departure,
      timezone(dep.timezone, f.actual_departure)    AS actual_departure_local,
      f.actual_arrival,
      timezone(arr.timezone, f.actual_arrival)      AS actual_arrival_local,
      (f.actual_arrival - f.actual_departure)       AS actual_duration
FROM bookings.flights f,
    bookings.airports dep,
    bookings.airports arr
WHERE ((f.departure_airport = dep.airport_code) AND (f.arrival_airport = arr.airport_code));
$$);

Теперь создание схемы завершено. Следующая тема — миграция данных.

17.3.3. Миграция данных #

При миграции данных важен порядок полей в исходной и целевой схеме. Порядок полей и их тип в таблицах нераспределённой и распределённой БД должен быть одинаковым.

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

Postgres Pro Shardman предоставляет удобные инструменты для миграции данных. При уже созданной схеме и выбранном ключе сегментирования остаётся определить правила миграции данных. Источником данных могут быть, как файлы экспорта данных в формате CSV, так и одиночный сервер СУБД.

Далеко не всегда удобно использовать CSV-файлы, так как они могут значительно увеличиваться в объёме и требовать дополнительных ресурсов для хранения и передачи.

Гораздо удобнее использовать прямую миграцию данных из БД в БД без этапа промежуточного хранения.

При миграции данных следует учитывать порядок загрузки данных. Таблицы могут быть связаны друг с другом внешним ключом, поэтому сначала нужно загружать данные таблиц, на которые будут ссылаться другие таблицы. Для этого в файле миграции следует задать приоритет, определяющий порядок загрузки данных из таблиц. Чем больше значение параметра priority, тем выше приоритет. Например, если заданы приоритеты 1, 2 и 3, то сначала будут загружены данные из таблиц с приоритетом 3, потом с приоритетом 2, последними — с приоритетом 1.

Команда shardmanctl load позволяет определить порядок миграции таблиц, который можно указать в конфигурационном YML-файле.

17.3.3.1. Наивный подход #

Ниже представлен пример содержимого файла migrate.yml: 

    version: "1.0"
migrate:
 connstr: "dbname=demo host=single-pg-instance port=5432 user=postgres password=******"
 jobs: 8
 batch: 2000
 options:
 schemas:
   - name: bookings
     # параметр all со значением false отключает автоматическое создание таблиц
     # таблицы уже созданы ранее, на этапе миграции схемы
     all: false
     tables:
       - name: airports
         # определяем глобальную таблицу
         type: global
         # так как таблицы связаны, нужно задать порядок миграции данных,
          # выставляем наивысший приоритет таблицам, данные которых
         # должны быть скопированы в первую очередь
         priority: 3
       - name: aircrafts
         type: global
         priority: 3
       - name: seats
         type: global
         priority: 3
       - name: bookings
         type: global
         priority: 3
       - name: flights
         type: global
         priority: 3
       - name: tickets
         type: sharded
         # определяем сегментированную таблицу
         # указываем ключ распределения
         distributedby: ticket_no
         partitions: 4
         priority: 2
       - name: ticket_flights
         type: sharded
         distributedby: ticket_no
         # определяем сегментированную и совместно размещённую таблицу
         # указываем имя таблицы с которой будет совместно размещена таблица tickets_flights
         colocatewith: tickets
         partitions: 4
         priority: 2
       - name: boarding_passes
         type: sharded
         distributedby: ticket_no
         colocatewith: tickets
         partitions: 4
         priority: 1

В данном файле определён источник данных — узел с именем single-pg-instance, его порт для подключения, имя пользователя и пароль, а также имя БД источника данных. Определены также и некоторые параметры работы утилиты (их может быть довольно много, как показано Подразделе «Загрузка данных из схемы PostgreSQL»). В файле также задано количество потоков — 8, то есть размер порции обрабатываемых данных в строках при миграции, а также приоритеты для последовательной обработки таблиц. Сначала мигрируют данные в глобальные таблицы, следом данные в распределённые таблицы tickets и ticket_flights, затем данные в таблицу boarding_passes. Значение priority задаёт приоритет загрузки данных, чем больше значение ключа, тем раньше будут загружены данные соответствующих таблиц. Миграция выполняется с помощью следующей команды: 

    shardmanctl load --schema migrate.yml

Если в конце работы утилиты выводится сообщение: «data loading completed successfully» (загрузка данных успешно завершена), то миграция данных прошла успешно.

17.3.3.2. Комплексный подход #

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

    ---
version: "1.0"
migrate:
 connstr: "dbname=demo host=single-pg-instance port=5432 user=postgres password=postgres"
 jobs: 8
 batch: 2000
 options:
 schemas:
   - name: bookings
     all: false
     tables:
       - name: airports
         type: global
         priority: 5
       - name: aircrafts
         type: global
         priority: 5
       - name: seats
         type: global
         priority: 5
       - name: flights
         type: global
         priority: 5
       - name: bookings
         type: sharded
         priority: 4
         partitions: 4
         distributedby: book_ref
       - name: tickets
         type: sharded
         distributedby: book_ref
         colocatewith: bookings
         partitions: 4
         priority: 3
       - name: ticket_flights
         type: sharded
         distributedby: book_ref
         colocatewith: bookings
         partitions: 4
         priority: 2
       - name: boarding_passes
         type: sharded
         distributedby: book_ref
         colocatewith: bookings
         partitions: 4
         priority: 1

17.3.4. Проверка миграции Postgres Pro Shardman #

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

17.3.4.1. Запрос q1 #

Запрос q1 довольно прост, он возвращает бронирование с указанным номером: 

    SELECT *
      FROM bookings.bookings b
    WHERE b.book_ref = '0824C5';

Для обычной СУБД PostgreSQL и ключа сегментирования ticket_no этот запрос выполняется сравнительно быстро. Скорость запроса для ключа сегментирования book_ref зависит от сегмента, в котором он выполняется. Если он выполняется в сегменте, где физически нет данных,Postgres Pro Shardman отправляет запрос в другой сегмент, что вызывает задержку по времени из-за сетевого взаимодействия.

17.3.4.2. Запрос q2 #

В результате запроса q2 возвращаются все билеты из указанного бронирования: 

    SELECT t.*
    FROM bookings.bookings b
    JOIN bookings.tickets t
      ON t.book_ref = b.book_ref
    WHERE b.book_ref = '0824C5';

С помощью ключа сегментирования book_ref запрос передаётся в сегменты, и глобальная таблица объединяется с секциями сегментированной таблицы: 

    Foreign Scan (actual rows=2 loops=1)
  Relations: (bookings_2_fdw b) INNER JOIN (tickets_2_fdw t)
  Network: FDW bytes sent=433 received=237

Посмотрите на план запроса для ключа сегментирования ticket_no: 

    Append (actual rows=2 loops=1)
  Network: FDW bytes sent=1263 received=205
  ->  Nested Loop (actual rows=1 loops=1)
        ->  Seq Scan on tickets_0 t_1 (actual rows=1 loops=1)
              Filter: (book_ref = '0824C5'::bpchar)
              Rows Removed by Filter: 207092
        ->  Index Only Scan using bookings_pkey on bookings b (actual rows=1 loops=1)
              Index Cond: (book_ref = '0824C5'::bpchar)
              Heap Fetches: 0
  ->  Async Foreign Scan (actual rows=1 loops=1)
        Relations: (tickets_1_fdw t_2) INNER JOIN (bookings b)
        Network: FDW bytes sent=421 received=205
  ->  Async Foreign Scan (actual rows=0 loops=1)
        Relations: (tickets_2_fdw t_3) INNER JOIN (bookings b)
        Network: FDW bytes sent=421
  ->  Async Foreign Scan (actual rows=0 loops=1)
        Relations: (tickets_3_fdw t_4) INNER JOIN (bookings b)
        Network: FDW bytes sent=421

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

Посмотрите на строку Network. Хорошим критерием оптимальности выполнения запросов в сегментах может быть значение received. Чем меньше это значение, тем лучше сегменты справляются с выполнением распределённых запросов. Большая часть обработки выполняется удалённо, и на узел-источник запроса возвращается уже готовый для дальнейшей обработки результат.

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

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

    Foreign Scan (actual rows=2 loops=1)
  Relations: (bookings_2_fdw b) INNER JOIN (tickets_2_fdw t)
  Network: FDW bytes sent=433 received=237

Происходит сетевой обмен только с одним сегментом, в котором выполняется запрос. Это сегмент shard-3 и секция tickets_2 таблицы tickets на четвёртом узле.

Если этот запрос выполняется в сегменте, где физически присутствуют данные, то скорость выполнения будет ещё выше.

Посмотрите на план: 

    Nested Loop (actual rows=2 loops=1)
    ->  Index Only Scan using bookings_2_pkey on bookings_2
    ->  Bitmap Heap Scan on tickets_2
          ->  Bitmap Index Scan on tickets_2_book_ref_idx

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

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

17.3.4.3. Запрос q3 #

В результате запроса q3 находятся все рейсы для одного из билетов в выбранном ранее бронировании: 

    SELECT tf.*, t.*
FROM bookings.tickets t
JOIN bookings.ticket_flights tf
  ON tf.ticket_no = t.ticket_no
WHERE t.ticket_no = '0005435126781';

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

Для ключа сегментирования book_ref обратите внимание, что по номеру бронирования можно вычислить номер билета и запросить его сразу с «правильного» сегмента.

В таком случае запрос выглядит следующим образом: 

    SELECT tf.*, t.*
FROM bookings.tickets t
JOIN bookings.ticket_flights tf
  ON tf.ticket_no = t.ticket_no
  AND t.book_ref = tf.book_ref
WHERE t.ticket_no = '0005435126781'
AND tf.book_ref = '0824C5';

Запрос выполняется медленнее в сегменте, не содержащем секцию с искомыми данными: 

    Foreign Scan (actual rows=6 loops=1)
  Relations: (tickets_1_fdw t) INNER JOIN (ticket_flights_1_fdw tf)
  Network: FDW bytes sent=434 received=369

Сетевая связь между сегментами присутствует в плане, так как он содержит узел Foreign Scan.

Важность указания в запросе ключа сегментирования, в случае с book_ref, можно проиллюстрировать таким запросом: 

    SELECT tf.*, t.*
FROM bookings.tickets t
JOIN bookings.ticket_flights tf
  ON tf.ticket_no = t.ticket_no
WHERE t.ticket_no = '0005435126781'
AND tf.book_ref = '0824C5';

Здесь ключ сегментирования намеренно не указан в join. Взглянем на план: 

    Nested Loop (actual rows=6 loops=1)
  Network: FDW bytes sent=1419 received=600
  ->  Foreign Scan on ticket_flights_2_fdw tf (actual rows=6 loops=1)
        Network: FDW bytes sent=381 received=395
  ->  Append (actual rows=1 loops=6)
        Network: FDW bytes sent=1038 received=205
        ->  Seq Scan on tickets_0 t_1 (actual rows=0 loops=6)
              Filter: (ticket_no = '0005435126781'::bpchar)
              Rows Removed by Filter: 207273
        ->  Async Foreign Scan on tickets_1_fdw t_2 (actual rows=0 loops=6)
              Network: FDW bytes sent=346 received=205
        ->  Async Foreign Scan on tickets_2_fdw t_3 (actual rows=1 loops=6)
              Network: FDW bytes sent=346
        ->  Async Foreign Scan on tickets_3_fdw t_4 (actual rows=0 loops=6)
              Network: FDW bytes sent=346

В отличие от предыдущих примеров заметны отклонения. Здесь запрос выполнялся на всех узлах, причём индекс при этом не использовался, поэтому для того, чтобы вернуть всего 6 строк, , Postgres Pro Shardman был вынужден просканировать последовательно секции таблицы tickets целиком, вернуть результат источнику запроса, после чего выполнить соединение (join) с таблицей ticket_flights. Наличие узлов Async Foreign Scan говорит о том, что в сегментах выполняется последовательное сканирование таблицы tickets.

17.3.4.4. Запрос q4 #

В результате данного запроса возвращаются все перелёты по всем билетам указанным в бронировании. Есть несколько вариантов, как его выполнить: включить в условии WHERE подзапрос с указанием номера бронирования, перечислить явно в конструкции IN номера билетов или использовать конструкцию WHERE... OR. Ниже показано, как будут работать все перечисленные варианты.

    SELECT tf.*, t.*
FROM bookings.tickets t
JOIN bookings.ticket_flights tf
  ON tf.ticket_no = t.ticket_no
WHERE t.ticket_no IN (
   SELECT t.ticket_no
     FROM bookings.bookings b
     JOIN bookings.tickets  t
       ON t.book_ref = b.book_ref
    WHERE b.book_ref = '0824C5'
);

Данный пример — попытка выполнить запрос из нераспределённой БД. Но он выполняется одинаково плохо для обоих вариантов ключей сегментирования.

План запроса будет примерно таким: 

    Hash Join (actual rows=12 loops=1)
  Hash Cond: (tf.ticket_no = t.ticket_no)
  ->  Append (actual rows=2360335 loops=1)
        ->  Async Foreign Scan on ticket_flights_0_fdw tf_1 (actual rows=589983 loops=1)
        ->  Async Foreign Scan on ticket_flights_1_fdw tf_2 (actual rows=590175 loops=1)
        ->  Seq Scan on ticket_flights_2 tf_3 (actual rows=590174 loops=1)
        ->  Async Foreign Scan on ticket_flights_3_fdw tf_4 (actual rows=590003 loops=1)
  ->  Hash (actual rows=2 loops=1)
        Buckets: 1024  Batches: 1  Memory Usage: 9kB
        ->  Hash Semi Join (actual rows=2 loops=1)
              Hash Cond: (t.ticket_no = t_5.ticket_no)
              ->  Append (actual rows=829071 loops=1)
                    ->  Async Foreign Scan on tickets_0_fdw t_1 (actual rows=207273 loops=1)
                    ->  Async Foreign Scan on tickets_1_fdw t_2 (actual rows=207058 loops=1)
                    ->  Seq Scan on tickets_2 t_3 (actual rows=207431 loops=1)
                    ->  Async Foreign Scan on tickets_3_fdw t_4 (actual rows=207309 loops=1)
              ->  Hash (actual rows=2 loops=1)
                    Buckets: 1024  Batches: 1  Memory Usage: 9kB
                    ->  Nested Loop (actual rows=2 loops=1)
                          ->  Index Only Scan using tickets_2_pkey on tickets_2 t_5
                          ->  Materialize (actual rows=1 loops=2)
                                ->  Index Only Scan using bookings_2_pkey on bookings_2 b

Такой план говорит о том, что Postgres Pro Shardman справился с подзапросом в WHERE, и был вынужден запросить все строки таблиц tickets и ticket_flights, после чего выполнить их обработку на узле источнике запроса, что крайне неэффективно. Попробуем другие варианты:

Для запроса с ключом сегментирования ticket_no: 

        SELECT tf.*, t.*
    FROM bookings.tickets t
    JOIN bookings.ticket_flights tf
      ON tf.ticket_no = t.ticket_no
    WHERE t.ticket_no IN ('0005435126781','0005435126782');

план будет таким: 

        Append (actual rows=12 loops=1)
      Network: FDW bytes sent=1098 received=1656
      ->  Async Foreign Scan (actual rows=6 loops=1)
            Relations: (tickets_0_fdw t_1) INNER JOIN (ticket_flights_0_fdw tf_1)
            Network: FDW bytes sent=549 received=1656
      ->  Async Foreign Scan (actual rows=6 loops=1)
            Relations: (tickets_1_fdw t_2) INNER JOIN (ticket_flights_1_fdw tf_2)
            Network: FDW bytes sent=549

Этот план эффективнее, запрос выполняется на двух сегментах из четырёх, и требуется лишь выполнить Append для полученных результатов.

Вновь обратим внимание, что book_ref присутствует в обеих таблицах — tickets и ticket_flights. Запрос для ключа сегментирования book_ref: 

    SELECT tf.*, t.*
FROM bookings.tickets t
JOIN bookings.ticket_flights tf
ON tf.ticket_no = t.ticket_no
AND tf.book_ref = t.book_ref
WHERE t.book_ref = '0824C5';

при этом план запроса: 

    Foreign Scan (actual rows=12 loops=1)
  Relations: (tickets_2_fdw t) INNER JOIN (ticket_flights_2_fdw tf)
  Network: FDW bytes sent=547 received=1717

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

17.3.4.5. Запрос q5 #

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

    SELECT t.passenger_name, t.ticket_no
FROM bookings.tickets t
JOIN bookings.boarding_passes bp
  ON bp.ticket_no = t.ticket_no
GROUP BY t.passenger_name, t.ticket_no
HAVING max(bp.boarding_no) = 1
AND count(*) > 1;

Данный запрос выполняется одинаково медленно для обоих вариантов ключей сегментирования. План запроса с ключом сегментирования book_ref выглядит так: 

    HashAggregate (actual rows=424 loops=1)
  Group Key: t.ticket_no
  Filter: ((max(bp.boarding_no) = 1) AND (count(*) > 1))
  Batches: 85  Memory Usage: 4265kB  Disk Usage: 112008kB
  Rows Removed by Filter: 700748
  Network: FDW bytes sent=1215 received=77111136
  ->  Append (actual rows=1894295 loops=1)
        Network: FDW bytes sent=1215 received=77111136
        ->  Async Foreign Scan (actual rows=473327 loops=1)
              Relations: (tickets_0_fdw t_1) INNER JOIN (boarding_passes_0_fdw bp_1)
              Network: FDW bytes sent=404 received=813128
        ->  Async Foreign Scan (actual rows=472632 loops=1)
              Relations: (tickets_1_fdw t_2) INNER JOIN (boarding_passes_1_fdw bp_2)
              Network: FDW bytes sent=404
        ->  Async Foreign Scan (actual rows=475755 loops=1)
              Relations: (tickets_2_fdw t_3) INNER JOIN (boarding_passes_2_fdw bp_3)
              Network: FDW bytes sent=407
        ->  Hash Join (actual rows=472581 loops=1)
              Hash Cond: (bp_4.ticket_no = t_4.ticket_no)
              Network: FDW bytes received=28841344
              ->  Seq Scan on boarding_passes_3 bp_4 (actual rows=472581 loops=1)
              ->  Hash (actual rows=207118 loops=1)
                    Buckets: 65536  Batches: 4  Memory Usage: 3654kB
                    Network: FDW bytes received=9176680
                    ->  Seq Scan on tickets_3 t_4 (actual rows=207118 loops=1)
                          Network: FDW bytes received=9176680

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

    SELECT t.passenger_name, t.ticket_no
FROM bookings.tickets t
JOIN bookings.boarding_passes bp
  ON bp.ticket_no = t.ticket_no
  AND bp.book_ref=t.book_ref -- <= added book_ref
GROUP BY t.passenger_name, t.ticket_no
HAVING max(bp.boarding_no) = 1
AND count(*) > 1;

Посмотрите на план запроса: 

    GroupAggregate (actual rows=424 loops=1)
  Group Key: t.passenger_name, t.ticket_no
  Filter: ((max(bp.boarding_no) = 1) AND (count(*) > 1))
  Rows Removed by Filter: 700748
  Network: FDW bytes sent=1424 received=77092816
  ->  Merge Append (actual rows=1894295 loops=1)
        Sort Key: t.passenger_name, t.ticket_no
        Network: FDW bytes sent=1424 received=77092816
        ->  Foreign Scan (actual rows=472757 loops=1)
              Relations: (tickets_0_fdw t_1) INNER JOIN (boarding_passes_0_fdw bp_1)
              Network: FDW bytes sent=472 received=2884064
        ->  Sort (actual rows=472843 loops=1)
              Sort Key: t_2.passenger_name, t_2.ticket_no
              Sort Method: external merge  Disk: 21152kB
              Network: FDW bytes received=22753536
              ->  Hash Join (actual rows=472843 loops=1)
                    Hash Cond: ((bp_2.ticket_no = t_2.ticket_no) AND (bp_2.book_ref = t_2.book_ref))
                    Network: FDW bytes received=22753536
                    ->  Seq Scan on boarding_passes_1 bp_2 (actual rows=472843 loops=1)
                    ->  Hash (actual rows=207058 loops=1)
                          Buckets: 65536  Batches: 8  Memory Usage: 2264kB
                          Network: FDW bytes received=22753536
                          ->  Seq Scan on tickets_1 t_2 (actual rows=207058 loops=1)
                                Network: FDW bytes received=22753536
        ->  Foreign Scan (actual rows=474715 loops=1)
              Relations: (tickets_2_fdw t_3) INNER JOIN (boarding_passes_2_fdw bp_3)
              Network: FDW bytes sent=476 received=2884120
        ->  Foreign Scan (actual rows=473980 loops=1)
              Relations: (tickets_3_fdw t_4) INNER JOIN (boarding_passes_3_fdw bp_4)
              Network: FDW bytes sent=476 received=25745384

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

Для ключа сегментирования ticket_no план исходного запроса выглядит так: 

    HashAggregate (actual rows=424 loops=1)
  Group Key: t.ticket_no
  Filter: ((max(bp.boarding_no) = 1) AND (count(*) > 1))
  Batches: 85  Memory Usage: 4265kB  Disk Usage: 111824kB
  Rows Removed by Filter: 700748
  Network: FDW bytes sent=1188 received=77103620
  ->  Append (actual rows=1894295 loops=1)
        Network: FDW bytes sent=1188 received=77103620
        ->  Async Foreign Scan (actual rows=473327 loops=1)
              Relations: (tickets_0_fdw t_1) INNER JOIN (boarding_passes_0_fdw bp_1)
              Network: FDW bytes sent=394
        ->  Hash Join (actual rows=472632 loops=1)
              Hash Cond: (bp_2.ticket_no = t_2.ticket_no)
              Network: FDW bytes received=77103620
              ->  Seq Scan on boarding_passes_1 bp_2 (actual rows=472632 loops=1)
              ->  Hash (actual rows=206712 loops=1)
                    Buckets: 65536  Batches: 4  Memory Usage: 3654kB
                    Network: FDW bytes received=23859576
                    ->  Seq Scan on tickets_1 t_2 (actual rows=206712 loops=1)
                          Network: FDW bytes received=23859576
        ->  Async Foreign Scan (actual rows=475755 loops=1)
              Relations: (tickets_2_fdw t_3) INNER JOIN (boarding_passes_2_fdw bp_3)
              Network: FDW bytes sent=397
        ->  Async Foreign Scan (actual rows=472581 loops=1)
              Relations: (tickets_3_fdw t_4) INNER JOIN (boarding_passes_3_fdw bp_4)
              Network: FDW bytes sent=397

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

17.3.4.6. Запрос q6 #

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

    SELECT tf.ticket_no,f.departure_airport,
      f.arrival_airport,f.scheduled_arrival,
      lead(f.scheduled_departure) OVER w AS next_departure,
      lead(f.scheduled_departure) OVER w - f.scheduled_arrival AS gap
FROM bookings.bookings b
JOIN bookings.tickets t
 ON t.book_ref = b.book_ref
JOIN bookings.ticket_flights tf
 ON tf.ticket_no = t.ticket_no
JOIN bookings.flights f
 ON tf.flight_id = f.flight_id
WHERE b.book_date = bookings.now()::date - INTERVAL '7 day'

WINDOW w AS (
PARTITION BY tf.ticket_no
ORDER BY f.scheduled_departure);

Для этого запроса тип столбца book_date из timestamptz должен быть приведён к типу date. При приведении типа PostgreSQL приводит тип данных столбца к типу данных, указанному в условии фильтрации, но не наоборот. Поэтому Postgres Pro Shardman сначала должен получить все данные из других сегментов, привести тип, и только после этого применить правило фильтрации. План запроса будет выглядеть примерно так: 

    WindowAgg (actual rows=26 loops=1)
  Network: FDW bytes sent=1750 received=113339240
  ->  Sort (actual rows=26 loops=1)
        Sort Key: tf.ticket_no, f.scheduled_departure
        Sort Method: quicksort  Memory: 27kB
        Network: FDW bytes sent=1750 received=113339240
        ->  Append (actual rows=26 loops=1)
              Network: FDW bytes sent=1750 received=113339240
              ->  Hash Join (actual rows=10 loops=1)
                    Hash Cond: (t_1.book_ref = b.book_ref)
                    Network: FDW bytes sent=582 received=37717376
              ->  Hash Join (actual rows=6 loops=1)
                    Hash Cond: (t_2.book_ref = b.book_ref)
                    Network: FDW bytes sent=582 received=37700608
              ->  Hash Join (actual rows=2 loops=1)
                    Hash Cond: (t_3.book_ref = b.book_ref)
                    Network: FDW bytes sent=586 received=37921256
              ->  Nested Loop (actual rows=8 loops=1)
                    ->  Nested Loop (actual rows=8 loops=1)
                          ->  Hash Join (actual rows=2 loops=1)
                                Hash Cond: (t_4.book_ref = b.book_ref)
                                ->  Seq Scan on tickets_3 t_4 (actual rows=207118 loops=1)
                    ->  Index Scan using flights_pkey on flights f (actual rows=1 loops=8)
                          Index Cond: (flight_id = tf_4.flight_id)

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

Интервал будет вычисляться не на уровне базы данных, а на уровне приложения, запросу сразу будут передаваться данные типа timestamptz. Кроме того, может помочь создание дополнительного индекса: 

    CREATE INDEX if not exists bookings_date_idx ON bookings.bookings(book_date);

Запрос с ключом сегментирования book_ref будет выглядеть так: 

    SELECT tf.ticket_no,f.departure_airport,
      f.arrival_airport,f.scheduled_arrival,
      lead(f.scheduled_departure) OVER w AS next_departure,
      lead(f.scheduled_departure) OVER w - f.scheduled_arrival AS gap
FROM bookings.bookings b
JOIN bookings.tickets t
 ON t.book_ref = b.book_ref
JOIN bookings.ticket_flights tf
 ON tf.ticket_no = t.ticket_no
AND tf.book_ref = t.book_ref -- <= added book_ref
JOIN bookings.flights f
 ON tf.flight_id = f.flight_id
WHERE b.book_date = '2016-10-06 14:00:00+00'
WINDOW w AS (
PARTITION BY tf.ticket_no
ORDER BY f.scheduled_departure);

У данного запроса будет уже другой план: 

    WindowAgg (actual rows=18 loops=1)
  Network: FDW bytes sent=2268 received=892
  ->  Sort (actual rows=18 loops=1)
        Sort Key: tf.ticket_no, f.scheduled_departure
        Sort Method: quicksort  Memory: 26kB
        Network: FDW bytes sent=2268 received=892
        ->  Append (actual rows=18 loops=1)
              Network: FDW bytes sent=2268 received=892
              ->  Nested Loop (actual rows=4 loops=1)
                    ->  Nested Loop (actual rows=4 loops=1)
                          ->  Nested Loop (actual rows=1 loops=1)
                                ->  Bitmap Heap Scan on bookings_0 b_1
                                      Heap Blocks: exact=1
                                      ->  Bitmap Index Scan on bookings_0_book_date_idx
                                ->  Index Only Scan using tickets_0_pkey on tickets_0 t_1
                                      Index Cond: (book_ref = b_1.book_ref)
                                      Heap Fetches: 0
                          ->  Index Only Scan using ticket_flights_0_pkey on ticket_flights_0 tf_1
                                Heap Fetches: 0
                    ->  Index Scan using flights_pkey on flights f (actual rows=1 loops=4)
                          Index Cond: (flight_id = tf_1.flight_id)
              ->  Async Foreign Scan (actual rows=14 loops=1)
                    Network: FDW bytes sent=754 received=892
              ->  Async Foreign Scan (actual rows=0 loops=1)
                    Network: FDW bytes sent=757 -- received=0!
              ->  Async Foreign Scan (actual rows=0 loops=1)
                    Network: FDW bytes sent=757 -- received=0!

Он гораздо эффективнее. Во-первых, таблица не сканируется целиком, используется Index Only Scan. Во-вторых, видно, насколько снизился объём переданных по сети данных между узлами.

17.3.4.7. Запрос q7 #

Допустим, нужна статистика по количеству пассажиров на одно бронирование. Для её получения сначала посчитаем количество пассажиров в каждом бронировании, а затем — количество бронирований с каждым вариантом количества пассажиров. 

    SELECT tt.cnt, count(*)
FROM (
   SELECT count(*) cnt
   FROM bookings.tickets t
   GROUP BY t.book_ref
   ) tt
GROUP BY tt.cnt
ORDER BY tt.cnt;

В результате данного запроса будут обработаны все данные таблиц tickets и bookings, поэтому не получится избежать интенсивного обмена данными по сети между сегментами кластера. Также обратите внимание, что значение параметра work_mem должно быть достаточно большим во избежание использования диска при соединении таблиц. Измените значение параметра work_mem в кластере: 

    shardmanctl set work_mem='256MB';

План запроса с ключом сегментирования ticket_no: 

    GroupAggregate (actual rows=5 loops=1)
  Group Key: tt.cnt
  Network: FDW bytes sent=798 received=18338112
  ->  Sort (actual rows=593433 loops=1)
        Sort Key: tt.cnt
        Sort Method: quicksort  Memory: 57030kB
        Network: FDW bytes sent=798 received=18338112
        ->  Subquery Scan on tt (actual rows=593433 loops=1)
              Network: FDW bytes sent=798 received=18338112
              ->  Finalize HashAggregate (actual rows=593433 loops=1)
                    Group Key: t.book_ref
                    Batches: 1  Memory Usage: 81953kB
                    Network: FDW bytes sent=798 received=18338112
                    ->  Append (actual rows=763806 loops=1)
                          Network: FDW bytes sent=798 received=18338112
                          ->  Async Foreign Scan (actual rows=190886 loops=1)
                                Relations: Aggregate on (tickets_0_fdw t)
                                Network: FDW bytes sent=266 received=1558336
                          ->  Async Foreign Scan (actual rows=190501 loops=1)
                                Relations: Aggregate on (tickets_1_fdw t_1)
                                Network: FDW bytes sent=266
                          ->  Async Foreign Scan (actual rows=191589 loops=1)
                                Relations: Aggregate on (tickets_2_fdw t_2)
                                Network: FDW bytes sent=266
                          ->  Partial HashAggregate (actual rows=190830 loops=1)
                                Group Key: t_3.book_ref
                                Batches: 1  Memory Usage: 36881kB
                                Network: FDW bytes received=4981496
                                ->  Seq Scan on tickets_3 t_3 (actual rows=207118 loops=1)
                                      Network: FDW bytes received=4981496

План запроса с ключом сегментирования book_ref: 

    Sort (actual rows=5 loops=1)
  Sort Key: (count(*))
  Sort Method: quicksort  Memory: 25kB
  Network: FDW bytes sent=798 received=14239951
  ->  HashAggregate (actual rows=5 loops=1)
        Group Key: (count(*))
        Batches: 1  Memory Usage: 40kB
        Network: FDW bytes sent=798 received=14239951
        ->  Append (actual rows=593433 loops=1)
              Network: FDW bytes sent=798 received=14239951
              ->  GroupAggregate (actual rows=148504 loops=1)
                    Group Key: t.book_ref
                    ->  Index Only Scan using tickets_0_book_ref_idx on tickets_0 t (rows=207273)
                          Heap Fetches: 0
              ->  Async Foreign Scan (actual rows=148256 loops=1)
                    Relations: Aggregate on (tickets_1_fdw t_1)
                    Network: FDW bytes sent=266 received=1917350
              ->  Async Foreign Scan (actual rows=148270 loops=1)
                    Relations: Aggregate on (tickets_2_fdw t_2)
                    Network: FDW bytes sent=266
              ->  Async Foreign Scan (actual rows=148403 loops=1)
                    Relations: Aggregate on (tickets_3_fdw t_3)
                    Network: FDW bytes sent=266

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

Для ключа сегментирования ticket_no принимаются все частично агрегированные данные объединяемых таблиц (17 Мб), а вся остальная обработка выполняется на узле-источнике запроса.

Для ключа сегментирования book_ref, поскольку он включён в запрос, большая часть вычислений агрегатов выполняется на узлах, и на узел-источник запроса возвращается только результат (13 МБ), который затем финально обрабатывается.

17.3.4.8. Запрос q8 #

Этот запрос отвечает на следующие вопросы: какие сочетания имён и фамилий встречаются чаще всего и как понять какую долю от числа всех пассажиров составляют такие сочетания? Для получения результата в запросе используется оконная функция: 

    SELECT passenger_name,
      round( 100.0 * cnt / sum(cnt) OVER (), 2)
   AS percent
FROM (
 SELECT passenger_name,
        count(*) cnt
 FROM bookings.tickets
 GROUP BY passenger_name
) t
ORDER BY percent DESC;

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

    Sort (actual rows=27909 loops=1)
  Sort Key: (round(((100.0 * ((count(*)))::numeric) / sum((count(*))) OVER (?)), 2)) DESC
  Sort Method: quicksort  Memory: 3076kB
  Network: FDW bytes sent=816 received=2376448
  ->  WindowAgg (actual rows=27909 loops=1)
        Network: FDW bytes sent=816 received=2376448
        ->  Finalize HashAggregate (actual rows=27909 loops=1)
              Group Key: tickets.passenger_name
              Batches: 1  Memory Usage: 5649kB
              Network: FDW bytes sent=816 received=2376448
              ->  Append (actual rows=74104 loops=1)
                    Network: FDW bytes sent=816 received=2376448
                    ->  Partial HashAggregate (actual rows=18589 loops=1)
                          Group Key: tickets.passenger_name
                          Batches: 1  Memory Usage: 2833kB
                          ->  Seq Scan on tickets_0 tickets (actual rows=207273 loops=1)
                    ->  Async Foreign Scan (actual rows=18435 loops=1)
                          Relations: Aggregate on (tickets_1_fdw tickets_1)
                          Network: FDW bytes sent=272 received=2376448
                    ->  Async Foreign Scan (actual rows=18567 loops=1)
                          Relations: Aggregate on (tickets_2_fdw tickets_2)
                          Network: FDW bytes sent=272
                    ->  Async Foreign Scan (actual rows=18513 loops=1)
                          Relations: Aggregate on (tickets_3_fdw tickets_3)
                          Network: FDW bytes sent=272

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

17.3.4.9. Запрос q9 #

Данный запрос отвечает на следующие вопросы: кто летел позавчера рейсом Москва (SVO) — Новосибирск (OVB) на месте 1A, и когда он забронировал себе билет? «Позавчера» считается от функции booking.now, а не от текущей даты. Исходный запрос в нераспределённой схеме выглядит так: 

    SELECT
   t.passenger_name,
   b.book_date v
FROM bookings b
JOIN tickets t ON
   t.book_ref = b.book_ref
JOIN boarding_passes bp
   ON bp.ticket_no = t.ticket_no
JOIN flights f ON
   f.flight_id = bp.flight_id
WHERE f.departure_airport = 'SVO'
AND f.arrival_airport = 'OVB'
AND f.scheduled_departure::date = bookings.now()::date - INTERVAL '2 day'
AND bp.seat_no = '1A';

Как уже объяснялось в описании Запроса q6, использование INTERVAL вызовет приведение типов. Поэтому следует сразу от него избавиться и переписать запрос для ключа сегментирования book_ref следующим образом: 

    SELECT
   t.passenger_name,
   b.book_date v
FROM bookings b
JOIN tickets t ON
   t.book_ref = b.book_ref
JOIN boarding_passes bp
   ON bp.ticket_no = t.ticket_no
   AND bp.book_ref = b.book_ref -- <= added book_ref
JOIN flights f ON
   f.flight_id = bp.flight_id
WHERE f.departure_airport = 'SVO'
AND f.arrival_airport = 'OVB'
AND f.scheduled_departure
  BETWEEN '2016-10-11 14:00:00+00' AND '2016-10-13 14:00:00+00'
AND bp.seat_no = '1A';

Также создайте пару вспомогательных индексов: 

    CREATE INDEX idx_boarding_passes_seats
     ON boarding_passes((seat_no::text));
CREATE INDEX idx_flights_sched_dep
     ON flights(departure_airport,arrival_airport,scheduled_departure);

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

    Append (actual rows=1 loops=1)
  Network: FDW bytes sent=2484 received=102
  ->  Nested Loop (actual rows=1 loops=1)
        Join Filter: (bp_1.ticket_no = t_1.ticket_no)
        Rows Removed by Join Filter: 1
        ->  Nested Loop (actual rows=1 loops=1)
              ->  Hash Join (actual rows=1 loops=1)
                    Hash Cond: (bp_1.flight_id = f.flight_id)
                    ->  Bitmap Heap Scan on boarding_passes_0 bp_1 (actual rows=4919 loops=1)
                          Recheck Cond: ((seat_no)::text = '1A'::text)
                          Heap Blocks: exact=2632
                          ->  Bitmap Index Scan on boarding_passes_0_seat_no_idx (actual rows=4919)
                                Index Cond: ((seat_no)::text = '1A'::text)
                    ->  Hash (actual rows=2 loops=1)
                          Buckets: 1024  Batches: 1  Memory Usage: 9kB
                          ->  Bitmap Heap Scan on flights f (actual rows=2 loops=1)
                                Recheck Cond:
                     ((departure_airport = 'SVO'::bpchar) AND (arrival_airport = 'OVB'::bpchar) AND
                     (scheduled_departure >= '2016-10-11 14:00:00+00'::timestamp with time zone) AND
                     (scheduled_departure < '2016-10-13 14:00:00+00'::timestamp with time zone))
                                Heap Blocks: exact=2
                                ->  Bitmap Index Scan on idx_flights_sched_dep (actual rows=2 loops=1)
                                      Index Cond:
                                   ((departure_airport = 'SVO'::bpchar) AND
                                   (arrival_airport = 'OVB'::bpchar) AND
                       (scheduled_departure >= '2016-10-11 14:00:00+00'::timestamp with time zone) AND
                       (scheduled_departure <= '2016-10-13 14:00:00+00'::timestamp with time zone))
              ->  Index Scan using bookings_0_pkey on bookings_0 b_1 (actual rows=1 loops=1)
                    Index Cond: (book_ref = bp_1.book_ref)
        ->  Index Scan using tickets_0_book_ref_idx on tickets_0 t_1 (actual rows=2 loops=1)
              Index Cond: (book_ref = b_1.book_ref)
  ->  Async Foreign Scan (actual rows=0 loops=1)
        Relations: (((boarding_passes_1_fdw bp_2) INNER JOIN (flights f)) INNER JOIN (tickets_1_fdw t_2)) INNER JOIN (bookings_1_fdw b_2)
        Network: FDW bytes sent=826 received=68
  ->  Async Foreign Scan (actual rows=0 loops=1)
        Relations: (((boarding_passes_2_fdw bp_3) INNER JOIN (flights f)) INNER JOIN (tickets_2_fdw t_3)) INNER JOIN (bookings_2_fdw b_3)
        Network: FDW bytes sent=829 received=34
  ->  Async Foreign Scan (actual rows=0 loops=1)
        Relations: (((boarding_passes_3_fdw bp_4) INNER JOIN (flights f)) INNER JOIN (tickets_3_fdw t_4)) INNER JOIN (bookings_3_fdw b_4)
        Network: FDW bytes sent=829

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

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


[15] В контексте компьютерных наук выражение «naïve approach» (дословно – наивный метод, наивный подход) очень похоже на «метод решения в лоб» и означает первую базовую идею, приходящую в голову при столкновении с проблемой, которая часто не учитывает сложности, крайние случаи или полный объём требований. С одной стороны, это грубый и прямой метод, направленный на получение работоспособного решения, с другой — такие решения легко понять и реализовать, но при этом системные ресурсы могут использоваться неэффективно.

[16] Возможно выделение значений из разных диапазонов, то есть значение может скакать. Например, в первом сегменте можно получить значение 5, во втором — 140003, в третьем — 70003 и т. д.

Пред. Наверх След.
17.2. Установка дополнительно поставляемых модулей Начало Глава 18. Подготовка к работе и сопровождение сервера