mirror of
https://github.com/Syngnat/GoNavi.git
synced 2026-05-28 13:29:37 +08:00
✨ feat(starrocks): 新增 StarRocks 数据源与高级对象能力
- 后端接入:新增独立 starrocks 可选驱动,复用 MySQL wire 协议并支持默认 9030 端口 - 驱动管理:补齐 manifest、build tag、revision、driver-agent provider 和构建脚本 - 前端接入:新增 StarRocks 连接类型、图标、能力矩阵、URI 解析、保存回显和 SQL 自动 LIMIT - 方言增强:新增 StarRocks 类型、关键字、函数补全和专属建表 SQL 生成 - 高级对象:支持物化视图对象浏览、Rollup 模板、外部 Catalog 模板和高级表设计器参数 - CI 发布:将 StarRocks driver-agent 纳入 dev/release 构建与 release 资产校验
This commit is contained in:
@@ -6,6 +6,7 @@ func registerOptionalDatabaseFactories() {
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("mariadb"), "mariadb")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("oceanbase"), "oceanbase")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("diros"), "diros", "doris")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("starrocks"), "starrocks")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("sphinx"), "sphinx")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("sqlserver"), "sqlserver")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("sqlite"), "sqlite")
|
||||
|
||||
@@ -6,6 +6,7 @@ func registerOptionalDatabaseFactories() {
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("mariadb"), "mariadb")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("oceanbase"), "oceanbase")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("diros"), "diros", "doris")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("starrocks"), "starrocks")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("sphinx"), "sphinx")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("sqlserver"), "sqlserver")
|
||||
registerDatabaseFactory(newOptionalDriverAgentDatabase("sqlite"), "sqlite")
|
||||
|
||||
@@ -7,6 +7,7 @@ func init() {
|
||||
"mariadb": "src-4e1ec648c70c87ea",
|
||||
"oceanbase": "src-8e445fc4899d850f",
|
||||
"diros": "src-74927b3809258666",
|
||||
"starrocks": "src-3b5aad8a32f79b61",
|
||||
"sphinx": "src-269bd60a34df47d3",
|
||||
"sqlserver": "src-84553484c72e7253",
|
||||
"sqlite": "src-762863d48f653b89",
|
||||
|
||||
@@ -24,6 +24,7 @@ var optionalGoDrivers = map[string]struct{}{
|
||||
"mariadb": {},
|
||||
"oceanbase": {},
|
||||
"diros": {},
|
||||
"starrocks": {},
|
||||
"sphinx": {},
|
||||
"sqlserver": {},
|
||||
"sqlite": {},
|
||||
@@ -78,6 +79,8 @@ func driverDisplayName(driverType string) string {
|
||||
return "OceanBase"
|
||||
case "diros":
|
||||
return "Doris"
|
||||
case "starrocks":
|
||||
return "StarRocks"
|
||||
case "sphinx":
|
||||
return "Sphinx"
|
||||
case "postgres":
|
||||
|
||||
@@ -113,7 +113,7 @@ func TestNewCompatibleDriversAreOptionalAgentDrivers(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
SetExternalDriverDownloadDirectory(tmpDir)
|
||||
|
||||
for _, driverType := range []string{"oceanbase", "opengauss", "open_gauss"} {
|
||||
for _, driverType := range []string{"oceanbase", "opengauss", "open_gauss", "starrocks"} {
|
||||
if IsBuiltinDriver(driverType) {
|
||||
t.Fatalf("%s 不应是免安装内置驱动", driverType)
|
||||
}
|
||||
|
||||
210
internal/db/starrocks_impl.go
Normal file
210
internal/db/starrocks_impl.go
Normal file
@@ -0,0 +1,210 @@
|
||||
//go:build gonavi_full_drivers || gonavi_starrocks_driver
|
||||
|
||||
package db
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"GoNavi-Wails/internal/connection"
|
||||
"GoNavi-Wails/internal/ssh"
|
||||
"GoNavi-Wails/internal/utils"
|
||||
|
||||
mysqlDriver "github.com/go-sql-driver/mysql"
|
||||
)
|
||||
|
||||
const (
|
||||
starRocksDriverName = "starrocks"
|
||||
defaultStarRocksPort = 9030
|
||||
)
|
||||
|
||||
// StarRocksDB 使用独立 driver 名称接入,底层协议兼容 MySQL。
|
||||
type StarRocksDB struct {
|
||||
MySQLDB
|
||||
}
|
||||
|
||||
func init() {
|
||||
for _, name := range sql.Drivers() {
|
||||
if name == starRocksDriverName {
|
||||
return
|
||||
}
|
||||
}
|
||||
sql.Register(starRocksDriverName, &mysqlDriver.MySQLDriver{})
|
||||
}
|
||||
|
||||
func applyStarRocksURI(config connection.ConnectionConfig) connection.ConnectionConfig {
|
||||
uriText := strings.TrimSpace(config.URI)
|
||||
if uriText == "" {
|
||||
return config
|
||||
}
|
||||
|
||||
parsed, ok := parseMySQLCompatibleURI(uriText, "starrocks", "mysql")
|
||||
if !ok {
|
||||
return config
|
||||
}
|
||||
|
||||
if parsed.User != nil {
|
||||
if config.User == "" {
|
||||
config.User = parsed.User.Username()
|
||||
}
|
||||
if pass, ok := parsed.User.Password(); ok && config.Password == "" {
|
||||
config.Password = pass
|
||||
}
|
||||
}
|
||||
|
||||
if dbName := strings.TrimPrefix(parsed.Path, "/"); dbName != "" && config.Database == "" {
|
||||
config.Database = dbName
|
||||
}
|
||||
|
||||
defaultPort := config.Port
|
||||
if defaultPort <= 0 {
|
||||
defaultPort = defaultStarRocksPort
|
||||
}
|
||||
|
||||
hostsFromURI := make([]string, 0, 4)
|
||||
hostText := strings.TrimSpace(parsed.Host)
|
||||
if hostText != "" {
|
||||
for _, entry := range strings.Split(hostText, ",") {
|
||||
host, port, ok := parseHostPortWithDefault(entry, defaultPort)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
hostsFromURI = append(hostsFromURI, normalizeMySQLAddress(host, port))
|
||||
}
|
||||
}
|
||||
|
||||
if len(config.Hosts) == 0 && len(hostsFromURI) > 0 {
|
||||
config.Hosts = hostsFromURI
|
||||
}
|
||||
if strings.TrimSpace(config.Host) == "" && len(hostsFromURI) > 0 {
|
||||
host, port, ok := parseHostPortWithDefault(hostsFromURI[0], defaultPort)
|
||||
if ok {
|
||||
config.Host = host
|
||||
config.Port = port
|
||||
}
|
||||
}
|
||||
|
||||
if config.Topology == "" {
|
||||
topology := strings.TrimSpace(parsed.Query().Get("topology"))
|
||||
if topology != "" {
|
||||
config.Topology = strings.ToLower(topology)
|
||||
}
|
||||
}
|
||||
|
||||
return config
|
||||
}
|
||||
|
||||
func collectStarRocksAddresses(config connection.ConnectionConfig) []string {
|
||||
defaultPort := config.Port
|
||||
if defaultPort <= 0 {
|
||||
defaultPort = defaultStarRocksPort
|
||||
}
|
||||
|
||||
candidates := make([]string, 0, len(config.Hosts)+1)
|
||||
if len(config.Hosts) > 0 {
|
||||
candidates = append(candidates, config.Hosts...)
|
||||
} else {
|
||||
candidates = append(candidates, normalizeMySQLAddress(config.Host, defaultPort))
|
||||
}
|
||||
|
||||
result := make([]string, 0, len(candidates))
|
||||
seen := make(map[string]struct{}, len(candidates))
|
||||
for _, entry := range candidates {
|
||||
host, port, ok := parseHostPortWithDefault(entry, defaultPort)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
normalized := normalizeMySQLAddress(host, port)
|
||||
if _, exists := seen[normalized]; exists {
|
||||
continue
|
||||
}
|
||||
seen[normalized] = struct{}{}
|
||||
result = append(result, normalized)
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
func (s *StarRocksDB) getDSN(config connection.ConnectionConfig) (string, error) {
|
||||
database := config.Database
|
||||
protocol := "tcp"
|
||||
address := normalizeMySQLAddress(config.Host, config.Port)
|
||||
|
||||
if config.UseSSH {
|
||||
netName, err := ssh.RegisterSSHNetwork(config.SSH)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("创建 SSH 隧道失败:%w", err)
|
||||
}
|
||||
protocol = netName
|
||||
}
|
||||
|
||||
return buildMySQLCompatibleDSN(config, protocol, address, database), nil
|
||||
}
|
||||
|
||||
func resolveStarRocksCredential(config connection.ConnectionConfig, addressIndex int) (string, string) {
|
||||
primaryUser := strings.TrimSpace(config.User)
|
||||
primaryPassword := config.Password
|
||||
replicaUser := strings.TrimSpace(config.MySQLReplicaUser)
|
||||
replicaPassword := config.MySQLReplicaPassword
|
||||
|
||||
if addressIndex > 0 && replicaUser != "" {
|
||||
return replicaUser, replicaPassword
|
||||
}
|
||||
|
||||
if primaryUser == "" && replicaUser != "" {
|
||||
return replicaUser, replicaPassword
|
||||
}
|
||||
|
||||
return config.User, primaryPassword
|
||||
}
|
||||
|
||||
func (s *StarRocksDB) Connect(config connection.ConnectionConfig) error {
|
||||
runConfig := applyStarRocksURI(config)
|
||||
addresses := collectStarRocksAddresses(runConfig)
|
||||
if len(addresses) == 0 {
|
||||
return fmt.Errorf("连接建立后验证失败:未找到可用的 StarRocks 地址")
|
||||
}
|
||||
|
||||
var errorDetails []string
|
||||
for index, address := range addresses {
|
||||
candidateConfig := runConfig
|
||||
host, port, ok := parseHostPortWithDefault(address, defaultStarRocksPort)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
candidateConfig.Host = host
|
||||
candidateConfig.Port = port
|
||||
candidateConfig.User, candidateConfig.Password = resolveStarRocksCredential(runConfig, index)
|
||||
|
||||
dsn, err := s.getDSN(candidateConfig)
|
||||
if err != nil {
|
||||
errorDetails = append(errorDetails, fmt.Sprintf("%s 生成连接串失败: %v", address, err))
|
||||
continue
|
||||
}
|
||||
db, err := sql.Open(starRocksDriverName, dsn)
|
||||
if err != nil {
|
||||
errorDetails = append(errorDetails, fmt.Sprintf("%s 打开失败: %v", address, err))
|
||||
continue
|
||||
}
|
||||
|
||||
timeout := getConnectTimeout(candidateConfig)
|
||||
ctx, cancel := utils.ContextWithTimeout(timeout)
|
||||
pingErr := db.PingContext(ctx)
|
||||
cancel()
|
||||
if pingErr != nil {
|
||||
_ = db.Close()
|
||||
errorDetails = append(errorDetails, fmt.Sprintf("%s 验证失败: %v", address, pingErr))
|
||||
continue
|
||||
}
|
||||
|
||||
s.conn = db
|
||||
s.pingTimeout = timeout
|
||||
return nil
|
||||
}
|
||||
|
||||
if len(errorDetails) == 0 {
|
||||
return fmt.Errorf("连接建立后验证失败:未找到可用的 StarRocks 地址")
|
||||
}
|
||||
return fmt.Errorf("连接建立后验证失败:%s", strings.Join(errorDetails, ";"))
|
||||
}
|
||||
Reference in New Issue
Block a user