fix(kingbase): avoid privileged table metadata query

agents/drivers/kingbase-go/kingbase_metadata.go

@@ -246,6 +246,11 @@ LEFT JOIN %s.%s_description d ON d.objoid = c.oid AND d.objsubid = 0
 WHERE n.nspname = %s AND c.relkind IN ('r','p','v','m','f') ORDER BY c.relname`, catalog, catalogPrefix(catalog), catalog, catalogPrefix(catalog), catalog, catalogPrefix(catalog), quoteLiteral(effective))
 	}
 	rows, err := s.metadataQuery(query)
+	if err != nil && s.mode.mysqlCompat && isSysFreespacePermissionError(err) {
+		// Kingbase's information_schema.tables calls sys_freespace internally;
+		// restricted users need the catalog query that avoids that privileged function.
+		rows, err = s.metadataQuery(kingbaseMySQLCompatCatalogTablesQuery(effective))
+	}
 	if err != nil {
 		return nil, err
 	}
@@ -265,6 +270,31 @@ WHERE n.nspname = %s AND c.relkind IN ('r','p','v','m','f') ORDER BY c.relname`,
 	return pageTables(result, constraints), rows.Err()
 }
 
+func kingbaseMySQLCompatCatalogTablesQuery(schema string) string {
+	return fmt.Sprintf(`SELECT c.relname,
+CASE WHEN CAST(c.relkind AS varchar(16)) IN ('r', 'p') THEN 'TABLE' ELSE 'VIEW' END,
+d.description
+FROM sys_catalog.sys_class c
+JOIN sys_catalog.sys_namespace n ON n.oid = c.relnamespace
+LEFT JOIN sys_catalog.sys_description d ON CAST(d.objoid AS varchar(64)) = CAST(c.oid AS varchar(64)) AND d.objsubid = 0
+WHERE n.nspname = %s AND c.relkind IN ('r', 'p', 'v', 'm', 'f') ORDER BY c.relname`, quoteLiteral(schema))
+}
+
+func isSysFreespacePermissionError(err error) bool {
+	var kingbaseError *gokb.Error
+	if !errors.As(err, &kingbaseError) || kingbaseError.Code != gokb.ErrorCode("42501") {
+		return false
+	}
+	message := strings.ToLower(strings.Join([]string{
+		kingbaseError.Message,
+		kingbaseError.Detail,
+		kingbaseError.Hint,
+		kingbaseError.InternalQuery,
+		kingbaseError.Where,
+	}, " "))
+	return strings.Contains(message, "sys_freespace") || strings.Contains(message, "pg_relation_size_ex")
+}
+
 func (s *server) listObjects(schema string, constraints metadataListConstraints) ([]objectInfo, error) {
 	effective, err := s.effectiveSchema(schema)
 	if err != nil {

agents/drivers/kingbase-go/main_test.go

@@ -117,6 +117,18 @@ func (connection *fallbackConn) QueryContext(_ context.Context, query string, _
 	if strings.Contains(query, "information_schema.table_constraints") {
 		return &valueRows{columns: []string{"column_name"}}, nil
 	}
+	if strings.Contains(query, "information_schema.tables") {
+		return nil, &gokb.Error{
+			Code:    gokb.ErrorCode("42501"),
+			Message: "permission denied for function sys_freespace",
+		}
+	}
+	if strings.Contains(query, "SELECT c.relname") {
+		return &valueRows{
+			columns: []string{"relname", "table_type", "description"},
+			rows:    [][]driver.Value{{"orders", "TABLE", "orders table"}},
+		}, nil
+	}
 	if strings.Contains(query, "sys_get_expr(") {
 		return nil, &gokb.Error{Code: gokb.ErrorCode("42883"), Message: "function sys_get_expr(pg_node_tree, oid) does not exist"}
 	}
@@ -334,6 +346,35 @@ func TestColumnsFallbackToPgGetExprAndCacheChoice(t *testing.T) {
 	}
 }
 
+func TestListTablesFallsBackForSysFreespacePermission(t *testing.T) {
+	registerExpressionFallbackDriver.Do(func() { sql.Register("kingbase-expression-fallback-test", fallbackDriver{}) })
+	state := &fallbackDriverState{}
+	expressionFallbackState.Store(state)
+	db, err := sql.Open("kingbase-expression-fallback-test", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	db.SetMaxOpenConns(1)
+	t.Cleanup(func() { _ = db.Close() })
+
+	server := newServer()
+	server.db = db
+	server.mode.mysqlCompat = true
+	tables, err := server.listTables("public", metadataListConstraints{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(tables) != 1 || tables[0].Name != "orders" {
+		t.Fatalf("unexpected fallback tables: %#v", tables)
+	}
+
+	state.mu.Lock()
+	defer state.mu.Unlock()
+	if len(state.queries) != 2 || !strings.Contains(state.queries[0], "information_schema.tables") || !strings.Contains(state.queries[1], "sys_catalog.sys_class") {
+		t.Fatalf("unexpected fallback query sequence: %v", state.queries)
+	}
+}
+
 func TestDetectMySQLCompatModePrefersDatabaseModeOverSQLModePresence(t *testing.T) {
 	oracle := "oracle"
 	state := &modeDetectionDriverState{