148. Базы данных из C: SQLite, PostgreSQL (libpq), Redis (hiredis)

Прибор, собирающий данные, должен их где-то хранить, и файл со строками быстро перестаёт справляться: нужен поиск, отбор по датам, среднее за час, надёжность при выключении питания. Это задача базы данных. Три из них, каждая со своим местом: SQLite — файл на диске прибора без сервера; PostgreSQL — сервер для многих клиентов; Redis — быстрая память для ключей и очередей. Все три опрошены из C в stm-sys:1 (_verify/sys148/): SQLite прямо в контейнере, PostgreSQL и Redis — в соседних контейнерах одной сети.

Все три написаны на C, как и большинство СУБД вообще, — что само по себе говорит о языке.

Как это называется в документации

Оригинал	Перевод
DBMS	СУБД
table, row, column	таблица, строка, столбец
primary key	первичный ключ
transaction, commit, rollback	транзакция, подтверждение, откат
ACID (atomicity, consistency, isolation, durability)	атомарность, согласованность, изоляция, долговечность
prepared statement, bind parameter	подготовленный запрос, параметр
SQL injection	SQL-инъекция
OLTP / OLAP	обработка транзакций / аналитика
normalization	нормализация: без повторяющихся данных
key-value store	хранилище «ключ — значение»
TTL (time to live)	срок жизни ключа
embedded database	встраиваемая база: библиотека, а не сервер

148.1 Что такое база и зачем транзакции

Реляционная база — таблицы со строками и столбцами, связанные ключами; язык запросов — SQL. Обещание базы — ACID: операция либо выполняется целиком, либо не выполняется (атомарность); данные всегда в допустимом состоянии (согласованность); параллельные клиенты не видят чужих недоделанных изменений (изоляция); что подтверждено — не пропадёт даже при отключении питания (долговечность). Последнее — то, за что база берёт время: COMMIT ждёт, пока данные лягут на диск.

OLTP — много коротких транзакций (запись измерений, заказы); OLAP — редкие тяжёлые запросы по всему массиву (отчёты). Для прибора почти всегда первое. Нормализация — правило «каждый факт хранится один раз»: имя датчика — в таблице датчиков, измерение ссылается на него номером, а не повторяет имя в каждой строке.

148.2 SQLite: база в одном файле

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

    if (sqlite3_open("/tmp/meas.db", &db) != SQLITE_OK) { fprintf(stderr, "%s\n", sqlite3_errmsg(db)); return 1; }

Три вызова на всё: sqlite3_exec для команд без результата, sqlite3_prepare_v2 + sqlite3_bind_* + sqlite3_step для запросов с параметрами и строками ответа, sqlite3_finalize в конце:

    sqlite3_prepare_v2(db, "INSERT INTO meas(sensor, value) VALUES(?1, ?2)", -1, &st, NULL);   /* разбор один раз */

        sqlite3_bind_text(st, 1, i % 2 ? "temp" : "hum", -1, SQLITE_STATIC);   /* параметры, не склейка строк */
        sqlite3_bind_double(st, 2, 20 + i % 10);
        sqlite3_step(st);
        sqlite3_reset(st);

Запрос готовится один раз, дальше только подставляются параметры и вызывается step/reset. Результат SELECT читают в цикле while (sqlite3_step(st) == SQLITE_ROW) через sqlite3_column_*:

SQLite 3.40.1
hum   n=5500 avg=24.00 max=28
temp  n=5500 avg=25.00 max=29
ошибка SQL: no such column: nope
файл базы: 380928 байт

Ошибка приходит текстом через sqlite3_errmsg или char *err в sqlite3_exec — её освобождают sqlite3_free. Файл с 11 000 строк — 380 КБ; его же открывает утилита sqlite3 из командной строки: sqlite3 /tmp/meas.db "SELECT COUNT(*) FROM meas" → 11000.

148.3 Цена транзакции

Одна и та же вставка двумя способами:

