PHP code example of rasuvaeff / clickhouse-toolkit

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

rasuvaeff / clickhouse-toolkit example snippets

$qb = new ClickHouseQueryBuilder(allowedFields: ['id', 'status'], fieldTypes: ['id' => T::UInt64]);
$where = $qb->buildWhere(new Equals('status', 'active'));
$sql = $qb->buildSelect(table: 'events', where: $where->sql, limit: 20);

use Rasuvaeff\ClickHouseToolkit\ClickHouseClientFactory;
use Rasuvaeff\ClickHouseToolkit\ClickHouseConfig;
use Rasuvaeff\ClickHouseToolkit\ClickHouseDataType as T;
use Rasuvaeff\ClickHouseToolkit\ClickHouseQueryBuilder;
use SimPod\ClickHouseClient\Format\JsonEachRow;
use Yiisoft\Data\Reader\Filter\In;
use Yiisoft\Data\Reader\Sort;

// 1. Build a client.
$client = (new ClickHouseClientFactory(new ClickHouseConfig(
    host: 'clickhouse',
    port: 8123,
    database: 'app',
    username: 'default',
    password: '',
)))->create();

// 2. Build a safe, parameterized query from user-supplied filters.
$qb = new ClickHouseQueryBuilder(
    allowedFields: ['id', 'status', 'created_at'],
    fieldTypes: ['id' => T::UInt64, 'created_at' => T::DateTime],
    defaultSort: 'id DESC',
);

$where = $qb->buildWhere(new In('status', ['active', 'pending']));
$orderBy = $qb->buildOrderBy(Sort::only(['created_at'])->withOrder(['created_at' => 'desc']));
$sql = $qb->buildSelect(table: 'events', columns: ['id', 'status'], where: $where->sql, orderBy: $orderBy, limit: 20);

// 3. Execute.
$output = $where->isEmpty()
    ? $client->select($sql, new JsonEachRow())
    : $client->selectWithParams($sql, $where->params, new JsonEachRow());

foreach ($output->data as $row) {
    // ...
}

final readonly class ClickHouseConfig
{
    public function __construct(
        public string $host = '127.0.0.1',
        public int $port = 8123,
        public string $database = 'default',
        public string $username = 'default',
        public string $password = '',
        public bool $secure = false,   // true -> https://
    ) {}

    public function baseUri(): string; // e.g. "http://127.0.0.1:8123"
}

use Rasuvaeff\ClickHouseToolkit\ClickHouseClientFactory;
use Rasuvaeff\ClickHouseToolkit\ClickHouseConfig;

// Auto-discovers an installed PSR-18 client + PSR-17 factories:
$client = (new ClickHouseClientFactory(new ClickHouseConfig(
    host: 'ch.internal',
    secure: true,     // https
)))->create();

$client->executeQuery('SELECT 1');

use GuzzleHttp\Client;

$factory = new ClickHouseClientFactory(
    config: new ClickHouseConfig(host: 'ch.internal', secure: true),
    httpClient: new Client(['timeout' => 10.0]),
    // requestFactory / streamFactory / uriFactory are optional (auto-discovered when null)
);

public function __construct(
    private array $allowedFields,            // list<string>
    private array $fieldTypes = [],          // field => ClickHouse type, default "String" (use ClickHouseDataType constants)
    private string $defaultSort = '', // no ORDER BY by default; pass e.g. 'id DESC' for stable pagination
    private ?FilterInterface $mandatoryFilter = null,
    private ?string $serverTimezone = null,  // IANA timezone; DateTime values are converted before formatting
) {}

$qb = ClickHouseQueryBuilder::create(['id', 'status'], ['id' => T::UInt64])
    ->withMandatoryFilter(new Equals('tenant_id', $tenantId));

$where = $qb->buildWhere($userFilter); // (tenant_id = {p0:...}) AND (<user filter>)

use Rasuvaeff\ClickHouseToolkit\ClickHouseRawFilter;

$where = $qb->buildWhere(new ClickHouseRawFilter('toDate(created_at) = {d:Date}', ['d' => '2024-01-01']));

use Yiisoft\Data\Reader\Filter\AndX;
use Yiisoft\Data\Reader\Filter\Equals;
use Yiisoft\Data\Reader\Filter\GreaterThanOrEqual;

$where = $qb->buildWhere(new AndX(
    new Equals('status', 'active'),
    new GreaterThanOrEqual('user_id', 1000),
));

$selectSql = $qb->buildSelect(table: 'events', columns: ['id', 'status'], where: $where->sql, limit: 50);
$countSql  = $qb->buildCount(table: 'events', where: $where->sql);

$rows  = $client->selectWithParams($selectSql, $where->params, new JsonEachRow())->data;
$total = (int) ($client->selectWithParams($countSql, $where->params, new JsonEachRow())->data[0]['cnt'] ?? 0);

use Rasuvaeff\ClickHouseToolkit\ClickHouseFilterVisitor;
use Rasuvaeff\ClickHouseToolkit\ClickHouseQueryBuilder;

