1890 lines
50 KiB
Go
1890 lines
50 KiB
Go
package main
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"regexp"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
_ "gitee.com/XuguDB/go-xugu-driver"
|
|
)
|
|
|
|
const protocolVersion = 1
|
|
const defaultMaxRows = 1000
|
|
const defaultXuguPort = 5138
|
|
const xuguListDatabasesSQL = `
|
|
SELECT DB_NAME
|
|
FROM ALL_DATABASES
|
|
ORDER BY DB_NAME`
|
|
const xuguListSchemasSQL = `
|
|
SELECT SCHEMA_NAME
|
|
FROM ALL_SCHEMAS
|
|
ORDER BY SCHEMA_NAME`
|
|
const xuguPrimaryKeyColumnsSQL = `
|
|
SELECT c.DEFINE
|
|
FROM ALL_CONSTRAINTS c
|
|
JOIN ALL_TABLES t ON t.DB_ID = c.DB_ID AND t.TABLE_ID = c.TABLE_ID
|
|
JOIN ALL_SCHEMAS s ON s.DB_ID = t.DB_ID AND s.SCHEMA_ID = t.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?)
|
|
AND UPPER(t.TABLE_NAME) = UPPER(?)
|
|
AND c.CONS_TYPE = 'P'`
|
|
const xuguListColumnsSQL = `
|
|
SELECT c.COL_NAME, c.TYPE_NAME, c.NOT_NULL, c.DEF_VAL, c.COMMENTS, c.SCALE, c."VARYING"
|
|
FROM ALL_COLUMNS c
|
|
JOIN ALL_TABLES t ON t.DB_ID = c.DB_ID AND t.TABLE_ID = c.TABLE_ID
|
|
JOIN ALL_SCHEMAS s ON s.DB_ID = t.DB_ID AND s.SCHEMA_ID = t.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?)
|
|
AND UPPER(t.TABLE_NAME) = UPPER(?)
|
|
AND (c.IS_HIDE IS NULL OR c.IS_HIDE = FALSE)
|
|
ORDER BY c.COL_NO`
|
|
const xuguListIndexesSQL = `
|
|
SELECT i.INDEX_NAME, i.KEYS, i.IS_UNIQUE, i.IS_PRIMARY, i.INDEX_TYPE, i.FILTER
|
|
FROM ALL_INDEXES i
|
|
JOIN ALL_TABLES t ON t.DB_ID = i.DB_ID AND t.TABLE_ID = i.TABLE_ID
|
|
JOIN ALL_SCHEMAS s ON s.DB_ID = t.DB_ID AND s.SCHEMA_ID = t.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?)
|
|
AND UPPER(t.TABLE_NAME) = UPPER(?)
|
|
ORDER BY i.INDEX_NAME`
|
|
|
|
var xuguDataTypes = []string{
|
|
"BOOLEAN",
|
|
"INTEGER",
|
|
"SMALLINT",
|
|
"BIGINT",
|
|
"FLOAT",
|
|
"NUMERIC",
|
|
"CHAR",
|
|
"VARCHAR",
|
|
"CLOB",
|
|
"DATE",
|
|
"TIME",
|
|
"TIMESTAMP",
|
|
"BINARY",
|
|
"VARBINARY",
|
|
"BLOB",
|
|
"XML",
|
|
"BOOL",
|
|
"INT",
|
|
"SHORT",
|
|
"LONGINT",
|
|
"LONG",
|
|
"REAL",
|
|
"DECIMAL",
|
|
"TEXT",
|
|
"NCHAR",
|
|
"NVARCHAR",
|
|
"NVARCHAR2",
|
|
}
|
|
|
|
type request struct {
|
|
ID json.RawMessage `json:"id"`
|
|
Method string `json:"method"`
|
|
Params map[string]json.RawMessage `json:"params"`
|
|
}
|
|
|
|
type response struct {
|
|
JSONRPC string `json:"jsonrpc,omitempty"`
|
|
ID json.RawMessage `json:"id,omitempty"`
|
|
Result any `json:"result,omitempty"`
|
|
Error *rpcError `json:"error,omitempty"`
|
|
}
|
|
|
|
type rpcError struct {
|
|
Code int `json:"code"`
|
|
Message string `json:"message"`
|
|
}
|
|
|
|
type connectParams struct {
|
|
Host string `json:"host"`
|
|
Port int `json:"port"`
|
|
Database string `json:"database"`
|
|
Username string `json:"username"`
|
|
Password string `json:"password"`
|
|
SysDBA bool `json:"sysdba"`
|
|
URLParams string `json:"url_params"`
|
|
ConnectionString string `json:"connection_string"`
|
|
}
|
|
|
|
type queryOptions struct {
|
|
SQL string `json:"sql"`
|
|
Database string `json:"database"`
|
|
Schema string `json:"schema"`
|
|
MaxRows int `json:"maxRows"`
|
|
FetchSize int `json:"fetchSize"`
|
|
}
|
|
|
|
type queryResult struct {
|
|
Columns []string `json:"columns"`
|
|
ColumnTypes []string `json:"column_types"`
|
|
Rows [][]any `json:"rows"`
|
|
AffectedRows int64 `json:"affected_rows"`
|
|
ExecutionTimeMS int64 `json:"execution_time_ms"`
|
|
Truncated bool `json:"truncated"`
|
|
}
|
|
|
|
func (r queryResult) MarshalJSON() ([]byte, error) {
|
|
type alias queryResult
|
|
value := alias(r)
|
|
if value.Columns == nil {
|
|
value.Columns = []string{}
|
|
}
|
|
if value.ColumnTypes == nil {
|
|
value.ColumnTypes = []string{}
|
|
}
|
|
if value.Rows == nil {
|
|
value.Rows = [][]any{}
|
|
}
|
|
return json.Marshal(value)
|
|
}
|
|
|
|
type queryPageResult struct {
|
|
Columns []string `json:"columns"`
|
|
ColumnTypes []string `json:"column_types"`
|
|
Rows [][]any `json:"rows"`
|
|
AffectedRows int64 `json:"affected_rows"`
|
|
ExecutionTimeMS int64 `json:"execution_time_ms"`
|
|
Truncated bool `json:"truncated"`
|
|
SessionID *string `json:"session_id"`
|
|
HasMore bool `json:"has_more"`
|
|
}
|
|
|
|
func (r queryPageResult) MarshalJSON() ([]byte, error) {
|
|
type alias queryPageResult
|
|
value := alias(r)
|
|
if value.Columns == nil {
|
|
value.Columns = []string{}
|
|
}
|
|
if value.ColumnTypes == nil {
|
|
value.ColumnTypes = []string{}
|
|
}
|
|
if value.Rows == nil {
|
|
value.Rows = [][]any{}
|
|
}
|
|
return json.Marshal(value)
|
|
}
|
|
|
|
type querySession struct {
|
|
rows *sql.Rows
|
|
columns []string
|
|
columnTypes []string
|
|
pending []any
|
|
remaining int
|
|
}
|
|
|
|
type databaseInfo struct {
|
|
Name string `json:"name"`
|
|
}
|
|
|
|
type tableInfo struct {
|
|
Name string `json:"name"`
|
|
TableType string `json:"table_type"`
|
|
Comment *string `json:"comment"`
|
|
}
|
|
|
|
type objectInfo struct {
|
|
Name string `json:"name"`
|
|
ObjectType string `json:"object_type"`
|
|
Schema string `json:"schema"`
|
|
Comment *string `json:"comment"`
|
|
}
|
|
|
|
type columnInfo struct {
|
|
Name string `json:"name"`
|
|
DataType string `json:"data_type"`
|
|
IsNullable bool `json:"is_nullable"`
|
|
ColumnDefault *string `json:"column_default"`
|
|
IsPrimaryKey bool `json:"is_primary_key"`
|
|
Extra *string `json:"extra"`
|
|
Comment *string `json:"comment"`
|
|
NumericPrecision *int `json:"numeric_precision"`
|
|
NumericScale *int `json:"numeric_scale"`
|
|
CharacterMaximumLength *int `json:"character_maximum_length"`
|
|
}
|
|
|
|
type indexInfo struct {
|
|
Name string `json:"name"`
|
|
Columns []string `json:"columns"`
|
|
IsUnique bool `json:"is_unique"`
|
|
IsPrimary bool `json:"is_primary"`
|
|
Filter *string `json:"filter"`
|
|
IndexType *string `json:"index_type"`
|
|
IncludedColumns []string `json:"included_columns"`
|
|
Comment *string `json:"comment"`
|
|
}
|
|
|
|
func (i indexInfo) MarshalJSON() ([]byte, error) {
|
|
type alias indexInfo
|
|
value := alias(i)
|
|
if value.Columns == nil {
|
|
value.Columns = []string{}
|
|
}
|
|
if value.IncludedColumns == nil {
|
|
value.IncludedColumns = []string{}
|
|
}
|
|
return json.Marshal(value)
|
|
}
|
|
|
|
type foreignKeyInfo struct {
|
|
Name string `json:"name"`
|
|
Column string `json:"column"`
|
|
RefTable string `json:"ref_table"`
|
|
RefColumn string `json:"ref_column"`
|
|
}
|
|
|
|
type triggerInfo struct {
|
|
Name string `json:"name"`
|
|
Event string `json:"event"`
|
|
Timing string `json:"timing"`
|
|
}
|
|
|
|
type server struct {
|
|
db *sql.DB
|
|
params connectParams
|
|
sessions map[string]*querySession
|
|
nextSessionID int64
|
|
}
|
|
|
|
func main() {
|
|
s := newServer()
|
|
encoder := json.NewEncoder(os.Stdout)
|
|
fmt.Fprintln(os.Stdout, `{"ready":true}`)
|
|
|
|
scanner := bufio.NewScanner(os.Stdin)
|
|
scanner.Buffer(make([]byte, 0, 64*1024), 512*1024*1024)
|
|
for scanner.Scan() {
|
|
line := strings.TrimSpace(scanner.Text())
|
|
if line == "" {
|
|
continue
|
|
}
|
|
resp, shutdown := s.handleLine(line)
|
|
if err := encoder.Encode(resp); err != nil {
|
|
fmt.Fprintf(os.Stderr, "failed to write response: %v\n", err)
|
|
return
|
|
}
|
|
if shutdown {
|
|
return
|
|
}
|
|
}
|
|
if err := scanner.Err(); err != nil && !errors.Is(err, io.EOF) {
|
|
fmt.Fprintf(os.Stderr, "failed to read stdin: %v\n", err)
|
|
}
|
|
}
|
|
|
|
func newServer() *server {
|
|
return &server{sessions: map[string]*querySession{}}
|
|
}
|
|
|
|
func (s *server) handleLine(line string) (response, bool) {
|
|
var req request
|
|
if err := json.Unmarshal([]byte(line), &req); err != nil {
|
|
return errorResponse(nil, err), false
|
|
}
|
|
if len(req.ID) == 0 {
|
|
req.ID = json.RawMessage("1")
|
|
}
|
|
result, shutdown, err := s.dispatch(req.Method, req.Params)
|
|
if err != nil {
|
|
return errorResponse(req.ID, err), false
|
|
}
|
|
return response{JSONRPC: "2.0", ID: req.ID, Result: result}, shutdown
|
|
}
|
|
|
|
func (s *server) dispatch(method string, params map[string]json.RawMessage) (any, bool, error) {
|
|
switch method {
|
|
case "handshake":
|
|
return map[string]any{
|
|
"protocolVersion": protocolVersion,
|
|
"agentProtocolVersion": protocolVersion,
|
|
"capabilities": []string{"connect", "test_connection", "metadata", "query", "ddl"},
|
|
}, false, nil
|
|
case "connect":
|
|
var cp connectParams
|
|
if err := decodeParams(params, &cp); err != nil {
|
|
return nil, false, err
|
|
}
|
|
return map[string]bool{"ok": true}, false, s.connect(cp)
|
|
case "test_connection":
|
|
var cp connectParams
|
|
if err := decodeParams(params, &cp); err != nil {
|
|
return nil, false, err
|
|
}
|
|
db, err := openDB(cp)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
defer db.Close()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
if err := db.PingContext(ctx); err != nil {
|
|
return nil, false, err
|
|
}
|
|
return map[string]bool{"ok": true}, false, nil
|
|
case "list_databases":
|
|
result, err := s.listDatabases()
|
|
return result, false, err
|
|
case "list_schemas":
|
|
if err := s.useDatabase(stringParam(params, "database")); err != nil {
|
|
return nil, false, err
|
|
}
|
|
result, err := s.listSchemas()
|
|
return result, false, err
|
|
case "list_tables":
|
|
if err := s.useDatabase(stringParam(params, "database")); err != nil {
|
|
return nil, false, err
|
|
}
|
|
schema := stringParam(params, "schema")
|
|
result, err := s.listTables(schema)
|
|
return result, false, err
|
|
case "list_objects":
|
|
if err := s.useDatabase(stringParam(params, "database")); err != nil {
|
|
return nil, false, err
|
|
}
|
|
schema := stringParam(params, "schema")
|
|
result, err := s.listObjects(schema)
|
|
return result, false, err
|
|
case "list_data_types":
|
|
return xuguDataTypes, false, nil
|
|
case "get_columns":
|
|
if err := s.useDatabase(stringParam(params, "database")); err != nil {
|
|
return nil, false, err
|
|
}
|
|
schema := stringParam(params, "schema")
|
|
table := stringParam(params, "table")
|
|
result, err := s.getColumns(schema, table)
|
|
return result, false, err
|
|
case "get_object_source":
|
|
if err := s.useDatabase(stringParam(params, "database")); err != nil {
|
|
return nil, false, err
|
|
}
|
|
schema := stringParam(params, "schema")
|
|
name := stringParam(params, "name")
|
|
objectType := stringParam(params, "object_type")
|
|
source, err := s.getObjectSource(schema, name, objectType)
|
|
return source, false, err
|
|
case "get_table_ddl":
|
|
if err := s.useDatabase(stringParam(params, "database")); err != nil {
|
|
return nil, false, err
|
|
}
|
|
schema := stringParam(params, "schema")
|
|
table := stringParam(params, "table")
|
|
ddl, err := s.getTableDDL(schema, table)
|
|
return ddl, false, err
|
|
case "execute_query":
|
|
var opts queryOptions
|
|
if err := decodeParams(params, &opts); err != nil {
|
|
return nil, false, err
|
|
}
|
|
result, err := s.executeQuery(opts)
|
|
return result, false, err
|
|
case "execute_query_page":
|
|
var opts queryOptions
|
|
if err := decodeParams(params, &opts); err != nil {
|
|
return nil, false, err
|
|
}
|
|
result, err := s.executeQueryPage(opts, intParam(params, "pageSize"))
|
|
return result, false, err
|
|
case "fetch_query_page":
|
|
result, err := s.fetchQueryPage(stringParam(params, "sessionId"), intParam(params, "pageSize"))
|
|
return result, false, err
|
|
case "close_query_session":
|
|
return s.closeQuerySession(stringParam(params, "sessionId")), false, nil
|
|
case "list_indexes":
|
|
if err := s.useDatabase(stringParam(params, "database")); err != nil {
|
|
return nil, false, err
|
|
}
|
|
schema := stringParam(params, "schema")
|
|
table := stringParam(params, "table")
|
|
result, err := s.listIndexes(schema, table)
|
|
return result, false, err
|
|
case "list_foreign_keys":
|
|
if err := s.useDatabase(stringParam(params, "database")); err != nil {
|
|
return nil, false, err
|
|
}
|
|
schema := stringParam(params, "schema")
|
|
table := stringParam(params, "table")
|
|
result, err := s.listForeignKeys(schema, table)
|
|
return result, false, err
|
|
case "list_triggers":
|
|
if err := s.useDatabase(stringParam(params, "database")); err != nil {
|
|
return nil, false, err
|
|
}
|
|
schema := stringParam(params, "schema")
|
|
table := stringParam(params, "table")
|
|
result, err := s.listTriggers(schema, table)
|
|
return result, false, err
|
|
case "get_explain_info":
|
|
sqlText := stringParam(params, "sql")
|
|
plan, err := s.getExplainInfo(sqlText)
|
|
return map[string]any{"plan": plan, "has_actual_stats": false}, false, err
|
|
case "execute_transaction":
|
|
result, err := s.executeTransaction(params)
|
|
return result, false, err
|
|
case "disconnect":
|
|
return map[string]bool{"ok": true}, false, s.disconnect()
|
|
case "shutdown":
|
|
_ = s.disconnect()
|
|
return map[string]bool{"ok": true}, true, nil
|
|
default:
|
|
return nil, false, fmt.Errorf("unknown method: %s", method)
|
|
}
|
|
}
|
|
|
|
func (s *server) connect(params connectParams) error {
|
|
_ = s.disconnect()
|
|
db, err := openDB(params)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
|
defer cancel()
|
|
if err := db.PingContext(ctx); err != nil {
|
|
db.Close()
|
|
return err
|
|
}
|
|
s.db = db
|
|
s.params = params
|
|
return nil
|
|
}
|
|
|
|
func (s *server) disconnect() error {
|
|
s.closeAllQuerySessions()
|
|
if s.db == nil {
|
|
return nil
|
|
}
|
|
err := s.db.Close()
|
|
s.db = nil
|
|
return err
|
|
}
|
|
|
|
func openDB(params connectParams) (*sql.DB, error) {
|
|
dsn := buildDSN(params)
|
|
db, err := sql.Open("xugu", dsn)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
db.SetMaxOpenConns(4)
|
|
db.SetMaxIdleConns(1)
|
|
db.SetConnMaxLifetime(30 * time.Minute)
|
|
return db, nil
|
|
}
|
|
|
|
func buildDSN(params connectParams) string {
|
|
connectionString := strings.TrimSpace(params.ConnectionString)
|
|
if looksLikeXuguDSN(connectionString) {
|
|
return connectionString
|
|
}
|
|
if parsed := parseXuguURL(connectionString); parsed.Host != "" {
|
|
if parsed.Database == "" {
|
|
parsed.Database = params.Database
|
|
}
|
|
if parsed.Username == "" {
|
|
parsed.Username = params.Username
|
|
}
|
|
if parsed.Password == "" {
|
|
parsed.Password = params.Password
|
|
}
|
|
return buildXuguDSN(parsed.Host, parsed.Port, parsed.Database, parsed.Username, parsed.Password, params.URLParams)
|
|
}
|
|
|
|
if jdbc := parseXuguJDBCURL(connectionString); jdbc.Host != "" {
|
|
return buildXuguDSN(jdbc.Host, jdbc.Port, jdbc.Database, params.Username, params.Password, params.URLParams)
|
|
}
|
|
|
|
return buildXuguDSN(params.Host, params.Port, params.Database, params.Username, params.Password, params.URLParams)
|
|
}
|
|
|
|
func looksLikeXuguDSN(value string) bool {
|
|
upper := strings.ToUpper(value)
|
|
return strings.Contains(upper, "IP=") && strings.Contains(upper, "DB=") && strings.Contains(upper, "USER=")
|
|
}
|
|
|
|
type xuguURLInfo struct {
|
|
Host string
|
|
Port int
|
|
Database string
|
|
Username string
|
|
Password string
|
|
}
|
|
|
|
var xuguJDBCRegexp = regexp.MustCompile(`(?i)^jdbc:xugu://([^/:]+)(?::([0-9]+))?/([^?;]+)`)
|
|
|
|
func parseXuguJDBCURL(value string) xuguURLInfo {
|
|
value = strings.TrimSpace(value)
|
|
match := xuguJDBCRegexp.FindStringSubmatch(value)
|
|
if len(match) != 4 {
|
|
return xuguURLInfo{}
|
|
}
|
|
return xuguURLInfo{Host: match[1], Port: parsePort(match[2]), Database: match[3]}
|
|
}
|
|
|
|
func parseXuguURL(value string) xuguURLInfo {
|
|
value = strings.TrimSpace(value)
|
|
if !strings.HasPrefix(strings.ToLower(value), "xugu://") {
|
|
return xuguURLInfo{}
|
|
}
|
|
withoutScheme := value[len("xugu://"):]
|
|
var userInfo string
|
|
hostPart := withoutScheme
|
|
if at := strings.LastIndex(hostPart, "@"); at >= 0 {
|
|
userInfo = hostPart[:at]
|
|
hostPart = hostPart[at+1:]
|
|
}
|
|
if slash := strings.IndexAny(hostPart, "/?"); slash >= 0 {
|
|
databasePart := strings.TrimLeft(hostPart[slash:], "/")
|
|
hostPart = hostPart[:slash]
|
|
if q := strings.Index(databasePart, "?"); q >= 0 {
|
|
databasePart = databasePart[:q]
|
|
}
|
|
info := parseHostPort(hostPart)
|
|
info.Database = databasePart
|
|
if userInfo != "" {
|
|
info.Username, info.Password = splitUserInfo(userInfo)
|
|
}
|
|
return info
|
|
}
|
|
info := parseHostPort(hostPart)
|
|
if userInfo != "" {
|
|
info.Username, info.Password = splitUserInfo(userInfo)
|
|
}
|
|
return info
|
|
}
|
|
|
|
func parseHostPort(value string) xuguURLInfo {
|
|
host := strings.TrimSpace(value)
|
|
port := 0
|
|
if idx := strings.LastIndex(host, ":"); idx > 0 {
|
|
port = parsePort(host[idx+1:])
|
|
host = host[:idx]
|
|
}
|
|
return xuguURLInfo{Host: host, Port: port}
|
|
}
|
|
|
|
func splitUserInfo(value string) (string, string) {
|
|
if idx := strings.Index(value, ":"); idx >= 0 {
|
|
return value[:idx], value[idx+1:]
|
|
}
|
|
return value, ""
|
|
}
|
|
|
|
func buildXuguDSN(host string, port int, database, username, password, urlParams string) string {
|
|
if port <= 0 {
|
|
port = defaultXuguPort
|
|
}
|
|
parts := []string{
|
|
"IP=" + host,
|
|
"DB=" + strings.TrimSpace(database),
|
|
"User=" + strings.TrimSpace(username),
|
|
"PWD=" + strings.TrimSpace(password),
|
|
"Port=" + strconv.Itoa(port),
|
|
}
|
|
if !hasDSNParam(urlParams, "CHAR_SET") && !hasDSNParam(urlParams, "CHARSET") {
|
|
parts = append(parts, "CHAR_SET=UTF8")
|
|
}
|
|
for _, param := range splitDSNParams(urlParams) {
|
|
parts = append(parts, param)
|
|
}
|
|
return strings.Join(parts, ";")
|
|
}
|
|
|
|
func hasDSNParam(raw, key string) bool {
|
|
key = strings.ToUpper(strings.TrimSpace(key))
|
|
for _, param := range splitDSNParams(raw) {
|
|
name, _, _ := strings.Cut(param, "=")
|
|
if strings.ToUpper(strings.TrimSpace(name)) == key {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func splitDSNParams(raw string) []string {
|
|
raw = strings.TrimSpace(strings.Trim(raw, ";"))
|
|
if raw == "" {
|
|
return nil
|
|
}
|
|
if strings.Contains(raw, ";") {
|
|
items := strings.Split(raw, ";")
|
|
result := make([]string, 0, len(items))
|
|
for _, item := range items {
|
|
item = strings.TrimSpace(item)
|
|
if item != "" {
|
|
result = append(result, item)
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
if strings.Contains(raw, "&") {
|
|
items := strings.Split(raw, "&")
|
|
result := make([]string, 0, len(items))
|
|
for _, item := range items {
|
|
item = strings.TrimSpace(item)
|
|
if item != "" {
|
|
result = append(result, item)
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
return []string{raw}
|
|
}
|
|
|
|
func parsePort(value string) int {
|
|
port, _ := strconv.Atoi(value)
|
|
return port
|
|
}
|
|
|
|
func (s *server) requireDB() (*sql.DB, error) {
|
|
if s.db == nil {
|
|
return nil, errors.New("agent is not connected")
|
|
}
|
|
return s.db, nil
|
|
}
|
|
|
|
func (s *server) useDatabase(database string) error {
|
|
database = strings.TrimSpace(database)
|
|
if database == "" {
|
|
return nil
|
|
}
|
|
if strings.EqualFold(database, configuredDatabaseName(s.params)) {
|
|
return nil
|
|
}
|
|
db, err := s.requireDB()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = db.Exec("USE " + quoteIdentifier(database))
|
|
return err
|
|
}
|
|
|
|
func (s *server) listDatabases() ([]databaseInfo, error) {
|
|
rows, err := s.queryRows(xuguListDatabasesSQL, nil)
|
|
if err != nil {
|
|
if fallback := fallbackDatabasesFromParams(s.params); len(fallback) > 0 && isXuguMetadataAccessError(err) {
|
|
return fallback, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []databaseInfo
|
|
for rows.Next() {
|
|
var name string
|
|
if err := rows.Scan(&name); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, databaseInfo{Name: name})
|
|
}
|
|
return emptyIfNil(result), rows.Err()
|
|
}
|
|
|
|
func fallbackDatabasesFromParams(params connectParams) []databaseInfo {
|
|
if name := configuredDatabaseName(params); name != "" {
|
|
return []databaseInfo{{Name: name}}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func configuredDatabaseName(params connectParams) string {
|
|
if name := strings.TrimSpace(params.Database); name != "" {
|
|
return name
|
|
}
|
|
connectionString := strings.TrimSpace(params.ConnectionString)
|
|
if parsed := parseXuguURL(connectionString); parsed.Database != "" {
|
|
return parsed.Database
|
|
}
|
|
if jdbc := parseXuguJDBCURL(connectionString); jdbc.Database != "" {
|
|
return jdbc.Database
|
|
}
|
|
if value := xuguDSNValue(connectionString, "DB"); value != "" {
|
|
return value
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func isXuguMetadataAccessError(err error) bool {
|
|
message := strings.ToUpper(err.Error())
|
|
return strings.Contains(message, "E18012") ||
|
|
strings.Contains(message, "权限不够") ||
|
|
strings.Contains(message, "ALL_DATABASES") ||
|
|
strings.Contains(message, "SYS_DATABASES") ||
|
|
strings.Contains(message, "ALL_SCHEMAS") ||
|
|
strings.Contains(message, "SYS_SCHEMAS") ||
|
|
strings.Contains(message, "ALL_TABLES") ||
|
|
strings.Contains(message, "SYS_TABLES") ||
|
|
strings.Contains(message, "ALL_VIEWS") ||
|
|
strings.Contains(message, "SYS_VIEWS") ||
|
|
strings.Contains(message, "ALL_COLUMNS") ||
|
|
strings.Contains(message, "ALL_CONSTRAINTS") ||
|
|
strings.Contains(message, "ALL_INDEXES") ||
|
|
strings.Contains(message, "SYS_COLUMNS") ||
|
|
strings.Contains(message, "SYS_CONSTRAINTS") ||
|
|
strings.Contains(message, "SYS_INDEXES") ||
|
|
strings.Contains(message, "SYS_TRIGGERS")
|
|
}
|
|
|
|
func xuguDSNValue(dsn string, key string) string {
|
|
for _, part := range strings.Split(dsn, ";") {
|
|
name, value, ok := strings.Cut(part, "=")
|
|
if ok && strings.EqualFold(strings.TrimSpace(name), key) {
|
|
return strings.TrimSpace(value)
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func (s *server) listSchemas() ([]string, error) {
|
|
rows, err := s.queryRows(xuguListSchemasSQL, nil)
|
|
if err != nil {
|
|
if fallback := strings.ToUpper(strings.TrimSpace(s.params.Username)); fallback != "" && isXuguMetadataAccessError(err) {
|
|
return []string{fallback}, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []string
|
|
for rows.Next() {
|
|
var schema string
|
|
if err := rows.Scan(&schema); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, schema)
|
|
}
|
|
return emptyIfNil(result), rows.Err()
|
|
}
|
|
|
|
func (s *server) currentSchema() (string, error) {
|
|
rows, err := s.queryRows(`
|
|
SELECT s.SCHEMA_NAME
|
|
FROM SYS_SCHEMAS s
|
|
JOIN SYS_USERS u ON u.DB_ID = s.DB_ID AND u.USER_ID = s.USER_ID
|
|
WHERE UPPER(u.USER_NAME) = UPPER(?)
|
|
ORDER BY CASE WHEN UPPER(s.SCHEMA_NAME) = UPPER(?) THEN 0 ELSE 1 END, s.SCHEMA_NAME`, []any{s.params.Username, s.params.Username})
|
|
if err != nil {
|
|
if fallback := strings.ToUpper(strings.TrimSpace(s.params.Username)); fallback != "" && isXuguMetadataAccessError(err) {
|
|
return fallback, nil
|
|
}
|
|
return "", err
|
|
}
|
|
defer rows.Close()
|
|
if rows.Next() {
|
|
var schema string
|
|
if err := rows.Scan(&schema); err != nil {
|
|
return "", err
|
|
}
|
|
return strings.ToUpper(schema), nil
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return "", err
|
|
}
|
|
return strings.ToUpper(strings.TrimSpace(s.params.Username)), nil
|
|
}
|
|
|
|
func (s *server) normalizeSchema(schema string) (string, error) {
|
|
schema = strings.TrimSpace(schema)
|
|
if schema == "" {
|
|
return s.currentSchema()
|
|
}
|
|
return strings.ToUpper(schema), nil
|
|
}
|
|
|
|
func (s *server) listTables(schema string) ([]tableInfo, error) {
|
|
schema, err := s.normalizeSchema(schema)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.queryRows(`
|
|
SELECT t.TABLE_NAME, 'TABLE' AS TABLE_TYPE, t.COMMENTS
|
|
FROM ALL_TABLES t
|
|
JOIN ALL_SCHEMAS s ON s.DB_ID = t.DB_ID AND s.SCHEMA_ID = t.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?)
|
|
UNION ALL
|
|
SELECT v.VIEW_NAME, 'VIEW' AS TABLE_TYPE, v.COMMENTS
|
|
FROM ALL_VIEWS v
|
|
JOIN ALL_SCHEMAS s ON s.DB_ID = v.DB_ID AND s.SCHEMA_ID = v.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?)
|
|
ORDER BY 2, 1`, []any{schema, schema})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []tableInfo
|
|
for rows.Next() {
|
|
var item tableInfo
|
|
if err := rows.Scan(&item.Name, &item.TableType, &item.Comment); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
return emptyIfNil(result), rows.Err()
|
|
}
|
|
|
|
func (s *server) listObjects(schema string) ([]objectInfo, error) {
|
|
schema, err := s.normalizeSchema(schema)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.queryRows(`
|
|
SELECT t.TABLE_NAME, 'TABLE' AS OBJECT_TYPE, t.COMMENTS
|
|
FROM ALL_TABLES t
|
|
JOIN ALL_SCHEMAS s ON s.DB_ID = t.DB_ID AND s.SCHEMA_ID = t.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?)
|
|
UNION ALL
|
|
SELECT v.VIEW_NAME, 'VIEW' AS OBJECT_TYPE, v.COMMENTS
|
|
FROM ALL_VIEWS v
|
|
JOIN ALL_SCHEMAS s ON s.DB_ID = v.DB_ID AND s.SCHEMA_ID = v.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?)
|
|
ORDER BY 2, 1`, []any{schema, schema})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []objectInfo
|
|
for rows.Next() {
|
|
var item objectInfo
|
|
item.Schema = schema
|
|
if err := rows.Scan(&item.Name, &item.ObjectType, &item.Comment); err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
return emptyIfNil(result), rows.Err()
|
|
}
|
|
|
|
func (s *server) getColumns(schema, table string) ([]columnInfo, error) {
|
|
schema, err := s.normalizeSchema(schema)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
table = strings.ToUpper(strings.TrimSpace(table))
|
|
primaryKeys, err := s.primaryKeyColumns(schema, table)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.queryRows(xuguListColumnsSQL, []any{schema, table})
|
|
if err != nil {
|
|
if isXuguMetadataAccessError(err) {
|
|
return s.columnsFromSelect(schema, table, primaryKeys)
|
|
}
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []columnInfo
|
|
for rows.Next() {
|
|
var item columnInfo
|
|
var notNull any
|
|
var scale *int
|
|
var varying any
|
|
if err := rows.Scan(
|
|
&item.Name,
|
|
&item.DataType,
|
|
¬Null,
|
|
&item.ColumnDefault,
|
|
&item.Comment,
|
|
&scale,
|
|
&varying,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
item.DataType = normalizeXuguColumnType(item.DataType, varying)
|
|
item.IsNullable = !truthy(notNull)
|
|
item.IsPrimaryKey = primaryKeys[strings.ToUpper(item.Name)]
|
|
item.NumericPrecision, item.NumericScale, item.CharacterMaximumLength = decodeXuguScale(item.DataType, scale)
|
|
result = append(result, item)
|
|
}
|
|
return emptyIfNil(result), rows.Err()
|
|
}
|
|
|
|
func (s *server) columnsFromSelect(schema, table string, primaryKeys map[string]bool) ([]columnInfo, error) {
|
|
rows, err := s.queryRows(
|
|
"SELECT * FROM "+quoteIdentifier(schema)+"."+quoteIdentifier(table)+" WHERE 1 = 0",
|
|
nil,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
types, err := rows.ColumnTypes()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result := make([]columnInfo, 0, len(types))
|
|
for _, columnType := range types {
|
|
item := columnInfo{
|
|
Name: columnType.Name(),
|
|
DataType: columnType.DatabaseTypeName(),
|
|
IsPrimaryKey: primaryKeys[strings.ToUpper(columnType.Name())],
|
|
}
|
|
if nullable, ok := columnType.Nullable(); ok {
|
|
item.IsNullable = nullable
|
|
} else {
|
|
item.IsNullable = true
|
|
}
|
|
if length, ok := columnType.Length(); ok {
|
|
value := int(length)
|
|
item.CharacterMaximumLength = &value
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
return emptyIfNil(result), nil
|
|
}
|
|
|
|
func (s *server) primaryKeyColumns(schema, table string) (map[string]bool, error) {
|
|
rows, err := s.queryRows(xuguPrimaryKeyColumnsSQL, []any{schema, table})
|
|
if err != nil {
|
|
if isXuguMetadataAccessError(err) {
|
|
return map[string]bool{}, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
result := map[string]bool{}
|
|
for rows.Next() {
|
|
var define string
|
|
if err := rows.Scan(&define); err != nil {
|
|
return nil, err
|
|
}
|
|
for _, column := range parseQuotedIdentifiers(define) {
|
|
result[strings.ToUpper(column)] = true
|
|
}
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (s *server) listIndexes(schema, table string) ([]indexInfo, error) {
|
|
schema, err := s.normalizeSchema(schema)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
table = strings.ToUpper(strings.TrimSpace(table))
|
|
rows, err := s.queryRows(xuguListIndexesSQL, []any{schema, table})
|
|
if err != nil {
|
|
if isXuguMetadataAccessError(err) {
|
|
return []indexInfo{}, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var result []indexInfo
|
|
for rows.Next() {
|
|
var item indexInfo
|
|
var keys string
|
|
var unique, primary any
|
|
var indexType any
|
|
if err := rows.Scan(&item.Name, &keys, &unique, &primary, &indexType, &item.Filter); err != nil {
|
|
return nil, err
|
|
}
|
|
item.Columns = parseIndexKeys(keys)
|
|
item.IsUnique = truthy(unique)
|
|
item.IsPrimary = truthy(primary)
|
|
item.IndexType = stringPtr(indexTypeName(indexType))
|
|
item.IncludedColumns = []string{}
|
|
result = append(result, item)
|
|
}
|
|
return emptyIfNil(result), rows.Err()
|
|
}
|
|
|
|
func (s *server) listForeignKeys(schema, table string) ([]foreignKeyInfo, error) {
|
|
schema, err := s.normalizeSchema(schema)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
table = strings.ToUpper(strings.TrimSpace(table))
|
|
rows, err := s.queryRows(`
|
|
SELECT c.CONS_NAME, c.DEFINE, rt.TABLE_NAME
|
|
FROM SYS_CONSTRAINTS c
|
|
JOIN SYS_TABLES t ON t.DB_ID = c.DB_ID AND t.TABLE_ID = c.TABLE_ID
|
|
JOIN SYS_TABLES rt ON rt.DB_ID = c.DB_ID AND rt.TABLE_ID = c.REF_TABLE_ID
|
|
JOIN SYS_SCHEMAS s ON s.DB_ID = t.DB_ID AND s.SCHEMA_ID = t.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?)
|
|
AND UPPER(t.TABLE_NAME) = UPPER(?)
|
|
AND c.CONS_TYPE = 'F'
|
|
ORDER BY c.CONS_NAME`, []any{schema, table})
|
|
if err != nil {
|
|
if isXuguMetadataAccessError(err) {
|
|
return []foreignKeyInfo{}, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []foreignKeyInfo
|
|
for rows.Next() {
|
|
var name, define, refTable string
|
|
if err := rows.Scan(&name, &define, &refTable); err != nil {
|
|
return nil, err
|
|
}
|
|
local, ref := parseForeignKeyColumns(define)
|
|
for i, column := range local {
|
|
item := foreignKeyInfo{Name: name, Column: column, RefTable: refTable}
|
|
if i < len(ref) {
|
|
item.RefColumn = ref[i]
|
|
}
|
|
result = append(result, item)
|
|
}
|
|
}
|
|
return emptyIfNil(result), rows.Err()
|
|
}
|
|
|
|
func (s *server) listTriggers(schema, table string) ([]triggerInfo, error) {
|
|
schema, err := s.normalizeSchema(schema)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
table = strings.ToUpper(strings.TrimSpace(table))
|
|
rows, err := s.queryRows(`
|
|
SELECT tr.TRIG_NAME, tr.TRIG_EVENT, tr.TRIG_TIME
|
|
FROM SYS_TRIGGERS tr
|
|
JOIN SYS_TABLES t ON t.DB_ID = tr.DB_ID AND t.TABLE_ID = tr.OBJ_ID
|
|
JOIN SYS_SCHEMAS s ON s.DB_ID = t.DB_ID AND s.SCHEMA_ID = t.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?)
|
|
AND UPPER(t.TABLE_NAME) = UPPER(?)
|
|
ORDER BY tr.TRIG_NAME`, []any{schema, table})
|
|
if err != nil {
|
|
if isXuguMetadataAccessError(err) {
|
|
return []triggerInfo{}, nil
|
|
}
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var result []triggerInfo
|
|
for rows.Next() {
|
|
var item triggerInfo
|
|
var event, timing any
|
|
if err := rows.Scan(&item.Name, &event, &timing); err != nil {
|
|
return nil, err
|
|
}
|
|
item.Event = triggerEventName(event)
|
|
item.Timing = triggerTimingName(timing)
|
|
result = append(result, item)
|
|
}
|
|
return emptyIfNil(result), rows.Err()
|
|
}
|
|
|
|
func (s *server) getObjectSource(schema, name, objectType string) (map[string]any, error) {
|
|
var err error
|
|
schema, err = s.normalizeSchema(schema)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
sourceSQL, args, err := objectSourceQuery(schema, name, objectType)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.queryRows(sourceSQL, args)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var builder strings.Builder
|
|
for rows.Next() {
|
|
var line string
|
|
if err := rows.Scan(&line); err != nil {
|
|
return nil, err
|
|
}
|
|
builder.WriteString(line)
|
|
}
|
|
return map[string]any{"name": name, "object_type": objectType, "schema": schema, "source": builder.String()}, rows.Err()
|
|
}
|
|
|
|
func (s *server) getTableDDL(schema, table string) (string, error) {
|
|
var err error
|
|
schema, err = s.normalizeSchema(schema)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
var ddl string
|
|
rows, err := s.queryRows("SELECT TO_CHAR(DBMS_METADATA.GET_DDL('TABLE', ?, ?)) FROM DUAL", []any{strings.ToUpper(table), schema})
|
|
if err == nil {
|
|
defer rows.Close()
|
|
if rows.Next() {
|
|
if scanErr := rows.Scan(&ddl); scanErr == nil && strings.TrimSpace(ddl) != "" {
|
|
if err := rows.Err(); err != nil {
|
|
return "", err
|
|
}
|
|
return s.appendTableIndexDDL(schema, table, ddl), nil
|
|
}
|
|
}
|
|
}
|
|
ddl, err = s.buildTableDDL(schema, table)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return s.appendTableIndexDDL(schema, table, ddl), nil
|
|
}
|
|
|
|
func (s *server) getExplainInfo(sqlText string) (string, error) {
|
|
if strings.TrimSpace(sqlText) == "" {
|
|
return "", errors.New("sql is required")
|
|
}
|
|
rows, err := s.queryRows("EXPLAIN "+trimStatementSQL(sqlText), nil)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer rows.Close()
|
|
columns, err := rows.Columns()
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
var builder strings.Builder
|
|
for rows.Next() {
|
|
values, err := scanRow(rows, len(columns))
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
builder.WriteString(joinValues(values, "\t"))
|
|
builder.WriteByte('\n')
|
|
}
|
|
return strings.TrimSpace(builder.String()), rows.Err()
|
|
}
|
|
|
|
func (s *server) executeTransaction(params map[string]json.RawMessage) (queryResult, error) {
|
|
var payload struct {
|
|
Statements []string `json:"statements"`
|
|
Schema string `json:"schema"`
|
|
}
|
|
if err := decodeParams(params, &payload); err != nil {
|
|
return queryResult{}, err
|
|
}
|
|
db, err := s.requireDB()
|
|
if err != nil {
|
|
return queryResult{}, err
|
|
}
|
|
tx, err := db.Begin()
|
|
if err != nil {
|
|
return queryResult{}, err
|
|
}
|
|
if strings.TrimSpace(payload.Schema) != "" {
|
|
if _, err := tx.Exec("SET SCHEMA " + quoteIdentifier(payload.Schema)); err != nil {
|
|
tx.Rollback()
|
|
return queryResult{}, err
|
|
}
|
|
}
|
|
var affected int64
|
|
start := time.Now()
|
|
for _, statement := range payload.Statements {
|
|
statement = trimStatementSQL(statement)
|
|
if statement == "" {
|
|
continue
|
|
}
|
|
result, err := tx.Exec(statement)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return queryResult{}, err
|
|
}
|
|
count, _ := result.RowsAffected()
|
|
affected += count
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return queryResult{}, err
|
|
}
|
|
return queryResult{
|
|
Columns: []string{},
|
|
ColumnTypes: []string{},
|
|
Rows: [][]any{},
|
|
AffectedRows: affected,
|
|
ExecutionTimeMS: time.Since(start).Milliseconds(),
|
|
}, nil
|
|
}
|
|
|
|
func (s *server) executeQueryPage(opts queryOptions, pageSize int) (queryPageResult, error) {
|
|
start := time.Now()
|
|
if err := s.useDatabase(opts.Database); err != nil {
|
|
return queryPageResult{}, err
|
|
}
|
|
if strings.TrimSpace(opts.Schema) != "" {
|
|
if err := s.setSchema(opts.Schema); err != nil {
|
|
return queryPageResult{}, err
|
|
}
|
|
}
|
|
sqlText := trimStatementSQL(opts.SQL)
|
|
if !isQuerySQL(sqlText) {
|
|
result, err := s.executeQuery(opts)
|
|
return queryPageResult{
|
|
Columns: result.Columns,
|
|
ColumnTypes: result.ColumnTypes,
|
|
Rows: result.Rows,
|
|
AffectedRows: result.AffectedRows,
|
|
ExecutionTimeMS: result.ExecutionTimeMS,
|
|
Truncated: result.Truncated,
|
|
SessionID: nil,
|
|
HasMore: false,
|
|
}, err
|
|
}
|
|
rows, err := s.queryRows(sqlText, nil)
|
|
if err != nil {
|
|
return queryPageResult{}, err
|
|
}
|
|
columns, err := rows.Columns()
|
|
if err != nil {
|
|
rows.Close()
|
|
return queryPageResult{}, err
|
|
}
|
|
columnTypes := columnTypeNames(rows)
|
|
maxRows := opts.MaxRows
|
|
if maxRows <= 0 {
|
|
maxRows = defaultMaxRows
|
|
}
|
|
session := &querySession{rows: rows, columns: columns, columnTypes: columnTypes, remaining: maxRows}
|
|
result, err := readQuerySessionPage(session, pageSize)
|
|
result.ExecutionTimeMS = time.Since(start).Milliseconds()
|
|
if err != nil {
|
|
rows.Close()
|
|
return queryPageResult{}, err
|
|
}
|
|
if result.HasMore {
|
|
sessionID := s.storeQuerySession(session)
|
|
result.SessionID = &sessionID
|
|
} else {
|
|
rows.Close()
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *server) fetchQueryPage(sessionID string, pageSize int) (queryPageResult, error) {
|
|
session := s.sessions[sessionID]
|
|
if session == nil {
|
|
return queryPageResult{Columns: []string{}, ColumnTypes: []string{}, Rows: [][]any{}, SessionID: nil, HasMore: false}, nil
|
|
}
|
|
result, err := readQuerySessionPage(session, pageSize)
|
|
if err != nil {
|
|
s.closeQuerySession(sessionID)
|
|
return queryPageResult{}, err
|
|
}
|
|
if result.HasMore {
|
|
result.SessionID = &sessionID
|
|
} else {
|
|
s.closeQuerySession(sessionID)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (s *server) storeQuerySession(session *querySession) string {
|
|
s.nextSessionID++
|
|
sessionID := fmt.Sprintf("xugu-%d", s.nextSessionID)
|
|
s.sessions[sessionID] = session
|
|
return sessionID
|
|
}
|
|
|
|
func (s *server) closeQuerySession(sessionID string) bool {
|
|
session := s.sessions[sessionID]
|
|
if session == nil {
|
|
return false
|
|
}
|
|
session.rows.Close()
|
|
delete(s.sessions, sessionID)
|
|
return true
|
|
}
|
|
|
|
func (s *server) closeAllQuerySessions() {
|
|
for sessionID := range s.sessions {
|
|
s.closeQuerySession(sessionID)
|
|
}
|
|
}
|
|
|
|
func readQuerySessionPage(session *querySession, pageSize int) (queryPageResult, error) {
|
|
if pageSize <= 0 {
|
|
pageSize = defaultMaxRows
|
|
}
|
|
result := queryPageResult{Columns: session.columns, ColumnTypes: session.columnTypes, Rows: [][]any{}, SessionID: nil, HasMore: false}
|
|
for len(result.Rows) < pageSize && session.remaining > 0 {
|
|
if session.pending != nil {
|
|
result.Rows = append(result.Rows, session.pending)
|
|
session.pending = nil
|
|
session.remaining--
|
|
continue
|
|
}
|
|
if !session.rows.Next() {
|
|
return result, session.rows.Err()
|
|
}
|
|
row, err := scanRow(session.rows, len(session.columns))
|
|
if err != nil {
|
|
return queryPageResult{}, err
|
|
}
|
|
result.Rows = append(result.Rows, row)
|
|
session.remaining--
|
|
}
|
|
if session.remaining <= 0 {
|
|
result.Truncated = true
|
|
return result, nil
|
|
}
|
|
if session.rows.Next() {
|
|
row, err := scanRow(session.rows, len(session.columns))
|
|
if err != nil {
|
|
return queryPageResult{}, err
|
|
}
|
|
session.pending = row
|
|
result.HasMore = true
|
|
return result, nil
|
|
}
|
|
return result, session.rows.Err()
|
|
}
|
|
|
|
func (s *server) executeQuery(opts queryOptions) (queryResult, error) {
|
|
start := time.Now()
|
|
if err := s.useDatabase(opts.Database); err != nil {
|
|
return queryResult{}, err
|
|
}
|
|
if strings.TrimSpace(opts.Schema) != "" {
|
|
if err := s.setSchema(opts.Schema); err != nil {
|
|
return queryResult{}, err
|
|
}
|
|
}
|
|
sqlText := trimStatementSQL(opts.SQL)
|
|
maxRows := opts.MaxRows
|
|
if maxRows <= 0 {
|
|
maxRows = defaultMaxRows
|
|
}
|
|
if isQuerySQL(sqlText) {
|
|
result, err := s.executeSelect(sqlText, maxRows)
|
|
result.ExecutionTimeMS = time.Since(start).Milliseconds()
|
|
return result, err
|
|
}
|
|
db, err := s.requireDB()
|
|
if err != nil {
|
|
return queryResult{}, err
|
|
}
|
|
execResult, err := db.Exec(sqlText)
|
|
if err != nil {
|
|
return queryResult{}, err
|
|
}
|
|
affected, _ := execResult.RowsAffected()
|
|
return queryResult{Columns: []string{}, ColumnTypes: []string{}, Rows: [][]any{}, AffectedRows: affected, ExecutionTimeMS: time.Since(start).Milliseconds()}, nil
|
|
}
|
|
|
|
func (s *server) executeSelect(sqlText string, maxRows int) (queryResult, error) {
|
|
rows, err := s.queryRows(sqlText, nil)
|
|
if err != nil {
|
|
return queryResult{}, err
|
|
}
|
|
defer rows.Close()
|
|
columns, err := rows.Columns()
|
|
if err != nil {
|
|
return queryResult{}, err
|
|
}
|
|
result := queryResult{Columns: columns, ColumnTypes: columnTypeNames(rows), Rows: [][]any{}}
|
|
for rows.Next() {
|
|
if len(result.Rows) >= maxRows {
|
|
result.Truncated = true
|
|
break
|
|
}
|
|
values, err := scanRow(rows, len(columns))
|
|
if err != nil {
|
|
return queryResult{}, err
|
|
}
|
|
result.Rows = append(result.Rows, values)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func scanRow(rows *sql.Rows, columnCount int) ([]any, error) {
|
|
values := make([]any, columnCount)
|
|
scanTargets := make([]any, columnCount)
|
|
for i := range values {
|
|
scanTargets[i] = &values[i]
|
|
}
|
|
if err := rows.Scan(scanTargets...); err != nil {
|
|
return nil, err
|
|
}
|
|
for i, value := range values {
|
|
values[i] = normalizeValue(value)
|
|
}
|
|
return values, nil
|
|
}
|
|
|
|
func columnTypeNames(rows *sql.Rows) []string {
|
|
types, err := rows.ColumnTypes()
|
|
if err != nil {
|
|
return []string{}
|
|
}
|
|
result := make([]string, 0, len(types))
|
|
for _, columnType := range types {
|
|
result = append(result, columnType.DatabaseTypeName())
|
|
}
|
|
return result
|
|
}
|
|
|
|
func (s *server) setSchema(schema string) error {
|
|
db, err := s.requireDB()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = db.Exec("SET SCHEMA " + quoteIdentifier(schema))
|
|
return err
|
|
}
|
|
|
|
func (s *server) queryRows(sqlText string, args []any) (*sql.Rows, error) {
|
|
db, err := s.requireDB()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(args) == 0 {
|
|
return db.Query(sqlText)
|
|
}
|
|
return db.Query(sqlText, args...)
|
|
}
|
|
|
|
func objectSourceQuery(schema, name, objectType string) (string, []any, error) {
|
|
objectType = strings.ToUpper(strings.TrimSpace(objectType))
|
|
name = strings.ToUpper(strings.TrimSpace(name))
|
|
switch objectType {
|
|
case "VIEW":
|
|
return `
|
|
SELECT TO_CHAR(v.DEFINE)
|
|
FROM SYS_VIEWS v
|
|
JOIN SYS_SCHEMAS s ON s.DB_ID = v.DB_ID AND s.SCHEMA_ID = v.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?) AND UPPER(v.VIEW_NAME) = UPPER(?)`, []any{schema, name}, nil
|
|
case "TRIGGER":
|
|
return `
|
|
SELECT TO_CHAR(t.DEFINE)
|
|
FROM SYS_TRIGGERS t
|
|
JOIN SYS_SCHEMAS s ON s.DB_ID = t.DB_ID AND s.SCHEMA_ID = t.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?) AND UPPER(t.TRIG_NAME) = UPPER(?)`, []any{schema, name}, nil
|
|
case "PROCEDURE", "FUNCTION":
|
|
return `
|
|
SELECT TO_CHAR(p.DEFINE)
|
|
FROM SYS_PROCEDURES p
|
|
JOIN SYS_SCHEMAS s ON s.DB_ID = p.DB_ID AND s.SCHEMA_ID = p.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?) AND UPPER(p.PROC_NAME) = UPPER(?)`, []any{schema, name}, nil
|
|
case "PACKAGE", "PACKAGE BODY":
|
|
return `
|
|
SELECT COALESCE(TO_CHAR(k.SPEC), '') || COALESCE(TO_CHAR(k.BODY), '')
|
|
FROM SYS_PACKAGES k
|
|
JOIN SYS_SCHEMAS s ON s.DB_ID = k.DB_ID AND s.SCHEMA_ID = k.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?) AND UPPER(k.PACK_NAME) = UPPER(?)`, []any{schema, name}, nil
|
|
default:
|
|
return "", nil, fmt.Errorf("object source is not supported for %s", objectType)
|
|
}
|
|
}
|
|
|
|
func (s *server) buildTableDDL(schema, table string) (string, error) {
|
|
columns, err := s.getColumns(schema, table)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if len(columns) == 0 {
|
|
return "", fmt.Errorf("table not found: %s.%s", schema, table)
|
|
}
|
|
var builder strings.Builder
|
|
builder.WriteString("CREATE TABLE ")
|
|
builder.WriteString(quoteIdentifier(schema))
|
|
builder.WriteByte('.')
|
|
builder.WriteString(quoteIdentifier(table))
|
|
builder.WriteString(" (\n")
|
|
for i, column := range columns {
|
|
if i > 0 {
|
|
builder.WriteString(",\n")
|
|
}
|
|
builder.WriteString(" ")
|
|
builder.WriteString(quoteIdentifier(column.Name))
|
|
builder.WriteByte(' ')
|
|
builder.WriteString(columnTypeDDL(column))
|
|
if column.ColumnDefault != nil && strings.TrimSpace(*column.ColumnDefault) != "" {
|
|
builder.WriteString(" DEFAULT ")
|
|
builder.WriteString(strings.TrimSpace(*column.ColumnDefault))
|
|
}
|
|
if !column.IsNullable {
|
|
builder.WriteString(" NOT NULL")
|
|
}
|
|
if column.Comment != nil && strings.TrimSpace(*column.Comment) != "" {
|
|
builder.WriteString(" COMMENT ")
|
|
builder.WriteString(quoteStringLiteral(strings.TrimSpace(*column.Comment)))
|
|
}
|
|
}
|
|
primary := make([]string, 0)
|
|
for _, column := range columns {
|
|
if column.IsPrimaryKey {
|
|
primary = append(primary, quoteIdentifier(column.Name))
|
|
}
|
|
}
|
|
if len(primary) > 0 {
|
|
builder.WriteString(",\n PRIMARY KEY (")
|
|
builder.WriteString(strings.Join(primary, ", "))
|
|
builder.WriteByte(')')
|
|
}
|
|
builder.WriteString("\n)")
|
|
if comment, err := s.tableComment(schema, table); err == nil && strings.TrimSpace(comment) != "" {
|
|
builder.WriteString("\nCOMMENT ")
|
|
builder.WriteString(quoteStringLiteral(strings.TrimSpace(comment)))
|
|
}
|
|
return builder.String(), nil
|
|
}
|
|
|
|
func (s *server) appendTableIndexDDL(schema, table, ddl string) string {
|
|
indexDDL, err := s.tableIndexDDL(schema, table)
|
|
if err == nil && strings.TrimSpace(indexDDL) != "" {
|
|
return appendDDLStatement(ddl, indexDDL)
|
|
}
|
|
indexes, err := s.listIndexes(schema, table)
|
|
if err != nil || len(indexes) == 0 {
|
|
return ddl
|
|
}
|
|
var builder strings.Builder
|
|
for _, index := range indexes {
|
|
if index.IsPrimary || len(index.Columns) == 0 {
|
|
continue
|
|
}
|
|
if builder.Len() > 0 {
|
|
builder.WriteString("\n")
|
|
}
|
|
if index.IsUnique {
|
|
builder.WriteString("CREATE UNIQUE INDEX ")
|
|
} else {
|
|
builder.WriteString("CREATE INDEX ")
|
|
}
|
|
builder.WriteString(quoteIdentifier(index.Name))
|
|
builder.WriteString(" ON ")
|
|
builder.WriteString(quoteIdentifier(schema))
|
|
builder.WriteByte('.')
|
|
builder.WriteString(quoteIdentifier(table))
|
|
builder.WriteByte('(')
|
|
for i, column := range index.Columns {
|
|
if i > 0 {
|
|
builder.WriteString(", ")
|
|
}
|
|
builder.WriteString(quoteIdentifier(column))
|
|
}
|
|
builder.WriteByte(')')
|
|
if index.IndexType != nil && strings.TrimSpace(*index.IndexType) != "" {
|
|
builder.WriteString(" INDEXTYPE IS ")
|
|
builder.WriteString(strings.TrimSpace(*index.IndexType))
|
|
}
|
|
builder.WriteByte(';')
|
|
}
|
|
if builder.Len() == 0 {
|
|
return ddl
|
|
}
|
|
return appendDDLStatement(ddl, builder.String())
|
|
}
|
|
|
|
func (s *server) tableIndexDDL(schema, table string) (string, error) {
|
|
rows, err := s.queryRows(
|
|
"SELECT TO_CHAR(DBMS_METADATA.GET_DDL('INDEX', ?, ?)) FROM DUAL",
|
|
[]any{strings.ToUpper(strings.TrimSpace(table)), schema},
|
|
)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer rows.Close()
|
|
var ddl string
|
|
if rows.Next() {
|
|
if err := rows.Scan(&ddl); err != nil {
|
|
return "", err
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return "", err
|
|
}
|
|
return ddl, nil
|
|
}
|
|
|
|
func (s *server) tableComment(schema, table string) (string, error) {
|
|
rows, err := s.queryRows(`
|
|
SELECT t.COMMENTS
|
|
FROM ALL_TABLES t
|
|
JOIN ALL_SCHEMAS s ON s.DB_ID = t.DB_ID AND s.SCHEMA_ID = t.SCHEMA_ID
|
|
WHERE UPPER(s.SCHEMA_NAME) = UPPER(?)
|
|
AND UPPER(t.TABLE_NAME) = UPPER(?)`, []any{schema, table})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer rows.Close()
|
|
var comment *string
|
|
if rows.Next() {
|
|
if err := rows.Scan(&comment); err != nil {
|
|
return "", err
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return "", err
|
|
}
|
|
if comment == nil {
|
|
return "", nil
|
|
}
|
|
return *comment, nil
|
|
}
|
|
|
|
func appendDDLStatement(ddl, extra string) string {
|
|
ddl = strings.TrimRight(ddl, "\r\n\t ")
|
|
extra = strings.TrimSpace(extra)
|
|
if extra == "" {
|
|
return ddl
|
|
}
|
|
if !strings.HasSuffix(ddl, ";") {
|
|
ddl += ";"
|
|
}
|
|
return ddl + "\n\n" + extra
|
|
}
|
|
|
|
func columnTypeDDL(column columnInfo) string {
|
|
dataType := strings.ToUpper(strings.TrimSpace(column.DataType))
|
|
if column.CharacterMaximumLength != nil {
|
|
return fmt.Sprintf("%s(%d)", dataType, *column.CharacterMaximumLength)
|
|
}
|
|
if column.NumericPrecision != nil && column.NumericScale != nil {
|
|
return fmt.Sprintf("%s(%d,%d)", dataType, *column.NumericPrecision, *column.NumericScale)
|
|
}
|
|
return dataType
|
|
}
|
|
|
|
func decodeXuguScale(dataType string, scale *int) (*int, *int, *int) {
|
|
if scale == nil || *scale < 0 {
|
|
return nil, nil, nil
|
|
}
|
|
upper := strings.ToUpper(dataType)
|
|
if strings.Contains(upper, "CHAR") || strings.Contains(upper, "BINARY") {
|
|
length := *scale
|
|
return nil, nil, &length
|
|
}
|
|
if strings.Contains(upper, "NUM") || strings.Contains(upper, "DECIMAL") {
|
|
precision := *scale / 65536
|
|
numericScale := *scale % 65536
|
|
return &precision, &numericScale, nil
|
|
}
|
|
return nil, nil, nil
|
|
}
|
|
|
|
func normalizeXuguColumnType(dataType string, varying any) string {
|
|
upper := strings.ToUpper(strings.TrimSpace(dataType))
|
|
if !truthy(varying) {
|
|
return dataType
|
|
}
|
|
switch upper {
|
|
case "CHAR":
|
|
return "VARCHAR"
|
|
case "BINARY":
|
|
return "VARBINARY"
|
|
default:
|
|
return dataType
|
|
}
|
|
}
|
|
|
|
var quotedIdentifierRegexp = regexp.MustCompile(`"([^"]+)"`)
|
|
|
|
func parseQuotedIdentifiers(value string) []string {
|
|
matches := quotedIdentifierRegexp.FindAllStringSubmatch(value, -1)
|
|
result := make([]string, 0, len(matches))
|
|
for _, match := range matches {
|
|
if len(match) == 2 {
|
|
result = append(result, match[1])
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
func parseIndexKeys(value string) []string {
|
|
quoted := parseQuotedIdentifiers(value)
|
|
if len(quoted) > 0 {
|
|
return quoted
|
|
}
|
|
parts := strings.Split(value, ",")
|
|
result := make([]string, 0, len(parts))
|
|
for _, part := range parts {
|
|
part = strings.Trim(strings.TrimSpace(part), `"`)
|
|
if part != "" {
|
|
result = append(result, part)
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
func parseForeignKeyColumns(define string) ([]string, []string) {
|
|
groups := regexp.MustCompile(`\(([^()]*)\)`).FindAllStringSubmatch(define, -1)
|
|
if len(groups) < 2 {
|
|
columns := parseQuotedIdentifiers(define)
|
|
if len(columns)%2 == 0 {
|
|
mid := len(columns) / 2
|
|
return columns[:mid], columns[mid:]
|
|
}
|
|
return columns, nil
|
|
}
|
|
return parseQuotedIdentifiers(groups[0][1]), parseQuotedIdentifiers(groups[1][1])
|
|
}
|
|
|
|
func truthy(value any) bool {
|
|
switch v := normalizeValue(value).(type) {
|
|
case bool:
|
|
return v
|
|
case int64:
|
|
return v != 0
|
|
case float64:
|
|
return v != 0
|
|
case string:
|
|
upper := strings.ToUpper(strings.TrimSpace(v))
|
|
return upper == "T" || upper == "TRUE" || upper == "1" || upper == "Y" || upper == "YES"
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func stringPtr(value string) *string {
|
|
if strings.TrimSpace(value) == "" {
|
|
return nil
|
|
}
|
|
return &value
|
|
}
|
|
|
|
func quoteStringLiteral(value string) string {
|
|
return "'" + strings.ReplaceAll(value, "'", "''") + "'"
|
|
}
|
|
|
|
func indexTypeName(value any) string {
|
|
switch fmt.Sprint(normalizeValue(value)) {
|
|
case "0":
|
|
return "BTREE"
|
|
case "1":
|
|
return "RTREE"
|
|
case "2":
|
|
return "FULLTEXT"
|
|
case "3":
|
|
return "BITMAP"
|
|
default:
|
|
return fmt.Sprint(normalizeValue(value))
|
|
}
|
|
}
|
|
|
|
func triggerEventName(value any) string {
|
|
switch fmt.Sprint(normalizeValue(value)) {
|
|
case "1":
|
|
return "INSERT"
|
|
case "2":
|
|
return "UPDATE"
|
|
case "3":
|
|
return "INSERT OR UPDATE"
|
|
case "4":
|
|
return "DELETE"
|
|
case "5":
|
|
return "INSERT OR DELETE"
|
|
case "6":
|
|
return "UPDATE OR DELETE"
|
|
case "7":
|
|
return "INSERT OR UPDATE OR DELETE"
|
|
case "8":
|
|
return "LOGON"
|
|
default:
|
|
return fmt.Sprint(normalizeValue(value))
|
|
}
|
|
}
|
|
|
|
func triggerTimingName(value any) string {
|
|
switch fmt.Sprint(normalizeValue(value)) {
|
|
case "1":
|
|
return "BEFORE"
|
|
case "2":
|
|
return "INSTEAD"
|
|
case "4":
|
|
return "AFTER"
|
|
default:
|
|
return fmt.Sprint(normalizeValue(value))
|
|
}
|
|
}
|
|
|
|
func joinValues(values []any, sep string) string {
|
|
parts := make([]string, len(values))
|
|
for i, value := range values {
|
|
if value == nil {
|
|
parts[i] = ""
|
|
} else {
|
|
parts[i] = fmt.Sprint(value)
|
|
}
|
|
}
|
|
return strings.Join(parts, sep)
|
|
}
|
|
|
|
func decodeParams(params map[string]json.RawMessage, target any) error {
|
|
if params == nil {
|
|
params = map[string]json.RawMessage{}
|
|
}
|
|
data, err := json.Marshal(params)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return json.Unmarshal(data, target)
|
|
}
|
|
|
|
func stringParam(params map[string]json.RawMessage, key string) string {
|
|
if params == nil || len(params[key]) == 0 {
|
|
return ""
|
|
}
|
|
var value string
|
|
_ = json.Unmarshal(params[key], &value)
|
|
return value
|
|
}
|
|
|
|
func intParam(params map[string]json.RawMessage, key string) int {
|
|
if params == nil || len(params[key]) == 0 {
|
|
return 0
|
|
}
|
|
var value int
|
|
_ = json.Unmarshal(params[key], &value)
|
|
return value
|
|
}
|
|
|
|
func errorResponse(id json.RawMessage, err error) response {
|
|
return response{JSONRPC: "2.0", ID: id, Error: &rpcError{Code: -1, Message: err.Error()}}
|
|
}
|
|
|
|
func trimStatementSQL(sqlText string) string {
|
|
return strings.TrimRight(strings.TrimSpace(sqlText), "; \t\r\n")
|
|
}
|
|
|
|
func isQuerySQL(sqlText string) bool {
|
|
lower := strings.ToLower(strings.TrimSpace(sqlText))
|
|
return strings.HasPrefix(lower, "select") || strings.HasPrefix(lower, "with")
|
|
}
|
|
|
|
func quoteIdentifier(value string) string {
|
|
return `"` + strings.ReplaceAll(value, `"`, `""`) + `"`
|
|
}
|
|
|
|
func normalizeValue(value any) any {
|
|
switch v := value.(type) {
|
|
case nil:
|
|
return nil
|
|
case []byte:
|
|
return string(v)
|
|
case time.Time:
|
|
return v.Format(time.RFC3339Nano)
|
|
case int:
|
|
return int64(v)
|
|
case int8:
|
|
return int64(v)
|
|
case int16:
|
|
return int64(v)
|
|
case int32:
|
|
return int64(v)
|
|
case int64:
|
|
return v
|
|
case uint:
|
|
return uint64(v)
|
|
case uint8:
|
|
return uint64(v)
|
|
case uint16:
|
|
return uint64(v)
|
|
case uint32:
|
|
return uint64(v)
|
|
case uint64:
|
|
return v
|
|
case float32:
|
|
return float64(v)
|
|
case float64, bool, string:
|
|
return v
|
|
case fmt.Stringer:
|
|
return v.String()
|
|
default:
|
|
return fmt.Sprint(v)
|
|
}
|
|
}
|
|
|
|
func emptyIfNil[T any](values []T) []T {
|
|
if values == nil {
|
|
return []T{}
|
|
}
|
|
return values
|
|
}
|