diff --git a/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php b/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php index 6befd077..a70f0a05 100644 --- a/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php +++ b/modules/EcommerceFinance/Services/UpstreamSchemaCatalogSynchronizer.php @@ -53,10 +53,7 @@ class UpstreamSchemaCatalogSynchronizer } $now = now(); - $normalized = array_map( - fn (array $row): array => $this->normalize($definition['key'], $row, $now), - $rows - ); + $normalized = $this->normalizeAndDeduplicate($definition['key'], $rows, $now); $objectCount = collect($normalized) ->map(fn (array $row): string => $row['schema_name'] . '.' . $row['object_name']) ->unique() @@ -116,6 +113,30 @@ class UpstreamSchemaCatalogSynchronizer ]; } + /** + * 部分旧 SQL Server 库的 INFORMATION_SCHEMA 会返回重复字段。 + * 共享库的目录键不允许重复,因此写入前按数据库常见的不区分大小写规则去重, + * 避免少量脏元数据让整个来源的目录同步回滚。 + */ + private function normalizeAndDeduplicate(string $source, array $rows, $now): array + { + $unique = []; + + foreach ($rows as $row) { + $normalized = $this->normalize($source, $row, $now); + $identity = implode("\0", [ + mb_strtolower($normalized['source_key'], 'UTF-8'), + mb_strtolower($normalized['schema_name'], 'UTF-8'), + mb_strtolower($normalized['object_name'], 'UTF-8'), + mb_strtolower($normalized['column_name'], 'UTF-8'), + ]); + + $unique[$identity] ??= $normalized; + } + + return array_values($unique); + } + private function nullableInteger(mixed $value): ?int { return $value === null || $value === '' ? null : (int) $value; diff --git a/tests/Unit/EcommerceFinance/UpstreamSchemaCatalogSynchronizerTest.php b/tests/Unit/EcommerceFinance/UpstreamSchemaCatalogSynchronizerTest.php index 7fa5e1bf..e9b98d62 100644 --- a/tests/Unit/EcommerceFinance/UpstreamSchemaCatalogSynchronizerTest.php +++ b/tests/Unit/EcommerceFinance/UpstreamSchemaCatalogSynchronizerTest.php @@ -101,6 +101,51 @@ class UpstreamSchemaCatalogSynchronizerTest extends TestCase } } + public function test_duplicate_metadata_rows_do_not_abort_catalog_sync(): void + { + $duplicate = [ + 'schema_name' => 'dbo', + 'object_name' => 'tuihuo', + 'object_type' => 'table', + 'column_name' => '发货量', + 'ordinal_position' => 3, + 'data_type' => 'decimal', + 'max_length' => null, + 'numeric_precision' => 18, + 'numeric_scale' => 2, + 'is_nullable' => 'YES', + ]; + $query = Mockery::mock(UpstreamReadOnlyQuery::class); + $query->shouldReceive('select')->once()->andReturn([ + $duplicate, + $duplicate, + [ + ...$duplicate, + 'column_name' => '序号', + 'ordinal_position' => 1, + 'data_type' => 'int', + 'numeric_precision' => 10, + 'numeric_scale' => 0, + ], + ]); + $factory = Mockery::mock(UpstreamReadOnlyQueryFactory::class); + $factory->shouldReceive('forSource')->once()->with('catalog_test')->andReturn($query); + + $result = (new UpstreamSchemaCatalogSynchronizer( + app(UpstreamSourceRegistry::class), + $factory + ))->sync('catalog_test'); + + $this->assertSame(1, $result['objects']); + $this->assertSame(2, $result['columns']); + $this->assertSame(1, DB::table('upstream_schema_columns') + ->where('source_key', 'catalog_test') + ->where('schema_name', 'dbo') + ->where('object_name', 'tuihuo') + ->where('column_name', '发货量') + ->count()); + } + private function existingRow(): array { return [