$qb = ClickHouseQueryBuilder::create(['id'], ['id' => 'UInt64'])
    ->withVisitor(new MyCustomVisitor());

use Rasuvaeff\ClickHouseToolkit\ClickHouseDataReader;
use Rasuvaeff\ClickHouseToolkit\ClickHouseDataType as T;
use Rasuvaeff\ClickHouseToolkit\ClickHouseQueryBuilder;
use Yiisoft\Data\Reader\Filter\Equals;
use Yiisoft\Data\Reader\Sort;

$reader = new ClickHouseDataReader(
    client: $client,
    table: 'events',
    queryBuilder: new ClickHouseQueryBuilder(
        allowedFields: ['id', 'type', 'created_at'],
    fieldTypes: ['id' => T::UInt64, 'created_at' => T::DateTime],
        defaultSort: 'id DESC',
    ),
    mapper: static fn (array $row): array => ['id' => (int) $row['id'], 'type' => (string) $row['type']],
    columns: ['id', 'type'],
);

$page = $reader
    ->withFilter(new Equals('type', 'click'))
    ->withSort(Sort::only(['id'])->withOrder(['id' => 'desc']))
    ->withLimit(20)
    ->withOffset(40);

$total = $page->count();   // ignores limit/offset
$rows  = $page->read();    // mapped values

use Rasuvaeff\ClickHouseToolkit\ClickHouseKeysetReader;
use Rasuvaeff\ClickHouseToolkit\ClickHouseQueryBuilder;
use Rasuvaeff\ClickHouseToolkit\ClickHouseDataType as T;
use Yiisoft\Data\Reader\Filter\Equals;

$reader = new ClickHouseKeysetReader(
    client: $client,
    table: 'events',
    queryBuilder: new ClickHouseQueryBuilder(
        allowedFields: ['id', 'status'],
        fieldTypes: ['id' => T::UInt64],
    ),
    mapper: static fn (array $row): int => (int) $row['id'],
    keyColumns: ['id' => T::UInt64],   // ordered map column => ClickHouse type
    columns: ['id', 'status'],          // key columns are added automatically
    pageSize: 1000,
    filter: new Equals('status', 'active'),
);

foreach ($reader->stream() as $id) {   // one page in memory at a time
    // ...
}

keyColumns: ['created_at' => T::DateTime, 'id' => T::UInt64],
// boundary: (created_at, id) > ({ck0:DateTime}, {ck1:UInt64})

use Rasuvaeff\ClickHouseToolkit\ClickHouseBatchWriter;

$writer = new ClickHouseBatchWriter(
    client: $client,
    table: 'events',
    columns: ['id', 'type', 'user_id', 'created_at'],
    batchSize: 1000,
);

$writer->write($rows); // $rows: iterable<array<string, mixed>> — a generator keeps memory flat

$writer = new ClickHouseBatchWriter(
    client: $client,
    table: 'events',
    columns: ['id', 'type', 'user_id', 'created_at'],
    batchSize: 10_000,
    settings: ['async_insert' => 1, 'wait_for_async_insert' => 0],
);

use Rasuvaeff\ClickHouseToolkit\ClickHouseDataType as T;
use Rasuvaeff\ClickHouseToolkit\ClickHouseTableBuilder;

ClickHouseTableBuilder::create($client, 'events')
    ->ifNotExists()
    ->column('id', T::UInt64)
    ->column('created_at', T::DateTime)
    ->engine('MergeTree()')
    ->partitionBy('toYYYYMM(created_at)')
    ->primaryKey('id')
    ->orderBy('(id, created_at)')
    ->execute();

use Rasuvaeff\ClickHouseToolkit\ClickHousePartitionManager;

$pm = new ClickHousePartitionManager($client);

foreach ($pm->getPartitions('events') as $p) {
    // ['partition' => '202401', 'partition_id' => '202401', 'rows' => 12345, 'bytes' => 987654]
}

$pm->dropPartition('events', '202401');
$pm->detachPartition('events', '202401');
$pm->attachPartition('events', '202401');
$pm->freezePartition('events', '202401');
$pm->clearColumnInPartition('events', '202401', 'payload');
$pm->movePartition('events', 'events_archive', '202401');     // MOVE … TO TABLE
$pm->replacePartition('events', 'events_mirror', '202401');   // REPLACE … FROM

use Rasuvaeff\ClickHouseToolkit\ClickHouseMutationBuilder;

$mb = new ClickHouseMutationBuilder($client);

$mb->update('events', 'status = {st:String}', 'id = {id:UInt64}', ['st' => 'archived', 'id' => 42]);
$mb->delete('events', 'created_at < {cutoff:DateTime}', ['cutoff' => '2023-01-01 00:00:00']);

$mb->waitForMutations('events', timeout: 30.0); // poll system.mutations until done -> bool

