Код плагина находится на гитхабе — http://ift.tt/1HaKcOW. Приветсвуются pull-requests с улучшениями (особенно по части улучшения поддержки типов), багфиксами и просто косметическими улучшениями. JSON был выбран за простоту. Это не окончательный вариант, возможно после тестирования на реальных данных окажется что нужен более производительный формат, и придется переделать. В статье я не буду приводить весь код плагина, а только части про которые мне кажется нужно рассказать.
Необходимые требования для создания плагина: знание С, установленные средства сборки (gcc, cmake), установленные пакеты (в debian-системах) postgresql-9.4, postgresql-server-dev-9.4 и аналогичные в других системах. После установки postgresql, в postgresql.conf надо установить значение max_replication_slots = 1 (или больше) и wal_level = logical.
Сам плагин представляет собой подключаемую библиотеку на C, из которой вызываются callback функции на события postgresql. При инициализации вызывается функция _PG_output_plugin_init со структурой, полям которой полям нужно назначить свои функции:
- startup_cb — функция, вызываемая при инициализации плагина
- begin_cb — начало транзакции
- change_cb — запись данных
- commit_cb — потверждение транзакции
- shutdown_cb — деинициализация плагина
Функция которая заполняет структуру:
void _PG_output_plugin_init(OutputPluginCallbacks *cb)
{
cb->startup_cb = decoder_json_startup;
cb->begin_cb = decoder_json_begin_txn;
cb->change_cb = decoder_json_change;
cb->commit_cb = decoder_json_commit_txn;
cb->shutdown_cb = decoder_json_shutdown;
}
Теперь осталось определить эти пять функций.
decoder_json_startup вызывается в начале декодирования и используется для задания опций декодирования и создания своего контекста памяти:/* initialize this plugin */
static void
decoder_json_startup(LogicalDecodingContext *ctx,
OutputPluginOptions *opt,
bool is_init)
{
ListCell *option;
DecoderRawData *data;
data = palloc(sizeof(DecoderRawData));
data->context = AllocSetContextCreate(ctx->context,
"Raw decoder context",
ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
data->include_transaction = false;
data->sort_keys = false;
ctx->output_plugin_private = data;
/* Default output format */
opt->output_type = OUTPUT_PLUGIN_TEXTUAL_OUTPUT;
foreach(option, ctx->output_plugin_options)
{
DefElem *elem = lfirst(option);
Assert(elem->arg == NULL || IsA(elem->arg, String));
if (strcmp(elem->defname, "include_transaction") == 0)
{
/* if option does not provide a value, it means its value is true */
if (elem->arg == NULL)
data->include_transaction = true;
else if (!parse_bool(strVal(elem->arg), &data->include_transaction))
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("could not parse value \"%s\" for parameter \"%s\"",
strVal(elem->arg), elem->defname)));
}
else if (strcmp(elem->defname, "sort_keys") == 0) {
/* if option does not provide a value, it means its value is true */
if (elem->arg == NULL)
data->sort_keys = true;
else if (!parse_bool(strVal(elem->arg), &data->sort_keys))
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("could not parse value \"%s\" for parameter \"%s\"",
strVal(elem->arg), elem->defname)));
}
else
{
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("option \"%s\" = \"%s\" is unknown",
elem->defname,
elem->arg ? strVal(elem->arg) : "(null)")));
}
}
}
Здесь парсятся параметры, переданные в плагин и сохраняются в структуру. Созданный контекст памяти используется потом в
decoder_json_change чтобы корректно почистить используемые ресурсы. Важные моменты:- свои данные сохраняются в ctx->output_plugin_private
opt->output_type = OUTPUT_PLUGIN_TEXTUAL_OUTPUT— так задается что вывод плагина будет текстовый
decoder_json_shutdown вызывается в конце декодирования, и используется для чистки ресурсов./* cleanup this plugin's resources */
static void
decoder_json_shutdown(LogicalDecodingContext *ctx)
{
DecoderRawData *data = ctx->output_plugin_private;
/* cleanup our own resources via memory context reset */
MemoryContextDelete(data->context);
}
Дальше самое интересное. Надо определить функции
decoder_json_begin_txn, decoder_json_commit_txn и decoder_json_change которые собственно и генерируют строки, получаемые командами pg_logical_slot_peek_changes и pg_logical_slot_get_changes. Сгенеренную строку надо добавить в слот, это делается командами:OutputPluginPrepareWrite(ctx, true);
appendStringInfoString(ctx->out, "some string");
OutputPluginWrite(ctx, true);
Функции
decoder_json_begin_txn и decoder_json_commit_txn пишут (или просто пропускают, если есть такое условие) команды начала и конца транзакции в слот — строки 'begin' и 'commit' соответсвенно.Функция decoder_json_change вызывается на событие изменения данных. В этой функции определяется какое именно событие произошло (INSERT, UPDATE, DELETE) и для каждого из них создается своя структура. Для UPDATE и DELETE важно наличие уникального (not null) или первичного ключа в таблице, иначе просто невозможно определить изменяемую (удаляемую) строку. Это зависит от значения параметра REPLICA IDENTITY для таблицы.
Эта функция принимает 4 параметра:
- LogicalDecodingContext *ctx — контекст
- ReorderBufferTXN *txn
- Relation relation — сведения о изменяемой таблице
- ReorderBufferChange *change — сведения о данных
Кратко про функцию можно сказать что тип операции определяется через
change->action. Далее по данным в change (change->data.tp.newtuple и change->data.tp.oldtuple) создается JSON структура. JSON генерируется с помощью библиотеки libjansson.Вот здесь то и начинаются сложности. Если REPLICA IDENTITY для таблицы установлен в NOTHING или DEFAULT при отсутствующем первичном ключе, то невозможно определить изменяемые строки и в лог попадут только записи добавления. При обновлении или удалении данных с таблицы с DEFAULT, FULL, INDEX и при наличии уникального ключа, то его значение берется из newtuple или из oldtuple (если значение ключа изменяется запросом). При отсутствии уникального ключа и если FULL, то для идентификации используются все значения из oldtuple.
В результате строится JSON структура, вида {"a": 0, "r": "public.some_table", "c": {"id": 1}, "d": {"a": 2}}, где a — это тип действия, r — название таблицы, c — значения для идентификации строки, d — собственно данные.
Проверим работу. Сборка плагина и запуск тестов:
git clone http://ift.tt/1HaKcOZ
cd decoder_json
# разрешаем запись всем в папку с библиотеками постгреса - ни в коем случае нельзя делать на продакшене
sudo chmod a+rw `pg_config --pkglibdir`
chmod a+rwx ./
# скачиваем и собираем libjansson, для генерации JSON
make deps
# переключаемся под юзера postgres
sudo su postgres
make
make test
Тестирование плагина вручную:
# переключаемся под юзера postgres, создаем тестовую бд и открываем консоль работы с бд
sudo su postgres
createdb test_db
psql test_db
# psql консоль
test_db=# create table test1 (id serial primary key, name varchar);
test_db=# SELECT * FROM pg_create_logical_replication_slot('custom_slot', 'decoder_json');
slot_name | xlog_position
-------------+---------------
custom_slot | 0/4D9F870
(1 row)
Здесь мы указываем название слота и подключаемый плагин. В ответе мы видим название слота, и место (xlog позиция) с которого собственно начинается запись данных в слот. То что мы указали наш плагин, не означает что он уже работает, само декодирование начинается только когда мы забираем данные. Для этого используются функции
pg_logical_slot_peek_changes и pg_logical_slot_get_changes. Они отличаются тем что get функция после получения данных чистит очередь.Добавление данных:
test_db=# insert into test1 values (1, 'bb');
INSERT 0 1
test_db=# insert into test1 values (2, 'bb');
INSERT 0 1
test_db=# select * from pg_logical_slot_get_changes('custom_slot', NULL, NULL, 'include_transaction', 'on');
location | xid | data
-----------+-------+-----------------------------------------------------
0/BAB0968 | 48328 | begin
0/BAB0968 | 48328 | {"a":0,"r":"public.test1","d":{"id":1,"name":"bb"}}
0/BAB09F0 | 48328 | commit
0/BAB09F0 | 48329 | begin
0/BAB09F0 | 48329 | {"a":0,"r":"public.test1","d":{"id":2,"name":"bb"}}
0/BAB0A78 | 48329 | commit
(6 rows)
Изменение данных
test_db=# update test1 set name = 'dd' where id=2;
UPDATE 1
test_db=# select * from pg_logical_slot_get_changes('custom_slot', NULL, NULL, 'include_transaction', 'on');
location | xid | data
-----------+-------+------------------------------------------------------------------
0/BB4C700 | 48338 | begin
0/BB4C700 | 48338 | {"c":{"id":2},"a":1,"r":"public.test1","d":{"id":2,"name":"dd"}}
0/BB4C798 | 48338 | commit
(3 rows)
Удаление данных:
test_db=# delete from test1 where id=2;
DELETE 1
test_db=# select * from pg_logical_slot_get_changes('custom_slot', NULL, NULL, 'include_transaction', 'on');
location | xid | data
-----------+-------+-----------------------------------------
0/BB4C8A8 | 48339 | begin
0/BB4C8A8 | 48339 | {"c":{"id":2},"a":2,"r":"public.test1"}
0/BB4C9C8 | 48339 | commit
(3 rows)
Использованные и полезные ресурсы:
- документация PostgresqlSQL — http://ift.tt/15Yxq9x
- пример плагина из PostgreSQL (http://ift.tt/1a2drsT)
- michael.otacoo.com/ — очень полезный блог, плагин decoder_raw автора этого блога использовался как основа для моего плагина.
- http://ift.tt/1HaKfdy — плагин который использует google protocol buffers как выходной формат.
This entry passed through the Full-Text RSS service - if this is your content and you're reading it on someone else's site, please read the FAQ at http://ift.tt/jcXqJW.
Комментариев нет:
Отправить комментарий