fix: resume verified archive generations
This commit is contained in:
parent
628646b9f8
commit
46fd357b59
@ -55,6 +55,25 @@ class BackfillBusinessDataColumnar extends Command
|
||||
$columnNames = array_map(static fn (array $column): string => (string) $column['column_name'], $columns);
|
||||
$state = $placements->state($datasetKey);
|
||||
$restart = (bool) $this->option('restart');
|
||||
$verified = ! $restart
|
||||
&& is_array($state)
|
||||
&& in_array((string) ($state['phase'] ?? ''), ['verified', 'active'], true)
|
||||
&& ! empty($state['backfill_generation'])
|
||||
&& ! empty($state['verified_at']);
|
||||
if ($verified) {
|
||||
$generation = (string) $state['backfill_generation'];
|
||||
$this->info(sprintf(
|
||||
'%s:复用已对账代次 %s(%d 行);MariaDB 原表仍保留。',
|
||||
$datasetKey,
|
||||
$generation,
|
||||
(int) ($state['target_rows'] ?? 0),
|
||||
));
|
||||
if ((bool) $this->option('archive')) {
|
||||
$this->archive($datasetKey, $dataset, $generation, $columnar, $placements);
|
||||
}
|
||||
|
||||
return self::SUCCESS;
|
||||
}
|
||||
$resume = ! $restart
|
||||
&& is_array($state)
|
||||
&& in_array((string) ($state['phase'] ?? ''), ['backfilling', 'failed'], true)
|
||||
@ -193,6 +212,17 @@ class BackfillBusinessDataColumnar extends Command
|
||||
for ($partition = 0; $partition < 32; $partition++) {
|
||||
$partitionKey = sprintf('hash-%02d', $partition);
|
||||
$objectKey = sprintf('%s/%s/%s/%s.parquet', $prefix, $slug, $generation, $partitionKey);
|
||||
$existing = $placements->connection()->table('business_data_archives')
|
||||
->where('dataset_key', $datasetKey)
|
||||
->where('generation', $generation)
|
||||
->where('partition_key', $partitionKey)
|
||||
->first();
|
||||
if ($existing !== null
|
||||
&& (string) ($existing->status ?? '') === 'verified'
|
||||
&& $disk->exists($objectKey)
|
||||
&& (int) $disk->size($objectKey) === (int) ($existing->byte_size ?? 0)) {
|
||||
continue;
|
||||
}
|
||||
$stream = $columnar->parquetPartition($dataset, $generation, $partition);
|
||||
$resource = StreamWrapper::getResource($stream);
|
||||
try {
|
||||
|
||||
@ -107,7 +107,7 @@ class BusinessDataColumnarBackfillTest extends TestCase
|
||||
['column_name' => 'updated_at', 'data_type' => 'datetime2'],
|
||||
];
|
||||
$schema = Mockery::mock(ClickHouseSchemaManager::class);
|
||||
$schema->shouldReceive('ensure')->once()->andReturn($columns);
|
||||
$schema->shouldReceive('ensure')->twice()->andReturn($columns);
|
||||
$columnar = Mockery::mock(ColumnarBusinessMirrorRepository::class);
|
||||
$columnar->shouldReceive('storeBackfillBatch')->once();
|
||||
$columnar->shouldReceive('count')->once()->andReturn(2);
|
||||
@ -119,14 +119,17 @@ class BusinessDataColumnarBackfillTest extends TestCase
|
||||
'logical_checksum' => '0000000000000000',
|
||||
]);
|
||||
$clickHouse = Mockery::mock(ClickHouseClient::class);
|
||||
$clickHouse->shouldReceive('enabled')->once()->andReturnTrue();
|
||||
$clickHouse->shouldReceive('ping')->once()->andReturnTrue();
|
||||
$clickHouse->shouldReceive('enabled')->twice()->andReturnTrue();
|
||||
$clickHouse->shouldReceive('ping')->twice()->andReturnTrue();
|
||||
$this->app->instance(ClickHouseSchemaManager::class, $schema);
|
||||
$this->app->instance(ColumnarBusinessMirrorRepository::class, $columnar);
|
||||
$this->app->instance(ClickHouseClient::class, $clickHouse);
|
||||
|
||||
$this->artisan('business-data:backfill-columnar', [
|
||||
'dataset' => 'source.schema.fact', '--batch' => 100, '--max-batches' => 1, '--archive' => true,
|
||||
'dataset' => 'source.schema.fact', '--batch' => 100, '--max-batches' => 1,
|
||||
])->expectsOutputToContain('已对账')->assertSuccessful();
|
||||
$this->artisan('business-data:backfill-columnar', [
|
||||
'dataset' => 'source.schema.fact', '--archive' => true,
|
||||
])->expectsOutputToContain('32 个 Parquet 哈希分区')->assertSuccessful();
|
||||
|
||||
$archives = DB::connection('business_mirror')->table('business_data_archives')->get();
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user