diff --git a/pkg/dms-common/pkg/catalogschema/catalog_schema.go b/pkg/dms-common/pkg/catalogschema/catalog_schema.go new file mode 100644 index 000000000..fdb57f291 --- /dev/null +++ b/pkg/dms-common/pkg/catalogschema/catalog_schema.go @@ -0,0 +1,93 @@ +// Package catalogschema 提供 SQL Server「合并层」catalog/schema 名称的解析、拼接与 +// 过滤工具。SQL Server 的库/模式为两级结构（database.schema），DMS 在选库入口将其 +// 合并为单层名称（整库 "db" 或 "db.schema"）呈现；本包统一维护这一层名称语义，供 +// dms-ee 结构采集/脱敏与 provision 建账户授权两侧共享，避免两端各自实现导致行为漂移。 +package catalogschema + +import "strings" + +// dbTypeSQLServer 与 dms internal 常量 DBTypeSQLServer、provision driver.DBTypeSQLServer +// 取值保持一致；此处以字符串常量内联，避免公共包反向依赖各仓库的 internal 常量。 +const dbTypeSQLServer = "SQL Server" + +// defaultSchemaName 是 SQL Server 库下的默认 schema。 +const defaultSchemaName = "dbo" + +// CatalogSchemaSplit 表示一个合并层名称拆分后的结果。 +// WholeDB 为 true 表示该名称指向整库（无显式 schema）。 +type CatalogSchemaSplit struct { + Catalog string + Schema string + WholeDB bool +} + +// IsSchemaLevelDBType 标识需要 database+schema 合并层的数据库类型。 +func IsSchemaLevelDBType(dbType string) bool { + return strings.EqualFold(dbType, dbTypeSQLServer) +} + +// BuildFullCatalogSchemaName 合并 catalog 与 schema 为 "catalog.schema"；schema 为空时默认 dbo。 +func BuildFullCatalogSchemaName(catalog, schemaName string) string { + if catalog == "" { + return schemaName + } + if schemaName == "" { + schemaName = defaultSchemaName + } + return catalog + "." + schemaName +} + +// SplitCatalogSchema 将合并名拆回 catalog/schema；无点表示整库作用域。 +// +// 合并名由 BuildFullCatalogSchemaName 以 "catalog.schema" 生成。因数据库名与 +// schema 名理论上都可能含 "."，无法从字符串完美还原；这里以最后一个 "." 作为 +// 分隔（LastIndex），即假定 schema 段不含 "."（SQL Server 默认 dbo 等均满足）， +// 从而正确兼容数据库名本身含 "." 的情形（如 "my.db.dbo" → catalog="my.db"）。 +func SplitCatalogSchema(dbType, name string) CatalogSchemaSplit { + if !IsSchemaLevelDBType(dbType) || name == "" { + return CatalogSchemaSplit{Schema: name} + } + if idx := strings.LastIndex(name, "."); idx > 0 { + return CatalogSchemaSplit{ + Catalog: name[:idx], + Schema: name[idx+1:], + } + } + return CatalogSchemaSplit{ + Catalog: name, + WholeDB: true, + } +} + +// SchemaNameMatchesFilter 判断 schema 名是否命中任务过滤（支持整库 db 匹配其下 db.schema）。 +func SchemaNameMatchesFilter(dbType string, filter map[string]struct{}, schemaName string) bool { + if len(filter) == 0 { + return true + } + if _, ok := filter[schemaName]; ok { + return true + } + if !IsSchemaLevelDBType(dbType) { + return false + } + for selected := range filter { + selectedSplit := SplitCatalogSchema(dbType, selected) + if !selectedSplit.WholeDB { + continue + } + nameSplit := SplitCatalogSchema(dbType, schemaName) + if nameSplit.Catalog == selectedSplit.Catalog { + return true + } + } + return false +} + +// ConnectionCatalogForMergedName 返回 JDBC/连接串使用的 catalog（database）名。 +func ConnectionCatalogForMergedName(dbType, mergedName string) string { + split := SplitCatalogSchema(dbType, mergedName) + if split.Catalog != "" { + return split.Catalog + } + return "" +} diff --git a/pkg/dms-common/pkg/catalogschema/catalog_schema_test.go b/pkg/dms-common/pkg/catalogschema/catalog_schema_test.go new file mode 100644 index 000000000..a6a1c9780 --- /dev/null +++ b/pkg/dms-common/pkg/catalogschema/catalog_schema_test.go @@ -0,0 +1,83 @@ +package catalogschema + +import "testing" + +const testDBType = "SQL Server" + +func TestIsSchemaLevelDBType(t *testing.T) { + if !IsSchemaLevelDBType("SQL Server") { + t.Fatalf("SQL Server should be schema-level") + } + if !IsSchemaLevelDBType("sql server") { + t.Fatalf("match should be case-insensitive") + } + if IsSchemaLevelDBType("MySQL") { + t.Fatalf("MySQL should not be schema-level") + } +} + +func TestBuildFullCatalogSchemaName(t *testing.T) { + if got := BuildFullCatalogSchemaName("TestDB", "dbo"); got != "TestDB.dbo" { + t.Fatalf("got %q, want TestDB.dbo", got) + } + if got := BuildFullCatalogSchemaName("TestDB", ""); got != "TestDB.dbo" { + t.Fatalf("empty schema should default to dbo, got %q", got) + } + if got := BuildFullCatalogSchemaName("", "public"); got != "public" { + t.Fatalf("empty catalog should return schema as-is, got %q", got) + } +} + +func TestSchemaNameMatchesFilter(t *testing.T) { + filter := map[string]struct{}{"TestDB": {}} + if !SchemaNameMatchesFilter(testDBType, filter, "TestDB.dbo") { + t.Fatalf("whole-db filter should match its schema") + } + if SchemaNameMatchesFilter(testDBType, filter, "OtherDB.dbo") { + t.Fatalf("whole-db filter should not match a different db") + } + if !SchemaNameMatchesFilter(testDBType, map[string]struct{}{}, "AnyDB.dbo") { + t.Fatalf("empty filter should match everything") + } + if !SchemaNameMatchesFilter(testDBType, map[string]struct{}{"TestDB.dbo": {}}, "TestDB.dbo") { + t.Fatalf("exact filter should match") + } +} + +func TestSplitCatalogSchema(t *testing.T) { + cases := []struct { + name string + dbType string + in string + catalog string + schema string + whole bool + }{ + {"whole db", testDBType, "TestDB", "TestDB", "", true}, + {"db.schema", testDBType, "TestDB.dbo", "TestDB", "dbo", false}, + {"dotted catalog", testDBType, "my.db.dbo", "my.db", "dbo", false}, + {"empty", testDBType, "", "", "", false}, + {"non schema-level", "MySQL", "TestDB.dbo", "", "TestDB.dbo", false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := SplitCatalogSchema(tc.dbType, tc.in) + if got.Catalog != tc.catalog || got.Schema != tc.schema || got.WholeDB != tc.whole { + t.Fatalf("SplitCatalogSchema(%q, %q) = %+v, want catalog=%q schema=%q whole=%v", + tc.dbType, tc.in, got, tc.catalog, tc.schema, tc.whole) + } + }) + } +} + +func TestConnectionCatalogForMergedName(t *testing.T) { + if got := ConnectionCatalogForMergedName(testDBType, "TestDB.dbo"); got != "TestDB" { + t.Fatalf("got %q, want TestDB", got) + } + if got := ConnectionCatalogForMergedName(testDBType, "TestDB"); got != "TestDB" { + t.Fatalf("whole-db name should return catalog, got %q", got) + } + if got := ConnectionCatalogForMergedName("MySQL", "somedb"); got != "" { + t.Fatalf("non schema-level should return empty, got %q", got) + } +}