Скачать

Выпуск Picopyn 2.0.0

· Александр Толстой

Команда Picodata выпустила Picopyn 2.0.0 — Python-драйвер для Picodata с пулом соединений, автоматическим обнаружением узлов кластера и маршрутизацией запросов, учитывающей шардирование данных. Новая версия требует Picodata 26.1.1 или новее. Как и ранее, пакет можно установить из PyPI или исходного кода. Подробный changelog можно изучить в документации драйвера, в текущей же статье рассмотрим основные изменения с примерами.

Асинхронный API

Отслеживание топологии кластера

Главное изменение релиза: сбор топологии в асинхронном Picopyn. Драйвер теперь в фоновом режиме собирает топологию кластера (тиры, репликасеты, инстансы) и карту бакетов. Эти данные используются для маршрутизации запросов, а в режиме discovery — ещё и для подстраивания состава пула соединений.

from picopyn.asynchronous import Pool

# в режиме discovery пул по умолчанию собирает топологию с дефолтным интервалом
pool_with_topology = Pool(
    dsn="postgresql://admin:pass@host1:5432,host2:5432",
    enable_discovery=True,
    max_size=10,
)
# чтобы отключить сбор топологии, нужно явно установить `topology_update_interval=None`
discovered_pool_without_topology = Pool(
    dsn="postgresql://admin:pass@host1:5432,host2:5432",
    enable_discovery=True,
    max_size=10,
    topology_update_interval=None,
)
# интервал сбора топологии задаётся в секундах
pool_with_slow_topology_updates = Pool(
    dsn="postgresql://admin:pass@host1:5432,host2:5432",
    enable_discovery=True,
    max_size=10,
    topology_update_interval=600,
)

# в режиме bootstrap пул собирает топологию по тем же правилам, но
# не использует её для изменения состава пула
bootstrapped_pool_with_topology = Pool(
    dsn="postgresql://admin:pass@host1:5432,host2:5432",
    enable_discovery=False,
    max_size=10,
)

Маршрутизация запросов

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

Сейчас маршрутизация работает только для запросов, которые соответствуют каждому из этих условий:

  • запрос параметризован;
  • запрос выполняет операцию INSERT;
  • вставляется одна строка данных.

Автоматическая маршрутизация

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

Пример:

import asyncio

from picopyn.asynchronous import Pool


async def main():
    pool = Pool(
        dsn="postgresql://admin:pass@host1:5432,host2:5432",
        enable_discovery=True,
        max_size=10,
    )
    await pool.connect()

    # ключ шардирования таблицы — по нему и считается бакет
    await pool.execute("""
        CREATE TABLE "warehouse" (
            id INTEGER NOT NULL,
            item TEXT NOT NULL,
            PRIMARY KEY (id)
        ) USING memtx DISTRIBUTED BY (id) OPTION (TIMEOUT = 3.0);
    """)

    query = 'INSERT INTO "warehouse" VALUES ($1::int, $2::text)'

    # первый вызов уходит на соединение, выбранное балансировкой,
    # в фоновом режиме драйвер запрашивает метаданные у Picodata, тем самым
    # не блокируя и не задерживая первое выполнение запроса
    await pool.execute(query, 1, "first")

    # повторные вызовы того же запроса драйвер отправляет сразу
    # на мастер репликасета, которому принадлежит этот бакет,
    # если в пуле есть соединение с этим мастером
    await pool.execute(query, 2, "second")

    await pool.close()


asyncio.run(main())

Ручная маршрутизация

Для использования маршрутизации напрямую:

from picopyn.bucket import calculate_bucket_id

query = 'INSERT INTO "warehouse" VALUES ($1::int, $2::text)'
params = (1, "test")

# метаданные распределения запроса: тир и параметры ключа шардирования
meta = await pool.get_query_metadata(query)

bucket_count = pool.topology.tiers[meta.tier]
bucket_id = calculate_bucket_id(meta.dk_meta, params, bucket_count)

async with pool.acquire_by_tier_and_bucket_id(meta.tier, bucket_id) as conn:
    await conn.execute(query, *params)

Приведение состава пула в соответствие с топологией

В режиме discovery при включённом сборе топологии состав пула теперь согласуется с топологией: при изменениях пул сразу перестраивает состав подключений. При выключенном discovery (bootstrap-режиме) топология используется только для наблюдения и не влияет на состав пула.

Пример перестройки состава пула после изменения топологии:

pool-reconcile

Корректное завершение (graceful shutdown) работы пула

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

await pool.close(timeout=10)

Синхронный API

Закрытие соединений через менеджер контекста

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

from picopyn.synchronous import connect

with connect("postgresql://admin:pass@localhost:5432") as conn:
    cur = conn.cursor()
    cur.execute('SELECT * FROM "warehouse"')
    print(cur.fetchall())
# здесь соединение уже закрыто

У пула для этого есть Pool.connection: на выходе из блока соединение не закрывается, а возвращается в пул. Синхронный пул намеренно не предоставляет методы acquire/release: забрать соединение можно только через менеджер контекста, поэтому его нельзя случайно забыть вернуть в пул.

from picopyn.synchronous import Pool

pool = Pool(
    dsn="postgresql://admin:pass@host1:5432,host2:5432",
    enable_discovery=True,
    max_size=10,
)
pool.open()

with pool.connection() as conn:
    cur = conn.cursor()
    cur.execute('INSERT INTO "warehouse" VALUES (%s, %s)', (1, "test"))
# соединение вернулось в пул

pool.close()

Свойство autocommit

Picodata пока не поддерживает управление транзакциями (COMMIT/ROLLBACK), поэтому соединение всегда работает в режиме autocommit. Согласно спецификации DB-API 2.0, свойство autocommit теперь доступно. Его значение всегда равно True, а попытка его изменить возвращает ошибку.

from picopyn.synchronous import connect

with connect("postgresql://admin:pass@localhost:5432") as conn:
    assert conn.autocommit is True

    conn.autocommit = False  # ValueError: autocommit=False is not supported

Планы на ближайшие релизы

  • Упрощение API для прямого выполнения запросов к шардированным таблицам
  • Поддержка приведения состава пула в соответствие с топологией при bootstrap-режиме (enable_discovery=False)
  • Оптимизация балансировки соединений между инстансами Picodata
  • Внедрение работы с топологией в синхронную часть драйвера