fix: harden columnar schema and long exports
This commit is contained in:
parent
31d99539fd
commit
9b9a42aa82
@ -25,6 +25,7 @@ return [
|
||||
'password' => env('BUSINESS_CLICKHOUSE_PASSWORD', ''),
|
||||
'connect_timeout_seconds' => (float) env('BUSINESS_CLICKHOUSE_CONNECT_TIMEOUT', 3),
|
||||
'query_timeout_seconds' => (float) env('BUSINESS_CLICKHOUSE_QUERY_TIMEOUT', 30),
|
||||
'stream_timeout_seconds' => (float) env('BUSINESS_CLICKHOUSE_STREAM_TIMEOUT', 1800),
|
||||
],
|
||||
|
||||
'archive' => [
|
||||
|
||||
@ -157,6 +157,9 @@ class ClickHouseClient
|
||||
}
|
||||
|
||||
$options = $this->authOptions() + ['query' => $query, 'stream' => $stream];
|
||||
if ($stream) {
|
||||
$options['timeout'] = (float) $this->config()['stream_timeout_seconds'];
|
||||
}
|
||||
if ($body !== null) {
|
||||
$options['body'] = $body;
|
||||
}
|
||||
@ -196,6 +199,7 @@ class ClickHouseClient
|
||||
'password' => '',
|
||||
'connect_timeout_seconds' => 3,
|
||||
'query_timeout_seconds' => 30,
|
||||
'stream_timeout_seconds' => 1800,
|
||||
];
|
||||
}
|
||||
}
|
||||
|
||||
@ -3,6 +3,7 @@
|
||||
namespace Modules\EcommerceFinance\Services;
|
||||
|
||||
use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
|
||||
use RuntimeException;
|
||||
|
||||
class ClickHouseSchemaManager
|
||||
{
|
||||
@ -16,6 +17,10 @@ class ClickHouseSchemaManager
|
||||
/** @return list<array<string,mixed>> */
|
||||
public function ensure(BusinessMirrorDataset $dataset): array
|
||||
{
|
||||
if (! in_array($this->placements->targetStore($dataset->key()), ['clickhouse', 'parquet'], true)) {
|
||||
throw new RuntimeException('business_data_dataset_not_columnar');
|
||||
}
|
||||
|
||||
$columns = $this->sourceSchemas->columns($dataset);
|
||||
$definitions = ['`_mirror_key` String'];
|
||||
foreach ($columns as $column) {
|
||||
@ -62,7 +67,8 @@ class ClickHouseSchemaManager
|
||||
|
||||
return match ($type) {
|
||||
'bit', 'boolean' => 'UInt8',
|
||||
'tinyint' => 'Int8',
|
||||
// SQL Server tinyint is unsigned (0..255), unlike MySQL tinyint.
|
||||
'tinyint' => 'UInt8',
|
||||
'smallint' => 'Int16',
|
||||
'int', 'integer' => 'Int32',
|
||||
'bigint' => 'Int64',
|
||||
|
||||
@ -20,6 +20,7 @@ class ClickHouseClientTest extends TestCase
|
||||
new Response(200, [], "Ok.\n"),
|
||||
new Response(200, [], "{\"value\":1}\n{\"value\":2}\n"),
|
||||
new Response(200, [], ''),
|
||||
new Response(200, [], 'PAR1'),
|
||||
]));
|
||||
$handler->push(Middleware::history($history));
|
||||
$http = new Client(['handler' => $handler]);
|
||||
@ -41,6 +42,8 @@ class ClickHouseClientTest extends TestCase
|
||||
parse_str($history[2]['request']->getUri()->getQuery(), $insertQuery);
|
||||
$this->assertStringContainsString('FORMAT JSONEachRow', $insertQuery['query']);
|
||||
$this->assertSame("{\"id\":1,\"name\":\"测试\"}\n", (string) $history[2]['request']->getBody());
|
||||
$this->assertSame('PAR1', (string) $client->stream('SELECT 1', 'Parquet'));
|
||||
$this->assertSame(1800.0, $history[3]['options']['timeout']);
|
||||
}
|
||||
|
||||
public function test_invalid_identifier_is_rejected_before_a_query_is_sent(): void
|
||||
|
||||
@ -5,7 +5,10 @@ namespace Tests\Unit\EcommerceFinance;
|
||||
use Mockery;
|
||||
use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
|
||||
use Modules\EcommerceFinance\Repositories\ColumnarBusinessMirrorRepository;
|
||||
use Modules\EcommerceFinance\Services\BusinessDataPlacement;
|
||||
use Modules\EcommerceFinance\Services\BusinessMirrorSchemaManager;
|
||||
use Modules\EcommerceFinance\Services\ClickHouseClient;
|
||||
use Modules\EcommerceFinance\Services\ClickHouseSchemaManager;
|
||||
use Tests\TestCase;
|
||||
|
||||
class ColumnarBusinessMirrorRepositoryTest extends TestCase
|
||||
@ -38,6 +41,35 @@ class ColumnarBusinessMirrorRepositoryTest extends TestCase
|
||||
]], '11111111-1111-4111-8111-111111111111');
|
||||
}
|
||||
|
||||
public function test_columnar_schema_uses_replacing_merge_tree_and_sql_server_safe_types(): void
|
||||
{
|
||||
$dataset = $this->dataset();
|
||||
$source = Mockery::mock(BusinessMirrorSchemaManager::class);
|
||||
$source->shouldReceive('columns')->once()->andReturn([
|
||||
['column_name' => 'Id', 'data_type' => 'int'],
|
||||
['column_name' => 'flag', 'data_type' => 'tinyint'],
|
||||
['column_name' => 'amount', 'data_type' => 'decimal', 'numeric_precision' => 20, 'numeric_scale' => 4],
|
||||
['column_name' => 'updated_at', 'data_type' => 'datetime2'],
|
||||
]);
|
||||
$placements = Mockery::mock(BusinessDataPlacement::class);
|
||||
$placements->shouldReceive('targetStore')->once()->with('source.schema.fact')->andReturn('clickhouse');
|
||||
$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('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)')
|
||||
);
|
||||
|
||||
$columns = (new ClickHouseSchemaManager($source, $client, $placements))->ensure($dataset);
|
||||
|
||||
$this->assertCount(4, $columns);
|
||||
}
|
||||
|
||||
private function dataset(): BusinessMirrorDataset
|
||||
{
|
||||
return new class implements BusinessMirrorDataset {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user