foreach ($mb->getMutations('events') as $m) {
    // ['mutation_id' => '...', 'command' => '...', 'is_done' => true, 'parts_to_do' => 0, 'latest_fail_reason' => '']
}

$mb->killMutation('events', $mutationId);

use Rasuvaeff\ClickHouseToolkit\ClickHouseMigrationRunner;

$runner = new ClickHouseMigrationRunner(
    client: $client,
    migrationsPath: __DIR__ . '/migrations',
    logger: $logger, // optional PSR-3
    placeholders: ['events_table' => 'my_events'], // optional; replaces {{events_table}}
);

$applied = $runner->run(); // list<string> of files applied this call

use Rasuvaeff\ClickHouseToolkit\ClickHouseMigrationGenerator;

$generator = new ClickHouseMigrationGenerator(__DIR__ . '/migrations');

$path = $generator->generate('add events index');
// Creates migrations/003_add_events_index.sql (003 = highest existing prefix + 1)
// with a header comment; write your DDL below the header.

use Rasuvaeff\ClickHouseToolkit\ClickHouseMigrationState;

$statuses = $runner->status(); // list<ClickHouseMigrationStatus>, sorted by name

foreach ($statuses as $status) {
    // $status->name       — '001_create_events.sql'
    // $status->state      — ClickHouseMigrationState::Applied
    // $status->checksum   — sha1 of the current file (null for Missing)
    // $status->appliedAt  — stored applied_at string (null for Pending)
}

use Rasuvaeff\ClickHouseToolkit\ClickHouseMigrationGenerator;
use Rasuvaeff\ClickHouseToolkit\ClickHouseMigrationRunner;
use Rasuvaeff\ClickHouseToolkit\Command\ClickHouseMigrationsGenerateCommand;
use Rasuvaeff\ClickHouseToolkit\Command\ClickHouseMigrationsRunCommand;
use Rasuvaeff\ClickHouseToolkit\Command\ClickHouseMigrationsStatusCommand;
use Symfony\Component\Console\Application;

$application = new Application('clickhouse-migrations');
$application->addCommands([
    new ClickHouseMigrationsGenerateCommand(new ClickHouseMigrationGenerator($migrationsPath)),
    new ClickHouseMigrationsStatusCommand($runner),
    new ClickHouseMigrationsRunCommand($runner),
]);
$application->run();

use Rasuvaeff\ClickHouseToolkit\ClickHouseDataType as T;

T::UInt64;                                  // 'UInt64'
T::nullable(T::String);                     // 'Nullable(String)'
T::array(T::nullable(T::String));           // 'Array(Nullable(String))'
T::map(T::String, T::UInt64);               // 'Map(String, UInt64)'
T::decimal(10, 2);                          // 'Decimal(10, 2)'
T::dateTime64(3, 'UTC');                    // "DateTime64(3, 'UTC')"
T::enum8(['active' => 1, 'inactive' => 2]); // "Enum8('active' = 1, 'inactive' = 2)"

$qb = new ClickHouseQueryBuilder(
    allowedFields: ['created_at'],
    fieldTypes: ['created_at' => T::DateTime],
    serverTimezone: 'UTC',
);

// A DateTimeImmutable in Europe/Moscow (+03:00) will be formatted as UTC.
$where = $qb->buildWhere(new Equals('created_at', new \DateTimeImmutable('2024-06-15 15:00:00+03:00')));
// params: ['p0' => '2024-06-15 12:00:00']

use Rasuvaeff\ClickHouseToolkit\ClickHouseClientFactory;
use Rasuvaeff\ClickHouseToolkit\ClickHouseConfig;
use Rasuvaeff\ClickHouseToolkit\ClickHouseMigrationRunner;
use Rasuvaeff\ClickHouseToolkit\ClickHouseMigrationRunnerInterface;
use SimPod\ClickHouseClient\Client\ClickHouseClient;
use SimPod\ClickHouseClient\Client\PsrClickHouseClient;

return [
    ClickHouseConfig::class => static fn (): ClickHouseConfig => new ClickHouseConfig(
        host: $_ENV['CLICKHOUSE_HOST'] ?? 'clickhouse',
        port: (int) ($_ENV['CLICKHOUSE_PORT'] ?? 8123),
        database: $_ENV['CLICKHOUSE_DB'] ?? 'app',
        username: $_ENV['CLICKHOUSE_USER'] ?? 'default',
        password: $_ENV['CLICKHOUSE_PASSWORD'] ?? '',
    ),

    PsrClickHouseClient::class => static fn (ClickHouseClientFactory $f): PsrClickHouseClient => $f->create(),
    ClickHouseClient::class => PsrClickHouseClient::class, // toolkit classes type-hint the interface

    ClickHouseMigrationRunnerInterface::class => static fn (ClickHouseClient $client): ClickHouseMigrationRunner => new ClickHouseMigrationRunner(
        client: $client,
        migrationsPath: dirname(__DIR__) . '/resources/clickhouse-migrations',
    ),
];