🐛 fix(db): 修复Oracle同义词字段注释回填

- 元数据查不到列定义时补查 ALL_SYNONYMS 解析真实表
- 对解析出的 owner 和 table 继续回查列注释与主键信息
- 新增 SBDEV 指向 DEV.PERSON_INFO 的字段注释回归测试
This commit is contained in:
Syngnat
2026-07-06 16:59:46 +08:00
parent ac68729734
commit 2a3bd5bd97
2 changed files with 121 additions and 0 deletions

View File

@@ -112,6 +112,60 @@ func TestOracleGetColumnsIncludesColumnComments(t *testing.T) {
}
}
func TestOracleGetColumnsResolvesSynonymTargetComments(t *testing.T) {
t.Parallel()
dbConn, state := openOracleRecordingDB(t)
state.mu.Lock()
state.disableDefaultTabColumns = true
state.queryResults[buildOracleSynonymLookupQuery("SBDEV", "PERSON_INFO")] = oracleRecordingQueryResult{
columns: []string{"TABLE_OWNER", "TABLE_NAME"},
rows: [][]driver.Value{
{"DEV", "PERSON_INFO"},
},
}
state.queryResults[buildOracleColumnsQuery("DEV", "PERSON_INFO")] = oracleRecordingQueryResult{
columns: []string{"COLUMN_NAME", "DATA_TYPE", "NULLABLE", "DATA_DEFAULT", "COLUMN_KEY", "COMMENT"},
rows: [][]driver.Value{
{"PID", "CHAR", "NO", nil, "PRI", "个人标识"},
{"XM", "VARCHAR2", "YES", nil, "", "姓名"},
},
}
state.mu.Unlock()
oracleDB := &OracleDB{conn: dbConn}
columns, err := oracleDB.GetColumns("SBDEV", "PERSON_INFO")
if err != nil {
t.Fatalf("GetColumns 返回错误: %v", err)
}
if len(columns) != 2 {
t.Fatalf("expected synonym target columns, got %#v", columns)
}
if columns[0].Name != "PID" || columns[0].Comment != "个人标识" || columns[0].Key != "PRI" {
t.Fatalf("expected first synonym column metadata from DEV.PERSON_INFO, got %#v", columns[0])
}
if columns[1].Name != "XM" || columns[1].Comment != "姓名" {
t.Fatalf("expected second synonym column comment from DEV.PERSON_INFO, got %#v", columns[1])
}
queries := state.snapshotQueries()
if len(queries) < 3 {
t.Fatalf("expected direct metadata probe + synonym lookup + target metadata probe, got %v", queries)
}
if queries[0] != buildOracleColumnsQuery("SBDEV", "PERSON_INFO") {
t.Fatalf("expected first metadata probe to use synonym owner, got %v", queries)
}
if !slices.Contains(queries, buildOracleSynonymLookupQuery("SBDEV", "PERSON_INFO")) {
t.Fatalf("expected synonym lookup against all_synonyms, got %v", queries)
}
if !slices.Contains(queries, buildOracleColumnsQuery("DEV", "PERSON_INFO")) {
t.Fatalf("expected metadata probe against resolved target table, got %v", queries)
}
if slices.Contains(queries, `SELECT * FROM "SBDEV"."PERSON_INFO" WHERE 1 = 0`) {
t.Fatalf("expected synonym metadata to resolve before falling back to empty-select inference, got %v", queries)
}
}
func TestOracleColumnsQueryFiltersPrimaryKeyLookupByTargetTable(t *testing.T) {
t.Parallel()

View File

@@ -384,6 +384,18 @@ func (o *OracleDB) GetColumns(dbName, tableName string) ([]connection.ColumnDefi
return parseOracleColumns(data), nil
}
for _, target := range o.lookupOracleSynonymTargets(dbName, tableName) {
query := buildOracleColumnsQuery(target.schema, target.table)
data, _, err := o.Query(query)
if err != nil {
return nil, err
}
if len(data) == 0 {
continue
}
return parseOracleColumns(data), nil
}
if columns, err := o.inferOracleColumnsFromSelect(dbName, tableName); err == nil && len(columns) > 0 {
return columns, nil
}
@@ -718,6 +730,61 @@ func oracleMetadataNamePairs(dbName string, tableName string) []oracleMetadataNa
return pairs
}
func (o *OracleDB) lookupOracleSynonymTargets(dbName string, tableName string) []oracleMetadataNamePair {
targets := make([]oracleMetadataNamePair, 0, 4)
seen := map[string]struct{}{}
add := func(schema string, table string) {
key := schema + "\x00" + table
if strings.TrimSpace(table) == "" {
return
}
if _, exists := seen[key]; exists {
return
}
seen[key] = struct{}{}
targets = append(targets, oracleMetadataNamePair{schema: schema, table: table})
}
for _, candidate := range oracleMetadataNamePairs(dbName, tableName) {
data, _, err := o.Query(buildOracleSynonymLookupQuery(candidate.schema, candidate.table))
if err != nil {
continue
}
for _, row := range data {
targetSchema := oracleRowString(row, "TABLE_OWNER", "table_owner")
targetTable := oracleRowString(row, "TABLE_NAME", "table_name")
for _, target := range oracleMetadataNamePairs(targetSchema, targetTable) {
add(target.schema, target.table)
}
if strings.TrimSpace(targetTable) != "" {
break
}
}
}
return targets
}
func buildOracleSynonymLookupQuery(schema string, table string) string {
metadataTableName := escapeOracleMetadataLiteralExact(table)
if strings.TrimSpace(schema) == "" {
return fmt.Sprintf(`SELECT table_owner AS "TABLE_OWNER", table_name AS "TABLE_NAME"
FROM all_synonyms
WHERE synonym_name = '%s'
AND db_link IS NULL
AND (owner = USER OR owner = 'PUBLIC')
ORDER BY CASE WHEN owner = USER THEN 0 WHEN owner = 'PUBLIC' THEN 1 ELSE 2 END`, metadataTableName)
}
metadataSchemaName := escapeOracleMetadataLiteralExact(schema)
return fmt.Sprintf(`SELECT table_owner AS "TABLE_OWNER", table_name AS "TABLE_NAME"
FROM all_synonyms
WHERE synonym_name = '%s'
AND db_link IS NULL
AND owner IN ('%s', 'PUBLIC')
ORDER BY CASE WHEN owner = '%s' THEN 0 WHEN owner = 'PUBLIC' THEN 1 ELSE 2 END`, metadataTableName, metadataSchemaName, metadataSchemaName)
}
func buildOracleColumnsQuery(schema string, table string) string {
metadataTableName := escapeOracleMetadataLiteralExact(table)
metadataSchemaName := escapeOracleMetadataLiteralExact(schema)