Расширения PostgreSQL: работа с таблицами

23 мая 2022

Наиболее простой способ работы с таблицами из расширений PostgreSQL заключается в использовании Server Programming Interface (SPI). С этим интерфейсом мы познакомились в рамках статьи Учимся писать расширения на языке C для PostgreSQL. Однако SPI имеет накладные расходы на парсинг и планирование запросов. Поэтому в простых сценариях выгоднее работать с таблицами напрямую. Звучит страшновато, но на самом деле это не так сложно.

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

Замечательную статью по этой теме в свое время написал Юрий Журавлев. Но за прошедшие ~7 лет код PostgreSQL сильно изменился. Если использовать примеры из статьи как есть, то они либо не скомпилируются, либо упадут с ошибкой. Интересно, быстро ли потеряет актуальность эта статья, и кто напишет ее обновленную версию?

Для примера рассмотрим расширение для работы с телефонной книгой:

CREATE TABLE phonebook(
  "id" SERIAL PRIMARY KEY NOT NULL,
  "name" NAME NOT NULL,
  "phone" INT NOT NULL);

CREATE INDEX phonebook_name_idx ON phonebook USING btree("name");

Тип NAME — это строка длиной до 63-х символов. PostgreSQL активно использует данный тип в своем каталоге, см описание таблиц pg_proc, pg_enum, и прочих. Для хранения коротких строк в своем расширении вы наверняка воспользуетесь именно NAME, а не TEXT. Далее станет понятно, почему.

В коде расширения схема БД должна быть описана как-то так:

#define PHONEBOOK_TABLE_NAME "phonebook"
#define PHONEBOOK_PKEY_SEQ_NAME "phonebook_id_seq"
#define PHONEBOOK_NAME_IDX_NAME "phonebook_name_idx"

typedef struct FormData_phonebook
{
    int32 id;
    NameData name;
    int32 phone;
} FormData_phonebook;

typedef FormData_phonebook* Form_phonebook;

typedef enum Anum_phonebook
{
    Anum_phonebook_id = 1,
    Anum_phonebook_name,
    Anum_phonebook_phone,
    _Anum_phonebook_max,
} Anum_phonebook;

#define Natts_phonebook (_Anum_phonebook_max - 1)

typedef enum Anum_phonebook_name_idx
{
    Anum_phonebook_name_idx_name = 1,
    _Anum_phonebook_name_idx_max,
} Anum_phonebook_name_idx;

#define Natts_phonebook_name_idx (_Anum_phonebook_name_idx_max - 1)

Думаю, что здесь особые пояснения не требуются. Имена структур и enum’ов соответствуют соглашениям, принятым в коде PostgreSQL.

Для работы с таблицей нам понадобится ее object identifier, или Oid. Oid’ы представляют собой первичные ключи в каталоге PostgreSQL. Каждая таблица, хранимая процедура, тип, и так далее — все имеет уникальный Oid. Существует ряд типов-алиасов к Oid. Например, для Oid’ов таблиц есть тип regclass. Внутри это точно такой же Oid, но пользователю regclass отображается, как имя соответствующей таблицы.

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

static Oid
name_to_oid(const char* name)
{
    return DatumGetObjectId(
        DirectFunctionCall1(to_regclass, CStringGetTextDatum(name))
    );
}

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

Datum
phonebook_insert(PG_FUNCTION_ARGS)
{
    Relation rel;
    HeapTuple tup;
    Datum values[Natts_phonebook];
    bool nulls[Natts_phonebook];
    Name name = PG_GETARG_NAME(0);
    int32 phone = PG_GETARG_INT32(1);
    Oid tbl_oid = name_to_oid(PHONEBOOK_TABLE_NAME);
    Oid pkey_seq_oid = name_to_oid(PHONEBOOK_PKEY_SEQ_NAME);
    int32 next_id = (int32)DatumGetInt64(
        DirectFunctionCall1(nextval_oid, ObjectIdGetDatum(pkey_seq_oid))
    );

    memset(nulls, false, sizeof(nulls));

    rel = table_open(tbl_oid, RowExclusiveLock);

    /* or: AttrNumberGetAttrOffset(Anum_phonebook_id) */
    values[Anum_phonebook_id - 1] = Int32GetDatum(next_id);
    values[Anum_phonebook_name - 1] = NameGetDatum(name);
    values[Anum_phonebook_phone - 1] = Int32GetDatum(phone);

    tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);

    /* inserts a new heap tuple, keeping indexes current */
    CatalogTupleInsert(rel, tup);
    heap_freetuple(tup);

    table_close(rel, RowExclusiveLock);
    PG_RETURN_INT32(next_id);
}

Следующий id мы здесь получаем точно так же, как это делает PostgreSQL — вызываем nextval(), передав ему Oid соответствующего sequence. Само собой разумеется, я не помню такие вещи наизусть. Все это прямым текстом написано в определении таблицы. Достаточно сказать \d phonebook в psql, и все сразу понятно.

Остальной код достаточно очевиден — открыть таблицу с правильной блокировкой, создать кортеж, записать кортеж в таблицу, освободить кортеж, закрыть таблицу. Всю сложную работу за нас тут делает CatalogTupleInsert(). Процедура не просто записывает кортеж, но и должным образом обновляет все индексы. Название как бы намекает, что PostgreSQL использует процедуру для работы с собственным каталогом.