1000 вставок по одной (автокоммит): 4386 мс
10000 вставок в одной транзакции: 9 мс

Без явной транзакции каждая вставка — своя транзакция с записью на диск и ожиданием: 4,4 мс на строку. В одной транзакции BEGIN … COMMIT десять тысяч строк легли за 9 мс — в тысячи раз быстрее на строку. Правило для прибора: копить измерения и писать пачкой раз в секунду, а не по одному. Подтверждённое при этом не теряется: при пропадании питания пропадает только пачка, которая ещё не дошла до COMMIT, — до секунды измерений.

Второй рычаг — режим журнала. PRAGMA journal_mode=WAL дописывает изменения в конец отдельного журнала и переносит их в базу позже: чтение не ждёт записи, а запись — чтения. С PRAGMA synchronous=NORMAL журнал ещё и не ждёт носителя на каждой транзакции. Те же 1000 автокоммитов в том же контейнере: в WAL — около 2,5 с, почти вдвое быстрее, в WAL с synchronous=NORMAL — около 15 мс. Цена второго: последние транзакции перед пропаданием питания могут откатиться, но сама база остаётся целой.

148.4 Параметры против инъекции

Собирать SQL склейкой строк — "... WHERE sensor = '" + имя + "'" — нельзя: имя x' OR '1'='1 превратит запрос в «все строки», а '; DROP TABLE — в удаление. Параметр ?1 передаётся отдельно от текста запроса и в SQL никогда не разбирается:

    sqlite3_bind_text(st, 1, "x' OR '1'='1", -1, SQLITE_STATIC);

WHERE sensor = «x' OR '1'='1» через параметр: 0 строк (инъекция не прошла)

То же правило в libpq ($1, PQexecParams) и во всех остальных библиотеках. Оно же экономит разбор: запрос с параметрами готовится один раз.

148.5 PostgreSQL: сервер и libpq

Когда клиентов много или база на другой машине, нужен сервер. PostgreSQL — один из самых распространённых открытых серверов. Клиентская библиотека на C — libpq; на ней стоит, например, psycopg2 для Python, а драйверы Go и Java реализуют протокол сами.

    PGconn *c = PQconnectdb(conn);
    if (PQstatus(c) != CONNECTION_OK) die(c, "connect");

Строка подключения — host=… user=… password=… dbname=…. Команды — PQexec, запросы с параметрами — PQexecParams с $1, $2; результат — PGresult, который обязательно освобождают PQclear:

        r = PQexecParams(c, "INSERT INTO meas(sensor, value) VALUES($1, $2::real)", 2, NULL, vals[i], NULL, NULL, 0);   /* $1, $2 — параметры */

libpq 150019, сервер 160014
строк 2, столбцов 3
  1 temp 21.5
  2 hum 40
ошибка: ERROR:  column "nope" does not exist

Версии — числами: старшая × 10 000 + младшая, то есть библиотека 15.19, сервер 16.14. Строка bad, вставленная между BEGIN и ROLLBACK, в выборке отсутствует — откат сработал. PQntuples/PQnfields — размер результата, PQgetvalue — ячейка строкой (для двоичного формата есть отдельный режим). Сервер для опыта — контейнер postgres:16-alpine, клиент — в соседнем контейнере той же сети Docker: host=stm148-pg.

148.6 Redis: ключи в памяти

Redis хранит всё в оперативной памяти и отвечает за микросекунды: счётчики, кэш, последние значения датчиков, очереди сообщений. Библиотека hiredis — один вызов redisCommand с форматом как у printf, ответ — redisReply с типом:

    redisReply *r = redisCommand(c, "SET sensor:temp %s", "21.5");
    printf("SET: %s\n", r->str); freeReplyObject(r);

SET: OK
GET: 21.5
INCR ×3: 3
LRANGE: 3 элементов: a b c
TTL token: 10 с
GET nope: (nil)
redis_version:7.4.10

