diff --git a/agents/drivers/kingbase-go/integration_test.go b/agents/drivers/kingbase-go/integration_test.go index b5e94cf570..0496a744c0 100644 --- a/agents/drivers/kingbase-go/integration_test.go +++ b/agents/drivers/kingbase-go/integration_test.go @@ -41,11 +41,16 @@ func TestKingbaseIntegration(t *testing.T) { t.Fatal(err) } t.Cleanup(func() { _ = server.disconnect() }) + schema, err := server.effectiveSchema("") + if err != nil { + t.Fatal(err) + } + qualifiedSchema := quoteIdentifier(schema) cleanup := []string{ - "DROP VIEW IF EXISTS public." + quoteIdentifier(view), - "DROP FUNCTION IF EXISTS public." + quoteIdentifier(function) + "()", - "DROP TABLE IF EXISTS public." + quoteIdentifier(child), - "DROP TABLE IF EXISTS public." + quoteIdentifier(parent), + "DROP VIEW IF EXISTS " + qualifiedSchema + "." + quoteIdentifier(view), + "DROP FUNCTION IF EXISTS " + qualifiedSchema + "." + quoteIdentifier(function) + "()", + "DROP TABLE IF EXISTS " + qualifiedSchema + "." + quoteIdentifier(child), + "DROP TABLE IF EXISTS " + qualifiedSchema + "." + quoteIdentifier(parent), } t.Cleanup(func() { for _, statement := range cleanup { @@ -53,60 +58,60 @@ func TestKingbaseIntegration(t *testing.T) { } }) - mustExecute(t, server, "CREATE TABLE public."+quoteIdentifier(parent)+" (id integer PRIMARY KEY, name varchar(64) NOT NULL)") - mustExecute(t, server, "COMMENT ON TABLE public."+quoteIdentifier(parent)+" IS '订单父表'") - mustExecute(t, server, "COMMENT ON COLUMN public."+quoteIdentifier(parent)+".id IS '主键编号'") - mustExecute(t, server, "COMMENT ON COLUMN public."+quoteIdentifier(parent)+".name IS '客户''名称'") - mustExecute(t, server, "CREATE TABLE public."+quoteIdentifier(child)+" (id integer PRIMARY KEY, parent_id integer REFERENCES public."+quoteIdentifier(parent)+"(id))") - mustExecute(t, server, "CREATE INDEX "+quoteIdentifier(child+"_parent_idx")+" ON public."+quoteIdentifier(child)+"(parent_id)") - mustExecute(t, server, "CREATE VIEW public."+quoteIdentifier(view)+" AS SELECT id, name FROM public."+quoteIdentifier(parent)) - mustExecute(t, server, "CREATE FUNCTION public."+quoteIdentifier(function)+"() RETURNS text AS $$ SELECT 'dbx'; $$ LANGUAGE SQL") + mustExecute(t, server, "CREATE TABLE "+qualifiedSchema+"."+quoteIdentifier(parent)+" (id integer PRIMARY KEY, name varchar(64) NOT NULL)") + mustExecute(t, server, "COMMENT ON TABLE "+qualifiedSchema+"."+quoteIdentifier(parent)+" IS '订单父表'") + mustExecute(t, server, "COMMENT ON COLUMN "+qualifiedSchema+"."+quoteIdentifier(parent)+".id IS '主键编号'") + mustExecute(t, server, "COMMENT ON COLUMN "+qualifiedSchema+"."+quoteIdentifier(parent)+".name IS '客户''名称'") + mustExecute(t, server, "CREATE TABLE "+qualifiedSchema+"."+quoteIdentifier(child)+" (id integer PRIMARY KEY, parent_id integer REFERENCES "+qualifiedSchema+"."+quoteIdentifier(parent)+"(id))") + mustExecute(t, server, "CREATE INDEX "+quoteIdentifier(child+"_parent_idx")+" ON "+qualifiedSchema+"."+quoteIdentifier(child)+"(parent_id)") + mustExecute(t, server, "CREATE VIEW "+qualifiedSchema+"."+quoteIdentifier(view)+" AS SELECT id, name FROM "+qualifiedSchema+"."+quoteIdentifier(parent)) + mustExecute(t, server, "CREATE FUNCTION "+qualifiedSchema+"."+quoteIdentifier(function)+"() RETURNS text AS $$ SELECT 'dbx'; $$ LANGUAGE SQL") - tables, err := server.listTables("public", metadataListConstraints{Filter: suffix}) + tables, err := server.listTables(schema, metadataListConstraints{Filter: suffix}) if err != nil || len(tables) < 3 { t.Fatalf("list tables failed: count=%d err=%v", len(tables), err) } - columns, err := server.getColumns("public", child) + columns, err := server.getColumns(schema, child) if err != nil || len(columns) != 2 || !columns[0].IsPrimaryKey { t.Fatalf("get columns failed: columns=%v err=%v", columns, err) } - parentColumns, err := server.getColumns("public", parent) + parentColumns, err := server.getColumns(schema, parent) if err != nil || len(parentColumns) != 2 || parentColumns[0].Comment == nil || *parentColumns[0].Comment != "主键编号" || parentColumns[1].Comment == nil || *parentColumns[1].Comment != "客户'名称" { t.Fatalf("get commented columns failed: columns=%v err=%v", parentColumns, err) } - ddl, err := server.getTableDDL("public", parent) + ddl, err := server.getTableDDL(schema, parent) if err != nil { t.Fatalf("get table DDL failed: %v", err) } - qualifiedParent := quoteIdentifier("public") + "." + quoteIdentifier(parent) + qualifiedParent := qualifiedSchema + "." + quoteIdentifier(parent) for _, expected := range []string{ "COMMENT ON TABLE " + qualifiedParent + " IS '订单父表';", - "COMMENT ON COLUMN " + qualifiedParent + "." + quoteIdentifier("id") + " IS '主键编号';", - "COMMENT ON COLUMN " + qualifiedParent + "." + quoteIdentifier("name") + " IS '客户''名称';", + "COMMENT ON COLUMN " + qualifiedParent + "." + quoteIdentifier(parentColumns[0].Name) + " IS '主键编号';", + "COMMENT ON COLUMN " + qualifiedParent + "." + quoteIdentifier(parentColumns[1].Name) + " IS '客户''名称';", } { if !strings.Contains(ddl, expected) { t.Fatalf("table DDL missing %q:\n%s", expected, ddl) } } - indexes, err := server.listIndexes("public", child) + indexes, err := server.listIndexes(schema, child) if err != nil || len(indexes) < 2 { t.Fatalf("list indexes failed: indexes=%v err=%v", indexes, err) } - foreignKeys, err := server.listForeignKeys("public", child) + foreignKeys, err := server.listForeignKeys(schema, child) if err != nil || len(foreignKeys) != 1 || foreignKeys[0].RefTable != parent { t.Fatalf("list foreign keys failed: keys=%v err=%v", foreignKeys, err) } - source, err := server.getObjectSource("public", function, "FUNCTION") + source, err := server.getObjectSource(schema, function, "FUNCTION") if err != nil || !strings.Contains(fmt.Sprint(source["source"]), function) { t.Fatalf("get function source failed: source=%v err=%v", source, err) } - viewSource, err := server.getObjectSource("public", view, "VIEW") + viewSource, err := server.getObjectSource(schema, view, "VIEW") if err != nil || !strings.Contains(fmt.Sprint(viewSource["source"]), parent) { t.Fatalf("get view source failed: source=%v err=%v", viewSource, err) } transactionParams := map[string]json.RawMessage{ - "schema": rawJSON("public"), + "schema": rawJSON(schema), "statements": rawJSON([]string{"INSERT INTO " + quoteIdentifier(parent) + " VALUES (1, 'one')", "INSERT INTO " + quoteIdentifier(child) + " VALUES (1, 1)"}), } if _, err := server.executeTransaction(transactionParams); err != nil { diff --git a/agents/drivers/kingbase-go/kingbase_metadata.go b/agents/drivers/kingbase-go/kingbase_metadata.go index 129b221b15..ef18bb8e57 100644 --- a/agents/drivers/kingbase-go/kingbase_metadata.go +++ b/agents/drivers/kingbase-go/kingbase_metadata.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "regexp" "sort" "strconv" "strings" @@ -37,6 +38,22 @@ type kingbaseMode struct { postgresCatalog bool mysqlCompat bool sqlServerIdentity bool + legacyV7 bool +} + +var kingbaseReleasePattern = regexp.MustCompile(`(?i)\bV0*([0-9]+)R`) + +func detectKingbaseV7(db *sql.DB) bool { + var version string + if err := db.QueryRow("SELECT version()").Scan(&version); err != nil { + return false + } + match := kingbaseReleasePattern.FindStringSubmatch(version) + if len(match) != 2 { + return false + } + major, err := strconv.Atoi(match[1]) + return err == nil && major == 7 } type databaseInfo struct { @@ -1399,8 +1416,14 @@ func (s *server) listIndexes(schema, table string) ([]indexInfo, error) { if s.mode.postgresCatalog { catalog, prefix = "pg_catalog", "pg" } - query := kingbaseListIndexesQuery(catalog, prefix, effective, table) - rows, err := s.metadataQuery(query) + if s.indexOrdinalityUnsupported { + return s.listIndexesWithoutOrdinality(catalog, prefix, effective, table) + } + rows, err := s.metadataQuery(kingbaseListIndexesQuery(catalog, prefix, effective, table)) + if err != nil && isUnsupportedWithOrdinality(err) { + s.indexOrdinalityUnsupported = true + return s.listIndexesWithoutOrdinality(catalog, prefix, effective, table) + } if err != nil { return nil, err } @@ -1429,6 +1452,118 @@ func (s *server) listIndexes(schema, table string) ([]indexInfo, error) { return result, rows.Err() } +func isUnsupportedWithOrdinality(err error) bool { + if err == nil { + return false + } + var driverError *gokb.Error + isSyntaxError := errors.As(err, &driverError) && string(driverError.Code) == "42601" + normalized := strings.ToLower(err.Error()) + return (isSyntaxError || strings.Contains(normalized, "syntax error") || strings.Contains(normalized, "语法错误")) && strings.Contains(normalized, "with ordinality") +} + +func (s *server) listIndexesWithoutOrdinality(catalog, prefix, schema, table string) ([]indexInfo, error) { + query := fmt.Sprintf(`SELECT i.relname, am.amname, ix.indisunique, ix.indisprimary, ix.indkey +FROM %s.%s_index ix JOIN %s.%s_class t ON t.oid = ix.indrelid +JOIN %s.%s_class i ON i.oid = ix.indexrelid JOIN %s.%s_namespace n ON n.oid = t.relnamespace +JOIN %s.%s_am am ON am.oid = i.relam +WHERE n.nspname = %s AND t.relname = %s ORDER BY i.relname`, catalog, prefix, catalog, prefix, catalog, prefix, catalog, prefix, catalog, prefix, quoteLiteral(schema), quoteLiteral(table)) + rows, err := s.metadataQuery(query) + if err != nil { + return nil, err + } + defer rows.Close() + type rawIndex struct { + name, kind string + unique, primary bool + attributeNums []int + } + rawIndexes := []rawIndex{} + for rows.Next() { + var item rawIndex + var raw any + if err := rows.Scan(&item.name, &item.kind, &item.unique, &item.primary, &raw); err != nil { + return nil, err + } + item.attributeNums, err = parseCatalogAttributeNumbers(raw) + if err != nil { + return nil, fmt.Errorf("failed to parse index %s columns: %w", item.name, err) + } + rawIndexes = append(rawIndexes, item) + } + if err := rows.Err(); err != nil { + return nil, err + } + if len(rawIndexes) == 0 { + return []indexInfo{}, nil + } + attributes, err := s.relationAttributesByNumber(catalog, prefix, schema, table) + if err != nil { + return nil, err + } + result := make([]indexInfo, 0, len(rawIndexes)) + for _, raw := range rawIndexes { + item := indexInfo{Name: raw.name, IsUnique: raw.unique, IsPrimary: raw.primary, IndexType: stringPtr(raw.kind), Columns: []string{}, IncludedColumns: []string{}} + for _, number := range raw.attributeNums { + if name := attributes[number]; name != "" { + item.Columns = append(item.Columns, name) + } + } + result = append(result, item) + } + return result, nil +} + +func parseCatalogAttributeNumbers(raw any) ([]int, error) { + var value string + switch typed := raw.(type) { + case nil: + return []int{}, nil + case string: + value = typed + case []byte: + value = string(typed) + default: + value = fmt.Sprint(typed) + } + value = strings.TrimSpace(strings.Trim(value, "{}")) + if value == "" { + return []int{}, nil + } + parts := strings.FieldsFunc(value, func(r rune) bool { return r == ',' || r == ' ' || r == '\t' }) + result := make([]int, 0, len(parts)) + for _, part := range parts { + number, err := strconv.Atoi(strings.TrimSpace(part)) + if err != nil { + return nil, fmt.Errorf("invalid attribute number %q", part) + } + result = append(result, number) + } + return result, nil +} + +func (s *server) relationAttributesByNumber(catalog, prefix, schema, table string) (map[int]string, error) { + query := fmt.Sprintf(`SELECT a.attnum, a.attname +FROM %s.%s_attribute a JOIN %s.%s_class c ON c.oid = a.attrelid +JOIN %s.%s_namespace n ON n.oid = c.relnamespace +WHERE n.nspname = %s AND c.relname = %s AND a.attnum > 0 AND NOT a.attisdropped`, catalog, prefix, catalog, prefix, catalog, prefix, quoteLiteral(schema), quoteLiteral(table)) + rows, err := s.metadataQuery(query) + if err != nil { + return nil, err + } + defer rows.Close() + result := map[int]string{} + for rows.Next() { + var number int + var name string + if err := rows.Scan(&number, &name); err != nil { + return nil, err + } + result[number] = name + } + return result, rows.Err() +} + func kingbaseListIndexesQuery(catalog, prefix, schema, table string) string { return fmt.Sprintf(`SELECT i.relname, am.amname, ix.indisunique, ix.indisprimary, a.attname, pos.n FROM %s.%s_index ix JOIN %s.%s_class t ON t.oid = ix.indrelid @@ -1452,6 +1587,9 @@ func (s *server) listForeignKeys(schema, table string) ([]foreignKeyInfo, error) if err != nil { return nil, err } + if s.mode.legacyV7 && !s.mode.mysqlCompat { + return s.listForeignKeysFromCatalog(effective, table) + } query := `SELECT fk.constraint_name, fk.column_name, pk.table_name, pk.column_name FROM information_schema.table_constraints tc JOIN information_schema.key_column_usage fk ON fk.constraint_schema = tc.constraint_schema AND fk.constraint_name = tc.constraint_name AND fk.table_schema = tc.table_schema AND fk.table_name = tc.table_name @@ -1471,7 +1609,82 @@ WHERE tc.table_schema = ` + quoteLiteral(effective) + ` AND tc.table_name = ` + } result = append(result, item) } - return result, rows.Err() + if err := rows.Err(); err != nil { + return nil, err + } + return result, nil +} + +func (s *server) listForeignKeysFromCatalog(schema, table string) ([]foreignKeyInfo, error) { + catalog, prefix := "sys_catalog", "sys" + if s.mode.postgresCatalog { + catalog, prefix = "pg_catalog", "pg" + } + query := fmt.Sprintf(`SELECT c.conname, c.conkey, c.confkey, rn.nspname, rt.relname +FROM %s.%s_constraint c JOIN %s.%s_class t ON t.oid = c.conrelid +JOIN %s.%s_namespace n ON n.oid = t.relnamespace +JOIN %s.%s_class rt ON rt.oid = c.confrelid +JOIN %s.%s_namespace rn ON rn.oid = rt.relnamespace +WHERE c.contype = 'f' AND n.nspname = %s AND t.relname = %s ORDER BY c.conname`, catalog, prefix, catalog, prefix, catalog, prefix, catalog, prefix, catalog, prefix, quoteLiteral(schema), quoteLiteral(table)) + rows, err := s.metadataQuery(query) + if err != nil { + return nil, err + } + defer rows.Close() + type rawForeignKey struct { + name, refSchema, refTable string + columns, refColumns []int + } + rawKeys := []rawForeignKey{} + for rows.Next() { + var item rawForeignKey + var columnsRaw, refColumnsRaw any + if err := rows.Scan(&item.name, &columnsRaw, &refColumnsRaw, &item.refSchema, &item.refTable); err != nil { + return nil, err + } + item.columns, err = parseCatalogAttributeNumbers(columnsRaw) + if err != nil { + return nil, fmt.Errorf("failed to parse foreign key %s columns: %w", item.name, err) + } + item.refColumns, err = parseCatalogAttributeNumbers(refColumnsRaw) + if err != nil { + return nil, fmt.Errorf("failed to parse foreign key %s referenced columns: %w", item.name, err) + } + rawKeys = append(rawKeys, item) + } + if err := rows.Err(); err != nil { + return nil, err + } + if len(rawKeys) == 0 { + return []foreignKeyInfo{}, nil + } + localAttributes, err := s.relationAttributesByNumber(catalog, prefix, schema, table) + if err != nil { + return nil, err + } + refAttributes := map[string]map[int]string{} + result := []foreignKeyInfo{} + for _, raw := range rawKeys { + key := raw.refSchema + "\x00" + raw.refTable + attributes := refAttributes[key] + if attributes == nil { + attributes, err = s.relationAttributesByNumber(catalog, prefix, raw.refSchema, raw.refTable) + if err != nil { + return nil, err + } + refAttributes[key] = attributes + } + for i, number := range raw.columns { + if i >= len(raw.refColumns) { + break + } + column, refColumn := localAttributes[number], attributes[raw.refColumns[i]] + if column != "" && refColumn != "" { + result = append(result, foreignKeyInfo{Name: raw.name, Column: column, RefTable: raw.refTable, RefColumn: refColumn}) + } + } + } + return result, nil } func (s *server) listTriggers(schema, table string) ([]triggerInfo, error) { @@ -1483,11 +1696,21 @@ func (s *server) listTriggers(schema, table string) ([]triggerInfo, error) { if s.mode.postgresCatalog { catalog, prefix = "pg_catalog", "pg" } - query := fmt.Sprintf(`SELECT tg.tgname, + queryForCatalog := func() string { + internalPredicate := "NOT tg.tgisinternal" + if s.triggerInternalUnsupported { + internalPredicate = "tg.tgkind <> 'c'" + } + return fmt.Sprintf(`SELECT tg.tgname, trim(trailing ',' FROM (CASE WHEN (tg.tgtype & 4) <> 0 THEN 'INSERT,' ELSE '' END || CASE WHEN (tg.tgtype & 8) <> 0 THEN 'DELETE,' ELSE '' END || CASE WHEN (tg.tgtype & 16) <> 0 THEN 'UPDATE,' ELSE '' END || CASE WHEN (tg.tgtype & 32) <> 0 THEN 'TRUNCATE,' ELSE '' END)), tg.tgtype FROM %s.%s_trigger tg JOIN %s.%s_class c ON c.oid = tg.tgrelid JOIN %s.%s_namespace n ON n.oid = c.relnamespace -WHERE n.nspname = %s AND c.relname = %s AND NOT tg.tgisinternal ORDER BY tg.tgname`, catalog, prefix, catalog, prefix, catalog, prefix, quoteLiteral(effective), quoteLiteral(table)) - rows, err := s.metadataQuery(query) +WHERE n.nspname = %s AND c.relname = %s AND %s ORDER BY tg.tgname`, catalog, prefix, catalog, prefix, catalog, prefix, quoteLiteral(effective), quoteLiteral(table), internalPredicate) + } + rows, err := s.metadataQuery(queryForCatalog()) + if err != nil && !s.triggerInternalUnsupported && isUndefinedColumn(err, "tgisinternal") { + s.triggerInternalUnsupported = true + rows, err = s.metadataQuery(queryForCatalog()) + } if err != nil { return nil, err } @@ -1542,10 +1765,27 @@ func (s *server) getObjectSource(schema, name, objectType string) (map[string]an query := fmt.Sprintf("SELECT %s(p.oid) FROM %s.%s_proc p JOIN %s.%s_namespace n ON n.oid=p.pronamespace WHERE n.nspname=%s AND p.proname=%s ORDER BY CASE WHEN p.prorettype=2278 THEN 0 ELSE 1 END LIMIT 1", definitionFunction, catalog, prefix, catalog, prefix, quoteLiteral(effective), quoteLiteral(name)) return s.requireDBQueryRow(query, &source) } - err = querySource(function) - if err != nil && function == "sys_get_functiondef" && isUndefinedFunction(err, function) { - s.usePgFunctionDefinition = true - err = querySource("pg_get_functiondef") + queryLegacySource := func() error { + legacyFunction := "GET_FUNC_DDL" + if kind == "PROCEDURE" { + legacyFunction = "GET_PROCEDURE_DDL" + } + query := fmt.Sprintf("SELECT DBMS_METADATA.%s(CAST(%s AS varchar(128)), CAST(%s AS varchar(128)))", legacyFunction, quoteLiteral(name), quoteLiteral(effective)) + return s.requireDBQueryRow(query, &source) + } + if s.useLegacyRoutineDefinition { + err = queryLegacySource() + } else { + err = querySource(function) + if err != nil && function == "sys_get_functiondef" && isUndefinedFunction(err, function) { + s.usePgFunctionDefinition = true + function = "pg_get_functiondef" + err = querySource(function) + } + if err != nil && !s.mode.postgresCatalog && function == "pg_get_functiondef" && isUndefinedFunction(err, function) { + s.useLegacyRoutineDefinition = true + err = queryLegacySource() + } } } if err != nil && err != sql.ErrNoRows { @@ -1665,10 +1905,36 @@ func (s *server) listTriggerDefinitions(schema, table string) ([]string, error) catalog, prefix = "pg_catalog", "pg" } triggerDefinitionFunction := kingbaseCatalogFunction(catalog, "sys_get_triggerdef", "pg_get_triggerdef") - query := fmt.Sprintf(`SELECT %s(tg.oid, true) + queryForSignature := func(includePretty bool) string { + arguments := "tg.oid" + if includePretty { + arguments += ", true" + } + internalPredicate := "NOT tg.tgisinternal" + if s.triggerInternalUnsupported { + internalPredicate = "tg.tgkind <> 'c'" + } + return fmt.Sprintf(`SELECT %s(%s) FROM %s.%s_trigger tg JOIN %s.%s_class c ON c.oid = tg.tgrelid JOIN %s.%s_namespace n ON n.oid = c.relnamespace -WHERE n.nspname = %s AND c.relname = %s AND NOT tg.tgisinternal ORDER BY tg.tgname`, triggerDefinitionFunction, catalog, prefix, catalog, prefix, catalog, prefix, quoteLiteral(effective), quoteLiteral(table)) - rows, err := s.metadataQuery(query) +WHERE n.nspname = %s AND c.relname = %s AND %s ORDER BY tg.tgname`, triggerDefinitionFunction, arguments, catalog, prefix, catalog, prefix, catalog, prefix, quoteLiteral(effective), quoteLiteral(table), internalPredicate) + } + var rows *sql.Rows + for { + includePretty := !s.triggerPrettyUnsupported + rows, err = s.metadataQuery(queryForSignature(includePretty)) + if err == nil { + break + } + if includePretty && isUndefinedFunction(err, triggerDefinitionFunction) { + s.triggerPrettyUnsupported = true + continue + } + if !s.triggerInternalUnsupported && isUndefinedColumn(err, "tgisinternal") { + s.triggerInternalUnsupported = true + continue + } + break + } if err != nil { return nil, err } diff --git a/agents/drivers/kingbase-go/main.go b/agents/drivers/kingbase-go/main.go index bab75a90fe..b4fcf1c3ee 100644 --- a/agents/drivers/kingbase-go/main.go +++ b/agents/drivers/kingbase-go/main.go @@ -136,10 +136,14 @@ type server struct { usePgDefaultExpression bool usePgViewDefinition bool usePgFunctionDefinition bool + useLegacyRoutineDefinition bool catalogIdentityUnsupported bool catalogOIDUnsupported bool infoColumnTypeUnsupported bool infoUdtNameUnsupported bool + indexOrdinalityUnsupported bool + triggerPrettyUnsupported bool + triggerInternalUnsupported bool currentSchema string schemaSet bool sessions map[string]*querySession @@ -461,13 +465,18 @@ func (s *server) connect(cp connectParams) error { s.db = db s.params = cp s.mode = detectKingbaseMode(db, cp.MySQLCompatMode) + s.mode.legacyV7 = detectKingbaseV7(db) s.usePgDefaultExpression = false s.usePgViewDefinition = false s.usePgFunctionDefinition = false + s.useLegacyRoutineDefinition = false s.catalogIdentityUnsupported = false s.catalogOIDUnsupported = false s.infoColumnTypeUnsupported = false s.infoUdtNameUnsupported = false + s.indexOrdinalityUnsupported = false + s.triggerPrettyUnsupported = false + s.triggerInternalUnsupported = false return nil } @@ -531,10 +540,14 @@ func (s *server) disconnect() error { s.usePgDefaultExpression = false s.usePgViewDefinition = false s.usePgFunctionDefinition = false + s.useLegacyRoutineDefinition = false s.catalogIdentityUnsupported = false s.catalogOIDUnsupported = false s.infoColumnTypeUnsupported = false s.infoUdtNameUnsupported = false + s.indexOrdinalityUnsupported = false + s.triggerPrettyUnsupported = false + s.triggerInternalUnsupported = false s.currentSchema = "" s.schemaSet = false if s.db == nil { diff --git a/agents/drivers/kingbase-go/main_test.go b/agents/drivers/kingbase-go/main_test.go index 7b92d82314..9ae76e527d 100644 --- a/agents/drivers/kingbase-go/main_test.go +++ b/agents/drivers/kingbase-go/main_test.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "io" + "strconv" "strings" "sync" "sync/atomic" @@ -933,6 +934,173 @@ func TestKingbaseListIndexesQuerySupportsSQLServerMode(t *testing.T) { } } +func TestParseCatalogAttributeNumbers(t *testing.T) { + tests := []struct { + name string + raw any + expected string + wantErr bool + }{ + {name: "int2vector string", raw: "1 2 4", expected: "1,2,4"}, + {name: "array string", raw: "{3,5}", expected: "3,5"}, + {name: "bytes", raw: []byte("6 7"), expected: "6,7"}, + {name: "empty", raw: nil, expected: ""}, + {name: "invalid", raw: "1 bad", wantErr: true}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + numbers, err := parseCatalogAttributeNumbers(test.raw) + if test.wantErr { + if err == nil { + t.Fatalf("expected parse error, got %v", numbers) + } + return + } + if err != nil { + t.Fatal(err) + } + parts := make([]string, len(numbers)) + for index, number := range numbers { + parts[index] = strconv.Itoa(number) + } + if actual := strings.Join(parts, ","); actual != test.expected { + t.Fatalf("unexpected numbers: %q", actual) + } + }) + } +} + +func TestListIndexesFallsBackWhenWithOrdinalityIsUnsupported(t *testing.T) { + state := &metadataDriverState{query: func(query string) (driver.Rows, error) { + switch { + case strings.Contains(query, "WITH ORDINALITY"): + return nil, &gokb.Error{Code: gokb.ErrorCode("42601"), Message: `syntax error at or near "WITH ORDINALITY"`} + case strings.Contains(query, "SELECT i.relname, am.amname") && strings.Contains(query, "ix.indkey"): + return &valueRows{ + columns: []string{"relname", "amname", "indisunique", "indisprimary", "indkey"}, + rows: [][]driver.Value{ + {"orders_customer_idx", "btree", false, false, "2 3"}, + {"orders_pkey", "btree", true, true, []byte("1")}, + }, + }, nil + case strings.Contains(query, "SELECT a.attnum, a.attname"): + return &valueRows{ + columns: []string{"attnum", "attname"}, + rows: [][]driver.Value{{int64(1), "id"}, {int64(2), "customer_id"}, {int64(3), "created_at"}}, + }, nil + default: + return nil, errors.New("unexpected query: " + query) + } + }} + server := newServer() + server.db = openMetadataDB(t, state) + + for range 2 { + indexes, err := server.listIndexes("PUBLIC", "orders") + if err != nil { + t.Fatal(err) + } + if len(indexes) != 2 || strings.Join(indexes[0].Columns, ",") != "customer_id,created_at" || strings.Join(indexes[1].Columns, ",") != "id" { + t.Fatalf("unexpected indexes: %#v", indexes) + } + } + + var ordinalityQueries int + for _, query := range state.snapshotQueries() { + if strings.Contains(query, "WITH ORDINALITY") { + ordinalityQueries++ + } + } + if ordinalityQueries != 1 || !server.indexOrdinalityUnsupported { + t.Fatalf("unsupported capability was not cached: queries=%v", state.snapshotQueries()) + } +} + +func TestListIndexesDoesNotFallbackForUnrelatedErrors(t *testing.T) { + expectedErr := errors.New("metadata connection reset") + state := &metadataDriverState{query: func(string) (driver.Rows, error) { return nil, expectedErr }} + server := newServer() + server.db = openMetadataDB(t, state) + + if _, err := server.listIndexes("PUBLIC", "orders"); !errors.Is(err, expectedErr) { + t.Fatalf("unexpected error: %v", err) + } + if queries := state.snapshotQueries(); len(queries) != 1 || server.indexOrdinalityUnsupported { + t.Fatalf("unrelated error triggered fallback: %v", queries) + } +} + +func TestListForeignKeysUsesCatalogForV7(t *testing.T) { + state := &metadataDriverState{query: func(query string) (driver.Rows, error) { + switch { + case strings.Contains(query, "FROM information_schema.table_constraints tc"): + return nil, errors.New("V7 must not query information_schema foreign keys") + case strings.Contains(query, "FROM sys_catalog.sys_constraint c"): + return &valueRows{ + columns: []string{"conname", "conkey", "confkey", "nspname", "relname"}, + rows: [][]driver.Value{{"orders_customer_fkey", "2 3", "1 2", "PUBLIC", "customers"}}, + }, nil + case strings.Contains(query, "SELECT a.attnum, a.attname") && strings.Contains(query, "c.relname = 'orders'"): + return &valueRows{columns: []string{"attnum", "attname"}, rows: [][]driver.Value{{int64(2), "customer_id"}, {int64(3), "customer_region"}}}, nil + case strings.Contains(query, "SELECT a.attnum, a.attname") && strings.Contains(query, "c.relname = 'customers'"): + return &valueRows{columns: []string{"attnum", "attname"}, rows: [][]driver.Value{{int64(1), "id"}, {int64(2), "region"}}}, nil + default: + return nil, errors.New("unexpected query: " + query) + } + }} + server := newServer() + server.db = openMetadataDB(t, state) + server.mode.legacyV7 = true + + keys, err := server.listForeignKeys("PUBLIC", "orders") + if err != nil { + t.Fatal(err) + } + if len(keys) != 2 || keys[0].Column != "customer_id" || keys[0].RefColumn != "id" || keys[1].Column != "customer_region" || keys[1].RefColumn != "region" { + t.Fatalf("unexpected foreign keys: %#v", keys) + } +} + +func TestListForeignKeysKeepsEmptyInformationSchemaResultOnV8(t *testing.T) { + state := &metadataDriverState{query: func(query string) (driver.Rows, error) { + if strings.Contains(query, "FROM information_schema.table_constraints tc") { + return &valueRows{columns: []string{"constraint_name", "column_name", "table_name", "column_name"}}, nil + } + return nil, errors.New("V8 empty result must not trigger a catalog query: " + query) + }} + server := newServer() + server.db = openMetadataDB(t, state) + + keys, err := server.listForeignKeys("PUBLIC", "orders") + if err != nil { + t.Fatal(err) + } + if len(keys) != 0 || len(state.snapshotQueries()) != 1 { + t.Fatalf("unexpected foreign keys or query count: keys=%#v queries=%v", keys, state.snapshotQueries()) + } +} + +func TestKingbaseV7VersionPattern(t *testing.T) { + for _, test := range []struct { + version string + v7 bool + }{ + {version: "Kingbase V007R001C002B0014", v7: true}, + {version: "KingbaseES V008R006C008B0014"}, + {version: "PostgreSQL 12.1"}, + } { + match := kingbaseReleasePattern.FindStringSubmatch(test.version) + actual := false + if len(match) == 2 { + major, err := strconv.Atoi(match[1]) + actual = err == nil && major == 7 + } + if actual != test.v7 { + t.Fatalf("version=%q: expected v7=%v, got %v", test.version, test.v7, actual) + } + } +} + func TestKingbaseCatalogFunctionsFollowMetadataMode(t *testing.T) { registerExpressionFallbackDriver.Do(func() { sql.Register("kingbase-expression-fallback-test", fallbackDriver{}) }) tests := []struct { @@ -975,6 +1143,87 @@ func TestKingbaseCatalogFunctionsFollowMetadataMode(t *testing.T) { } } +func TestListTriggerDefinitionsFallsBackToSingleArgument(t *testing.T) { + state := &metadataDriverState{query: func(query string) (driver.Rows, error) { + if strings.Contains(query, "sys_get_triggerdef(tg.oid, true)") { + return nil, &gokb.Error{Code: gokb.ErrorCode("42883"), Message: "function SYS_CATALOG.SYS_GET_TRIGGERDEF(OID, BOOLEAN) does not exist"} + } + if strings.Contains(query, "tg.tgisinternal") { + return nil, &gokb.Error{Code: gokb.ErrorCode("42703"), Message: "column TG.TGISINTERNAL does not exist"} + } + if strings.Contains(query, "sys_get_triggerdef(tg.oid)") { + if !strings.Contains(query, "tg.tgkind <> 'c'") { + return nil, errors.New("V7 trigger query did not exclude constraint triggers: " + query) + } + return &valueRows{columns: []string{"definition"}, rows: [][]driver.Value{{"CREATE TRIGGER orders_audit ..."}}}, nil + } + return nil, errors.New("unexpected query: " + query) + }} + server := newServer() + server.db = openMetadataDB(t, state) + + for range 2 { + definitions, err := server.listTriggerDefinitions("PUBLIC", "orders") + if err != nil { + t.Fatal(err) + } + if len(definitions) != 1 || definitions[0] != "CREATE TRIGGER orders_audit ..." { + t.Fatalf("unexpected definitions: %#v", definitions) + } + } + + var prettyQueries int + for _, query := range state.snapshotQueries() { + if strings.Contains(query, "tg.oid, true") { + prettyQueries++ + } + } + if prettyQueries != 1 || !server.triggerPrettyUnsupported || !server.triggerInternalUnsupported { + t.Fatalf("unsupported signature was not cached: %v", state.snapshotQueries()) + } +} + +func TestListTriggersFallsBackToV7InternalPredicate(t *testing.T) { + state := &metadataDriverState{query: func(query string) (driver.Rows, error) { + if strings.Contains(query, "tg.tgisinternal") { + return nil, &gokb.Error{Code: gokb.ErrorCode("42703"), Message: "column TG.TGISINTERNAL does not exist"} + } + if strings.Contains(query, "tg.tgkind <> 'c'") { + return &valueRows{columns: []string{"tgname", "events", "tgtype"}, rows: [][]driver.Value{{"orders_audit", "INSERT", int64(7)}}}, nil + } + return nil, errors.New("unexpected query: " + query) + }} + server := newServer() + server.db = openMetadataDB(t, state) + + for range 2 { + triggers, err := server.listTriggers("PUBLIC", "orders") + if err != nil { + t.Fatal(err) + } + if len(triggers) != 1 || triggers[0].Name != "orders_audit" { + t.Fatalf("unexpected triggers: %#v", triggers) + } + } + if queries := state.snapshotQueries(); len(queries) != 3 || !server.triggerInternalUnsupported { + t.Fatalf("unsupported column was not cached: %v", queries) + } +} + +func TestListTriggerDefinitionsDoesNotFallbackForUnrelatedErrors(t *testing.T) { + expectedErr := errors.New("permission denied for sys_trigger") + state := &metadataDriverState{query: func(string) (driver.Rows, error) { return nil, expectedErr }} + server := newServer() + server.db = openMetadataDB(t, state) + + if _, err := server.listTriggerDefinitions("PUBLIC", "orders"); !errors.Is(err, expectedErr) { + t.Fatalf("unexpected error: %v", err) + } + if queries := state.snapshotQueries(); len(queries) != 1 || server.triggerPrettyUnsupported || server.triggerInternalUnsupported { + t.Fatalf("unrelated error triggered fallback: %v", queries) + } +} + func TestMySQLCompatSchemaQueryKeepsUserSchemasWithSystemLikeNames(t *testing.T) { query := kingbaseMySQLCompatListSchemasSQL for _, prefix := range []string{"SYS", "XLOG"} { @@ -2050,6 +2299,41 @@ func TestObjectSourceDoesNotFallbackOnUnrelatedErrors(t *testing.T) { } } +func TestObjectSourceFallsBackToV7RoutineDDLAndCachesChoice(t *testing.T) { + state := &metadataDriverState{query: func(query string) (driver.Rows, error) { + switch { + case strings.Contains(query, "sys_get_functiondef("): + return nil, &gokb.Error{Code: gokb.ErrorCode("42883"), Message: "function SYS_GET_FUNCTIONDEF(OID) does not exist"} + case strings.Contains(query, "pg_get_functiondef("): + return nil, &gokb.Error{Code: gokb.ErrorCode("42883"), Message: "function PG_GET_FUNCTIONDEF(OID) does not exist"} + case strings.Contains(query, "DBMS_METADATA.GET_FUNC_DDL"): + if !strings.Contains(query, "CAST('format_name' AS varchar(128))") || !strings.Contains(query, "CAST('PUBLIC' AS varchar(128))") { + return nil, errors.New("legacy DDL query lost routine identity: " + query) + } + return &valueRows{columns: []string{"ddl"}, rows: [][]driver.Value{{"CREATE OR REPLACE FUNCTION PUBLIC.format_name() RETURN TEXT AS ..."}}}, nil + default: + return nil, errors.New("unexpected query: " + query) + } + }} + server := newServer() + server.db = openMetadataDB(t, state) + + for range 2 { + source, err := server.getObjectSource("PUBLIC", "format_name", "FUNCTION") + if err != nil { + t.Fatal(err) + } + if !strings.Contains(fmt.Sprint(source["source"]), "CREATE OR REPLACE FUNCTION") { + t.Fatalf("unexpected source: %#v", source) + } + } + + queries := state.snapshotQueries() + if len(queries) != 4 || !strings.Contains(queries[2], "DBMS_METADATA.GET_FUNC_DDL") || !strings.Contains(queries[3], "DBMS_METADATA.GET_FUNC_DDL") || !server.useLegacyRoutineDefinition { + t.Fatalf("V7 routine DDL choice was not cached: %v", queries) + } +} + func TestColumnsFallbackToPgGetExprAndCacheChoice(t *testing.T) { registerExpressionFallbackDriver.Do(func() { sql.Register("kingbase-expression-fallback-test", fallbackDriver{}) }) state := &fallbackDriverState{}