diff --git a/docs/业务数据分层存储.md b/docs/业务数据分层存储.md index a5c53ab1..c1d69fff 100644 --- a/docs/业务数据分层存储.md +++ b/docs/业务数据分层存储.md @@ -56,7 +56,7 @@ 1. `business-data:plan` 只读输出固定数据集、页面用途、目标层、同步策略、当前表体量与迁移阶段;固定数据集漏配或多配即失败。 2. `business-data:backfill-columnar ` 只从现有 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 查询完成真实页面验收后另行提交,避免有人只执行一条命令就绕过保护门。 diff --git a/modules/EcommerceFinance/Services/ClickHouseSchemaManager.php b/modules/EcommerceFinance/Services/ClickHouseSchemaManager.php index 19cedaea..f5612326 100644 --- a/modules/EcommerceFinance/Services/ClickHouseSchemaManager.php +++ b/modules/EcommerceFinance/Services/ClickHouseSchemaManager.php @@ -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 $column */ private function type(array $column): string { diff --git a/tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php b/tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php index f974a899..9bc51745 100644 --- a/tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php +++ b/tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php @@ -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 {