diff --git a/config/business_data.php b/config/business_data.php index 9bee3e39..7cca91bc 100644 --- a/config/business_data.php +++ b/config/business_data.php @@ -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' => [ diff --git a/modules/EcommerceFinance/Services/ClickHouseClient.php b/modules/EcommerceFinance/Services/ClickHouseClient.php index dc7cbc1b..3336b1c1 100644 --- a/modules/EcommerceFinance/Services/ClickHouseClient.php +++ b/modules/EcommerceFinance/Services/ClickHouseClient.php @@ -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, ]; } } diff --git a/modules/EcommerceFinance/Services/ClickHouseSchemaManager.php b/modules/EcommerceFinance/Services/ClickHouseSchemaManager.php index 92080fa5..19cedaea 100644 --- a/modules/EcommerceFinance/Services/ClickHouseSchemaManager.php +++ b/modules/EcommerceFinance/Services/ClickHouseSchemaManager.php @@ -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> */ 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', diff --git a/tests/Unit/EcommerceFinance/ClickHouseClientTest.php b/tests/Unit/EcommerceFinance/ClickHouseClientTest.php index 3bcb9be8..dac3d142 100644 --- a/tests/Unit/EcommerceFinance/ClickHouseClientTest.php +++ b/tests/Unit/EcommerceFinance/ClickHouseClientTest.php @@ -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 diff --git a/tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php b/tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php index 67f08f66..f974a899 100644 --- a/tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php +++ b/tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php @@ -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 {