mirror of
https://github.com/t8y2/dbx.git
synced 2026-10-21 06:00:39 +00:00
fix(kingbase): support V7 metadata catalogs
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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{}
|
||||
|
||||
Reference in New Issue
Block a user