INCR — атомарный счётчик, RPUSH/LRANGE — список (очередь), SET … EX 10 — ключ исчезнет через 10 секунд, GET несуществующего — REDIS_REPLY_NIL. Ключи принято именовать через двоеточие: sensor:temp. Долговечность у Redis по умолчанию слабее — снимки на диск раз в минуты; для измерений, которые нельзя терять, он не замена SQLite, а кэш перед ней.

148.7 Что выбрать

На микроконтроллере без ОС базы нет; ближайшее — журнал в Flash (раздел 11) и FatFS (раздел 42). SQLite на большинстве микроконтроллеров не помещается: даже урезанная сборка — сотни килобайт кода (полная библиотека на компьютере — 1,4 МБ), плюс ОЗУ под кэш страниц. На старших камнях с внешней памятью её запускают поверх FatFS через свой слой файлов (VFS).

148.8 Грабли

148.9 Итоги

• ACID; OLTP для прибора; факт хранится один раз.
• SQLite: open, prepare/bind/step/finalize; 11 000 строк — 380 КБ.
• Транзакция: 1000 автокоммитов — 4386 мс, 10 000 строк в одной — 9 мс.
• WAL: чтение не ждёт записи; с synchronous=NORMAL 1000 автокоммитов — около 15 мс, но последние транзакции при пропадании питания могут откатиться.
• Параметры, не склейка: инъекция x' OR '1'='1 дала 0 строк.
• libpq: PQconnectdb, PQexecParams с $1, PQclear; ROLLBACK откатил.
• hiredis: redisCommand с форматом, INCR, списки, EX; кэш, не хранилище.

Полные исходники

sqlite_demo.c — SQLite из C: подготовленные запросы, вставка в транзакции, агрегат, ошибка SQL.

/* Раздел 148: SQLite из C — таблица измерений, подготовленные запросы с параметрами,
   вставка 10 000 строк по одной и в транзакции, выборка с агрегатом, ошибка SQL. */
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <sqlite3.h>

static double now(void) { struct timespec t; clock_gettime(CLOCK_MONOTONIC, &t); return t.tv_sec + t.tv_nsec / 1e9; }

static void insert_n(sqlite3 *db, int n, int tx)
{
    sqlite3_stmt *st;
    sqlite3_prepare_v2(db, "INSERT INTO meas(sensor, value) VALUES(?1, ?2)", -1, &st, NULL);   /* разбор один раз */
    double t = now();
    if (tx) sqlite3_exec(db, "BEGIN", NULL, NULL, NULL);
    for (int i = 0; i < n; i++) {
        sqlite3_bind_text(st, 1, i % 2 ? "temp" : "hum", -1, SQLITE_STATIC);   /* параметры, не склейка строк */
        sqlite3_bind_double(st, 2, 20 + i % 10);
        sqlite3_step(st);
        sqlite3_reset(st);
    }
    if (tx) sqlite3_exec(db, "COMMIT", NULL, NULL, NULL);
    sqlite3_finalize(st);
    printf("%d вставок %s: %.0f мс\n", n, tx ? "в одной транзакции" : "по одной (автокоммит)", (now() - t) * 1e3);
}

