WuYongJie/CaiWuXiTong
1
0
Fork 0
Code Issues Pull Requests 2 Packages Projects Releases Wiki Activity

fix: 隔离 ClickHouse 回填代次 (#113)

Browse Source
This commit is contained in:
WuYongJie 2026-09-07 22:58:12 +08:00
parent a2b3db3afa
commit 62533293ae
3 changed files with 48 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/业务数据分层存储.md
View File

@ -56,7 +56,7 @@
1. `business-data:plan` 只读输出固定数据集、页面用途、目标层、同步策略、当前表体量与迁移阶段；固定数据集漏配或多配即失败。
2. `business-data:backfill-columnar <dataset>` 只从现有 MariaDB 镜像按 `_mirror_key` 断点读取，不连接上游；读取范围固定在任务启动时的 `_mirror_synced_at` 截止点。
3. ClickHouse 表使用 `ReplacingMergeTree`，以 `_mirror_key` 为排序/去重键，并按其哈希固定分为 32 个分区；同一业务键的所有版本必定落在同一分区。
3. ClickHouse 表使用 `ReplacingMergeTree`，以 `(_mirror_generation, _mirror_key)` 为排序/去重键，并按业务键哈希固定分为 32 个分区；同一代次内的业务键更新会被正确折叠，不同代次会同时保留。建表检查发现旧表缺少代次键时会拒绝继续回填，必须先以影子表完成可回滚修复。
4. 回填结束同时核对截止点精确行数、ClickHouse `FINAL` 行数和唯一键数；不一致只记录失败，不切换页面。
5. `--archive` 在对账后把 32 个哈希分区分别输出为 Parquet，经现有私有七牛 S3 磁盘流式上传；每个对象记录路径、行数、字节数和逻辑校验值。
6. 本阶段没有“激活”或“删除”命令。页面切换与 MariaDB 清理必须在各 Repository 的 ClickHouse 查询完成真实页面验收后另行提交，避免有人只执行一条命令就绕过保护门。

23
modules/EcommerceFinance/Services/ClickHouseSchemaManager.php
View File

@ -32,17 +32,18 @@ class ClickHouseSchemaManager
$definitions[] = '`_mirror_version` UInt64';
// Hash partitioning keeps every version of one business key in the same
// partition, so ReplacingMergeTree can reconcile updates without relying
// on a business date that might later be corrected upstream.
// partition. The generation must also be part of the replacement key:
// otherwise a newer backfill generation would erase the retained one.
$sql = sprintf(
'CREATE TABLE IF NOT EXISTS %s (%s) '
. 'ENGINE = ReplacingMergeTree(_mirror_version) '
. 'PARTITION BY modulo(cityHash64(_mirror_key), 32) '
. 'ORDER BY (_mirror_key) SETTINGS index_granularity = 8192',
. 'ORDER BY (_mirror_generation, _mirror_key) SETTINGS index_granularity = 8192',
$this->clickHouse->qualifiedTable($dataset->targetTable()),
implode(', ', $definitions),
);
$this->clickHouse->command($sql);
$this->assertGenerationScopedSortingKey($dataset);
return $columns;
}
@ -58,6 +59,22 @@ class ClickHouseSchemaManager
) === 1;
}
private function assertGenerationScopedSortingKey(BusinessMirrorDataset $dataset): void
{
$sortingKey = (string) $this->clickHouse->scalar(
'SELECT sorting_key FROM system.tables '
. 'WHERE database = {database:String} AND name = {table:String}',
[
'database' => (string) config('business_data.clickhouse.database', 'caiwu_business_analytics'),
'table' => $dataset->targetTable(),
],
);
$normalized = strtolower((string) preg_replace('/[`()\s]+/', '', $sortingKey));
if ($normalized !== '_mirror_generation,_mirror_key') {
throw new RuntimeException('business_data_columnar_generation_key_mismatch');
}
}
/** @param array<string,mixed> $column */
private function type(array $column): string
{

28
tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php
View File

@ -56,13 +56,18 @@ class ColumnarBusinessMirrorRepositoryTest extends TestCase
$client = Mockery::mock(ClickHouseClient::class);
$client->shouldReceive('identifier')->times(4)->andReturnUsing(static fn (string $name): string => '`' . $name . '`');
$client->shouldReceive('qualifiedTable')->once()->with('fact_rows')->andReturn('`analytics`.`fact_rows`');
$client->shouldReceive('scalar')->once()->withArgs(static fn (string $sql, array $parameters): bool =>
str_contains($sql, 'SELECT sorting_key FROM system.tables')
&& $parameters['database'] === 'caiwu_business_analytics'
&& $parameters['table'] === 'fact_rows'
)->andReturn('_mirror_generation, _mirror_key');
$client->shouldReceive('command')->once()->withArgs(static fn (string $sql): bool =>
str_contains($sql, '`flag` Nullable(UInt8)')
&& str_contains($sql, '`amount` Nullable(Decimal(20, 4))')
&& str_contains($sql, "`updated_at` Nullable(DateTime64(6, 'Asia/Shanghai'))")
&& str_contains($sql, 'ENGINE = ReplacingMergeTree(_mirror_version)')
&& str_contains($sql, 'PARTITION BY modulo(cityHash64(_mirror_key), 32)')
&& str_contains($sql, 'ORDER BY (_mirror_key)')
&& str_contains($sql, 'ORDER BY (_mirror_generation, _mirror_key)')
);
$columns = (new ClickHouseSchemaManager($source, $client, $placements))->ensure($dataset);
@ -70,6 +75,27 @@ class ColumnarBusinessMirrorRepositoryTest extends TestCase
$this->assertCount(4, $columns);
}
public function test_columnar_schema_rejects_a_generation_unsafe_existing_sorting_key(): void
{
$dataset = $this->dataset();
$source = Mockery::mock(BusinessMirrorSchemaManager::class);
$source->shouldReceive('columns')->once()->andReturn([
['column_name' => 'Id', 'data_type' => 'int'],
]);
$placements = Mockery::mock(BusinessDataPlacement::class);
$placements->shouldReceive('targetStore')->once()->with('source.schema.fact')->andReturn('clickhouse');
$client = Mockery::mock(ClickHouseClient::class);
$client->shouldReceive('identifier')->once()->with('Id')->andReturn('`Id`');
$client->shouldReceive('qualifiedTable')->once()->with('fact_rows')->andReturn('`analytics`.`fact_rows`');
$client->shouldReceive('command')->once();
$client->shouldReceive('scalar')->once()->andReturn('_mirror_key');
$this->expectException(\RuntimeException::class);
$this->expectExceptionMessage('business_data_columnar_generation_key_mismatch');
(new ClickHouseSchemaManager($source, $client, $placements))->ensure($dataset);
}
private function dataset(): BusinessMirrorDataset
{
return new class implements BusinessMirrorDataset {