Как проверить, что индексы действительно обновляются? PostgreSQL напрямую не позволяет читать содержимое индексов. Зато в комплекте c PostgreSQL идет замечательное расширение pageinspect. Помимо прочего, в расширении есть функция bt_page_items(), которая как раз позволяет читать из индексов. В качестве домашнего задания предлагаю вам убедиться, что приведенный код действительно не портит индексы таблицы.

Еще есть CatalogTupleUpdate() и CatalogTupleDelete() для обновления и удаления кортежей соответственно. Они не сложнее CatalogTupleInsert(), поэтому не будем подробно на них останавливаться. Заинтересованные читатели найдут примеры использования данных процедур в полной версии исходников к посту.

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

Datum
phonebook_lookup_seqscan(PG_FUNCTION_ARGS)
{
    Relation rel;
    HeapTuple tup;
    TableScanDesc scan;
    int32 found_phone = -1;
    Name name = PG_GETARG_NAME(0);
    Oid tbl_oid = name_to_oid(PHONEBOOK_TABLE_NAME);

    rel = table_open(tbl_oid, AccessShareLock);
    scan = table_beginscan(rel, GetTransactionSnapshot(), 0, NULL);

    while ((tup = heap_getnext(scan, ForwardScanDirection)) != NULL)
    {
        Form_phonebook record = (Form_phonebook) GETSTRUCT(tup);

        if(strcmp(record->name.data, name->data) == 0)
        {
            found_phone = record->phone;
            break;
        }
    }

    table_endscan(scan);
    table_close(rel, AccessShareLock);
    PG_RETURN_INT32(found_phone);
}

Особого внимания заслуживает макрос GETSTRUCT(). Он позволяет из кортежа получить указатель на ранее объявленную структуру FormData_phonebook. Такой код часто используется внутри PostgreSQL. Но чтобы это работало, должно выполняться несколько условий. Таблица не должна использовать типы переменного размера, такие, как TEXT. Все столбцы должны быть объявлены, как NOT NULL. Плюс к этому, следует соблюдать особую осторожность при изменении схемы БД. Последний вопрос выходит за рамки статьи.

Если вы не согласны на названные ограничения, вместо GETSTRUCT() следует воспользоваться heap_deform_tuple():

    /* ... */
    while ((tup = heap_getnext(scan, ForwardScanDirection)) != NULL)
    {
        Name rec_name;
        Datum values[Natts_phonebook];
        bool isnull[Natts_phonebook];

        heap_deform_tuple(tup, RelationGetDescr(rel), values, isnull);
        rec_name = DatumGetName(values[Anum_phonebook_name - 1]);

        if(strcmp(rec_name->data, name->data) == 0)
        {
            found_phone = DatumGetInt32(
                values[Anum_phonebook_phone - 1]
            );
            break;
        }
    }
    /* ... */

Это более дорогой способ, зато он работает как с TEXT, так и с NULL‘ами. Оправдано его использование или нет, определяется частотой обращений к таблице, и в целом решаемой задачей.

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

Datum
phonebook_lookup_index(PG_FUNCTION_ARGS)
{
    Relation rel, idxrel;
    IndexScanDesc scan;
    TupleTableSlot* slot;
    HeapTuple tup;
    ScanKeyData skey[1];
    int32 found_phone = -1;
    Name name = PG_GETARG_NAME(0);
    Oid tbl_oid = name_to_oid(PHONEBOOK_TABLE_NAME);
    Oid idx_oid = name_to_oid(PHONEBOOK_NAME_IDX_NAME);

    rel = table_open(tbl_oid, AccessShareLock);
    idxrel = index_open(idx_oid, AccessShareLock);
   
    scan = index_beginscan(rel, idxrel, GetTransactionSnapshot(),
                           1 /* nkeys */, 0 /* norderbys */);

    ScanKeyInit(&skey[0],
                /* numeration starts from 1; Note: idx, not rel! */
                Anum_phonebook_name_idx_name,
                BTEqualStrategyNumber, F_NAMEEQ,
                NameGetDatum(name));
    index_rescan(scan, skey, 1,
                 NULL /* orderbys */, 0 /* norderbys */);

    /* wrapper for MakeSingleTupleTableSlot() */
    slot = table_slot_create(rel, NULL);
    while (index_getnext_slot(scan, ForwardScanDirection, slot))
    {
        Form_phonebook record;
        bool should_free;

        tup = ExecFetchSlotHeapTuple(slot, false, &should_free);
        record = (Form_phonebook) GETSTRUCT(tup);

        if(strcmp(record->name.data, name->data) == 0)
        {
            found_phone = record->phone;
            if(should_free) heap_freetuple(tup);
            break;
        }

        if(should_free) heap_freetuple(tup);
    }

    index_endscan(scan);
    ExecDropSingleTupleTableSlot(slot);
    table_close(idxrel, AccessShareLock);
    table_close(rel, AccessShareLock);
    PG_RETURN_INT32(found_phone);
}

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

А у меня на этом все. Полную версию исходников к посту вы найдете на GitHub.

Дополнение: Внутренности PostgreSQL: страницы и кортежи

Метки: , , .