int main(void)
{
    sqlite3 *db;
    remove("/tmp/meas.db");
    if (sqlite3_open("/tmp/meas.db", &db) != SQLITE_OK) { fprintf(stderr, "%s\n", sqlite3_errmsg(db)); return 1; }
    printf("SQLite %s\n", sqlite3_libversion());
    char *err = NULL;
    sqlite3_exec(db, "CREATE TABLE meas(id INTEGER PRIMARY KEY, sensor TEXT NOT NULL, value REAL, "
                     "ts DATETIME DEFAULT CURRENT_TIMESTAMP)", NULL, NULL, &err);
    insert_n(db, 1000, 0);
    insert_n(db, 10000, 1);
    sqlite3_stmt *st;
    sqlite3_prepare_v2(db, "SELECT sensor, COUNT(*), AVG(value), MAX(value) FROM meas GROUP BY sensor ORDER BY sensor", -1, &st, NULL);
    while (sqlite3_step(st) == SQLITE_ROW)
        printf("%-5s n=%d avg=%.2f max=%.0f\n", sqlite3_column_text(st, 0), sqlite3_column_int(st, 1),
               sqlite3_column_double(st, 2), sqlite3_column_double(st, 3));
    sqlite3_finalize(st);
    if (sqlite3_exec(db, "SELECT nope FROM meas", NULL, NULL, &err) != SQLITE_OK) { printf("ошибка SQL: %s\n", err); sqlite3_free(err); }
    /* подстановка: параметр с кавычкой не ломает запрос */
    sqlite3_prepare_v2(db, "SELECT COUNT(*) FROM meas WHERE sensor = ?1", -1, &st, NULL);
    sqlite3_bind_text(st, 1, "x' OR '1'='1", -1, SQLITE_STATIC);
    sqlite3_step(st);
    printf("WHERE sensor = «x' OR '1'='1» через параметр: %d строк (инъекция не прошла)\n", sqlite3_column_int(st, 0));
    sqlite3_finalize(st);
    sqlite3_close(db);
    return 0;
}

pg_demo.c — PostgreSQL через libpq: параметризованный INSERT, выборка, откат транзакции.

/* Раздел 148: PostgreSQL через libpq — подключение, таблица, параметризованный INSERT,
   выборка, транзакция с откатом. Адрес сервера — в переменной PG_CONN. */
#include <stdio.h>
#include <stdlib.h>
#include <libpq-fe.h>

static void die(PGconn *c, const char *what) { fprintf(stderr, "%s: %s", what, PQerrorMessage(c)); PQfinish(c); exit(1); }

int main(void)
{
    const char *conn = getenv("PG_CONN") ? getenv("PG_CONN") : "host=stm148-pg user=postgres password=pw dbname=postgres";
    PGconn *c = PQconnectdb(conn);
    if (PQstatus(c) != CONNECTION_OK) die(c, "connect");
    printf("libpq %d, сервер %d\n", PQlibVersion(), PQserverVersion(c));
    PGresult *r = PQexec(c, "DROP TABLE IF EXISTS meas; CREATE TABLE meas(id SERIAL PRIMARY KEY, sensor TEXT NOT NULL, value REAL)");
    if (PQresultStatus(r) != PGRES_COMMAND_OK) die(c, "create");
    PQclear(r);
    const char *vals[2][2] = { { "temp", "21.5" }, { "hum", "40" } };
    for (int i = 0; i < 2; i++) {
        r = PQexecParams(c, "INSERT INTO meas(sensor, value) VALUES($1, $2::real)", 2, NULL, vals[i], NULL, NULL, 0);   /* $1, $2 — параметры */
        if (PQresultStatus(r) != PGRES_COMMAND_OK) die(c, "insert");
        PQclear(r);
    }
    r = PQexec(c, "BEGIN"); PQclear(r);
    r = PQexec(c, "INSERT INTO meas(sensor, value) VALUES('bad', 0)"); PQclear(r);
    r = PQexec(c, "ROLLBACK"); PQclear(r);                                        /* откат: строки bad не будет */
    r = PQexec(c, "SELECT id, sensor, value FROM meas ORDER BY id");
    if (PQresultStatus(r) != PGRES_TUPLES_OK) die(c, "select");
    printf("строк %d, столбцов %d\n", PQntuples(r), PQnfields(r));
    for (int i = 0; i < PQntuples(r); i++) printf("  %s %s %s\n", PQgetvalue(r, i, 0), PQgetvalue(r, i, 1), PQgetvalue(r, i, 2));
    PQclear(r);
    r = PQexec(c, "SELECT nope FROM meas");
    printf("ошибка: %s", PQresultErrorMessage(r));
    PQclear(r);
    PQfinish(c);
    return 0;
}

Дальше: раздел 149 — графика и игры: SDL2, OpenGL, игровой цикл.