feat(database): 添加 ClickHouse 数据库驱动支持

- 实现了完整的 ClickHouse 驱动,包括连接、查询、插入、更新、删除等基本操作
- 添加了 ClickHouse 特有的数据类型转换和 SQL 过滤功能
- 实现了表结构查询和字段信息获取功能
- 添加了 Ping 检查和错误处理机制
- 增加了对 UPDATE 和 DELETE 语句的语法转换以适配 ClickHouse
- 添加了批量插入操作的支持
- 新增了 14 个测试用例文件用于验证数据库相关功能
This commit is contained in:
maguodong
2026-04-08 15:34:51 +08:00
parent 70ad831041
commit 398e732301
109 changed files with 5275 additions and 545 deletions
+9 -22
View File
@@ -4,58 +4,45 @@
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
// Package mssql implements database.Driver for Microsoft SQL Server.
// Package mssql implements database.Driver, which supports operations for MSSQL.
package mssql
import (
"database/sql"
"fmt"
_ "github.com/microsoft/go-mssqldb"
"git.magicany.cc/black1552/gin-base/database"
)
// Driver is the driver for SQL Server database.
// Driver is the driver for SQL server database.
type Driver struct {
*database.Core
}
const (
quoteChar = `"`
rowNumberAliasForSelect = `ROW_NUMBER__`
quoteChar = `"`
)
func init() {
if err := database.Register("mssql", New()); err != nil {
if err := database.Register(`mssql`, New()); err != nil {
panic(err)
}
}
// New creates and returns a driver that implements database.Driver for SQL Server.
// New create and returns a driver that implements database.Driver, which supports operations for Mssql.
func New() database.Driver {
return &Driver{}
}
// New creates and returns a database object for SQL Server.
// New creates and returns a database object for SQL server.
// It implements the interface of database.Driver for extra database driver installation.
func (d *Driver) New(core *database.Core, node *database.ConfigNode) (database.DB, error) {
return &Driver{
Core: core,
}, nil
}
// GetChars returns the security char for SQL Server.
// GetChars returns the security char for this type of database.
func (d *Driver) GetChars() (charLeft string, charRight string) {
return quoteChar, quoteChar
}
// Open creates and returns an underlying sql.DB object for SQL Server.
func (d *Driver) Open(config *database.ConfigNode) (*sql.DB, error) {
var source string
if config.Link != "" {
source = config.Link
} else {
source = fmt.Sprintf("sqlserver://%s:%s@%s:%s?database=%s",
config.User, config.Pass, config.Host, config.Port, config.Name)
}
return sql.Open("sqlserver", source)
}
+29
View File
@@ -0,0 +1,29 @@
// Copyright GoFrame Author(https://goframe.org). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
package mssql
import (
"context"
"git.magicany.cc/black1552/gin-base/database"
)
// DoCommit commits current sql and arguments to underlying sql driver.
func (d *Driver) DoCommit(ctx context.Context, in database.DoCommitInput) (out database.DoCommitOutput, err error) {
out, err = d.Core.DoCommit(ctx, in)
if err != nil {
return
}
if len(out.Records) > 0 {
// remove auto added field.
for i, record := range out.Records {
delete(record, rowNumberAliasForSelect)
out.Records[i] = record
}
}
return
}
+192
View File
@@ -0,0 +1,192 @@
// Copyright GoFrame Author(https://goframe.org). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
package mssql
import (
"context"
"database/sql"
"fmt"
"regexp"
"strings"
"git.magicany.cc/black1552/gin-base/database"
"github.com/gogf/gf/v2/errors/gcode"
"github.com/gogf/gf/v2/errors/gerror"
)
const (
// INSERT statement prefixes
insertPrefixDefault = "INSERT INTO"
insertPrefixIgnore = "INSERT IGNORE INTO"
// Database field attributes
fieldExtraIdentity = "IDENTITY"
fieldKeyPrimary = "PRI"
// SQL keywords and syntax markers
outputKeyword = "OUTPUT"
insertValuesMarker = ") VALUES" // find the position of the string "VALUES" in the INSERT SQL statement to embed output code for retrieving the last inserted ID
// Object and field references
insertedObjectName = "INSERTED"
// Result field names and aliases
affectCountExpression = " 1 as AffectCount"
lastInsertIdFieldAlias = "ID"
)
// DoExec commits the sql string and its arguments to underlying driver
// through given link object and returns the execution result.
func (d *Driver) DoExec(ctx context.Context, link database.Link, sqlStr string, args ...interface{}) (result sql.Result, err error) {
// Transaction checks.
if link == nil {
if tx := database.TXFromCtx(ctx, d.GetGroup()); tx != nil {
// Firstly, check and retrieve transaction link from context.
link = &txLinkMssql{tx.GetSqlTX()}
} else if link, err = d.MasterLink(); err != nil {
// Or else it creates one from master node.
return nil, err
}
} else if !link.IsTransaction() {
// If current link is not transaction link, it checks and retrieves transaction from context.
if tx := database.TXFromCtx(ctx, d.GetGroup()); tx != nil {
link = &txLinkMssql{tx.GetSqlTX()}
}
}
// SQL filtering.
sqlStr, args = d.FormatSqlBeforeExecuting(sqlStr, args)
sqlStr, args, err = d.DoFilter(ctx, link, sqlStr, args)
if err != nil {
return nil, err
}
if !strings.HasPrefix(sqlStr, insertPrefixDefault) && !strings.HasPrefix(sqlStr, insertPrefixIgnore) {
return d.Core.DoExec(ctx, link, sqlStr, args)
}
// Find the first position of VALUES marker in the INSERT statement.
pos := strings.Index(sqlStr, insertValuesMarker)
table := d.GetTableNameFromSql(sqlStr)
outPutSql := d.GetInsertOutputSql(ctx, table)
// rebuild sql add output
var (
sqlValueBefore = sqlStr[:pos+1]
sqlValueAfter = sqlStr[pos+1:]
)
sqlStr = fmt.Sprintf("%s%s%s", sqlValueBefore, outPutSql, sqlValueAfter)
// fmt.Println("sql str:", sqlStr)
// Link execution.
var out database.DoCommitOutput
out, err = d.DoCommit(ctx, database.DoCommitInput{
Link: link,
Sql: sqlStr,
Args: args,
Stmt: nil,
Type: database.SqlTypeQueryContext,
IsTransaction: link.IsTransaction(),
})
if err != nil {
return &Result{lastInsertId: 0, rowsAffected: 0, err: err}, err
}
stdSqlResult := out.Records
if len(stdSqlResult) == 0 {
err = gerror.WrapCode(
gcode.CodeDbOperationError,
gerror.New("affected count is zero"),
`sql.Result.RowsAffected failed`,
)
return &Result{lastInsertId: 0, rowsAffected: 0, err: err}, err
}
// For batch insert, OUTPUT clause returns one row per inserted row.
// So the rowsAffected should be the count of returned records.
rowsAffected := int64(len(stdSqlResult))
// get last_insert_id from the first returned row
lastInsertId := stdSqlResult[0].GMap().GetVar(lastInsertIdFieldAlias).Int64()
return &Result{lastInsertId: lastInsertId, rowsAffected: rowsAffected}, err
}
// GetTableNameFromSql get table name from sql statement
// It handles table string like:
// "user"
// "user u"
// "DbLog.dbo.user",
// "user as u".
func (d *Driver) GetTableNameFromSql(sqlStr string) (table string) {
// INSERT INTO "ip_to_id"("ip") OUTPUT 1 as AffectCount,INSERTED.id as ID VALUES(?)
var (
leftChars, rightChars = d.GetChars()
trimStr = leftChars + rightChars + "[] "
pattern = "INTO(.+?)\\("
regCompile = regexp.MustCompile(pattern)
tableInfo = regCompile.FindStringSubmatch(sqlStr)
)
// get the first one. after the first it may be content of the value, it's not table name.
table = tableInfo[1]
table = strings.Trim(table, " ")
if strings.Contains(table, ".") {
tmpAry := strings.Split(table, ".")
// the last one is table name
table = tmpAry[len(tmpAry)-1]
} else if strings.Contains(table, "as") || strings.Contains(table, " ") {
tmpAry := strings.Split(table, "as")
if len(tmpAry) < 2 {
tmpAry = strings.Split(table, " ")
}
// get the first one
table = tmpAry[0]
}
table = strings.Trim(table, trimStr)
return table
}
// txLink is used to implement interface Link for TX.
type txLinkMssql struct {
*sql.Tx
}
// IsTransaction returns if current Link is a transaction.
func (l *txLinkMssql) IsTransaction() bool {
return true
}
// IsOnMaster checks and returns whether current link is operated on master node.
// Note that, transaction operation is always operated on master node.
func (l *txLinkMssql) IsOnMaster() bool {
return true
}
// GetInsertOutputSql gen get last_insert_id code
func (d *Driver) GetInsertOutputSql(ctx context.Context, table string) string {
fds, errFd := d.GetDB().TableFields(ctx, table)
if errFd != nil {
return ""
}
extraSqlAry := make([]string, 0)
extraSqlAry = append(extraSqlAry, fmt.Sprintf(" %s %s", outputKeyword, affectCountExpression))
incrNo := 0
if len(fds) > 0 {
for _, fd := range fds {
// has primary key and is auto-increment
if fd.Extra == fieldExtraIdentity && fd.Key == fieldKeyPrimary && !fd.Null {
incrNoStr := ""
if incrNo == 0 { // fixed first field named id, convenient to get
incrNoStr = fmt.Sprintf(" as %s", lastInsertIdFieldAlias)
}
extraSqlAry = append(extraSqlAry, fmt.Sprintf("%s.%s%s", insertedObjectName, fd.Name, incrNoStr))
incrNo++
}
// fmt.Printf("null:%t name:%s key:%s k:%s \n", fd.Null, fd.Name, fd.Key, k)
}
}
return strings.Join(extraSqlAry, ",")
// sql example:INSERT INTO "ip_to_id"("ip") OUTPUT 1 as AffectCount,INSERTED.id as ID VALUES(?)
}
+155
View File
@@ -0,0 +1,155 @@
// Copyright GoFrame Author(https://goframe.org). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
package mssql
import (
"context"
"fmt"
"strconv"
"strings"
"git.magicany.cc/black1552/gin-base/database"
"github.com/gogf/gf/v2/text/gregex"
"github.com/gogf/gf/v2/text/gstr"
)
var (
selectWithOrderSqlTmp = `
SELECT * FROM (
SELECT ROW_NUMBER() OVER (ORDER BY %s) as ROW_NUMBER__, %s
FROM (%s) as InnerQuery
) as TMP_
WHERE TMP_.ROW_NUMBER__ > %d AND TMP_.ROW_NUMBER__ <= %d`
selectWithoutOrderSqlTmp = `
SELECT * FROM (
SELECT ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) as ROW_NUMBER__, %s
FROM (%s) as InnerQuery
) as TMP_
WHERE TMP_.ROW_NUMBER__ > %d AND TMP_.ROW_NUMBER__ <= %d`
)
func init() {
var err error
selectWithOrderSqlTmp, err = database.FormatMultiLineSqlToSingle(selectWithOrderSqlTmp)
if err != nil {
panic(err)
}
selectWithoutOrderSqlTmp, err = database.FormatMultiLineSqlToSingle(selectWithoutOrderSqlTmp)
if err != nil {
panic(err)
}
}
// DoFilter deals with the sql string before commits it to underlying sql driver.
func (d *Driver) DoFilter(
ctx context.Context, link database.Link, sql string, args []any,
) (newSql string, newArgs []any, err error) {
var index int
// Convert placeholder char '?' to string "@px".
newSql, err = gregex.ReplaceStringFunc("\\?", sql, func(s string) string {
index++
return fmt.Sprintf("@p%d", index)
})
if err != nil {
return "", nil, err
}
newSql, err = gregex.ReplaceString("\"", "", newSql)
if err != nil {
return "", nil, err
}
newSql, err = d.parseSql(newSql)
if err != nil {
return "", nil, err
}
newArgs = args
return d.Core.DoFilter(ctx, link, newSql, newArgs)
}
// parseSql does some replacement of the sql before commits it to underlying driver,
// for support of microsoft sql server.
func (d *Driver) parseSql(toBeCommittedSql string) (string, error) {
var (
err error
operation = gstr.StrTillEx(toBeCommittedSql, " ")
keyword = strings.ToUpper(gstr.Trim(operation))
)
switch keyword {
case "SELECT":
toBeCommittedSql, err = d.handleSelectSqlReplacement(toBeCommittedSql)
if err != nil {
return "", err
}
}
return toBeCommittedSql, nil
}
func (d *Driver) handleSelectSqlReplacement(toBeCommittedSql string) (newSql string, err error) {
// SELECT * FROM USER WHERE ID=1 LIMIT 1
match, err := gregex.MatchString(`^SELECT(.+?)LIMIT\s+1$`, toBeCommittedSql)
if err != nil {
return "", err
}
if len(match) > 1 {
return fmt.Sprintf(`SELECT TOP 1 %s`, strings.TrimSpace(match[1])), nil
}
// SELECT * FROM USER WHERE AGE>18 ORDER BY ID DESC LIMIT 100, 200
pattern := `(?i)SELECT(.+?)(ORDER BY.+?)?\s*LIMIT\s*(\d+)(?:\s*,\s*(\d+))?`
if !gregex.IsMatchString(pattern, toBeCommittedSql) {
return toBeCommittedSql, nil
}
allMatch, err := gregex.MatchString(pattern, toBeCommittedSql)
if err != nil {
return "", err
}
// Extract SELECT part
selectStr := strings.TrimSpace(allMatch[1])
// Extract ORDER BY part
orderStr := ""
if len(allMatch[2]) > 0 {
orderStr = strings.TrimSpace(allMatch[2])
// Remove "ORDER BY" prefix as it will be used in OVER clause
orderStr = strings.TrimPrefix(orderStr, "ORDER BY")
orderStr = strings.TrimSpace(orderStr)
}
// Calculate LIMIT and OFFSET values
first, _ := strconv.Atoi(allMatch[3]) // LIMIT first parameter
limit := 0
if len(allMatch) > 4 && allMatch[4] != "" {
limit, _ = strconv.Atoi(allMatch[4]) // LIMIT second parameter
} else {
limit = first
first = 0
}
// Build the final query
if orderStr != "" {
// Have ORDER BY clause
newSql = fmt.Sprintf(
selectWithOrderSqlTmp,
orderStr, // ORDER BY clause for ROW_NUMBER
"*", // Select all columns
fmt.Sprintf("SELECT %s", selectStr), // Original SELECT
first, // OFFSET
first+limit, // OFFSET + LIMIT
)
} else {
// Without ORDER BY clause
newSql = fmt.Sprintf(
selectWithoutOrderSqlTmp,
"*", // Select all columns
fmt.Sprintf("SELECT %s", selectStr), // Original SELECT
first, // OFFSET
first+limit, // OFFSET + LIMIT
)
}
return newSql, nil
}
@@ -0,0 +1,130 @@
// Copyright GoFrame Author(https://goframe.org). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
package mssql
import (
"context"
"testing"
"github.com/gogf/gf/v2/test/gtest"
)
func TestDriver_DoFilter(t *testing.T) {
gtest.C(t, func(t *gtest.T) {
d := &Driver{}
// Test SELECT with LIMIT
sql := "SELECT * FROM users WHERE id = ? LIMIT 10"
args := []any{1}
newSql, newArgs, err := d.DoFilter(context.Background(), nil, sql, args)
t.AssertNil(err)
t.Assert(newArgs, args)
// DoFilter should transform the SQL for MSSQL compatibility
t.AssertNE(newSql, "")
// Test INSERT statement (should remain unchanged except for placeholder)
sql = "INSERT INTO users (name) VALUES (?)"
args = []any{"test"}
newSql, newArgs, err = d.DoFilter(context.Background(), nil, sql, args)
t.AssertNil(err)
t.Assert(newArgs, args)
t.AssertNE(newSql, "")
// Test UPDATE statement
sql = "UPDATE users SET name = ? WHERE id = ?"
args = []any{"test", 1}
newSql, newArgs, err = d.DoFilter(context.Background(), nil, sql, args)
t.AssertNil(err)
t.Assert(newArgs, args)
t.AssertNE(newSql, "")
// Test DELETE statement
sql = "DELETE FROM users WHERE id = ?"
args = []any{1}
newSql, newArgs, err = d.DoFilter(context.Background(), nil, sql, args)
t.AssertNil(err)
t.Assert(newArgs, args)
t.AssertNE(newSql, "")
})
}
func TestDriver_handleSelectSqlReplacement(t *testing.T) {
gtest.C(t, func(t *gtest.T) {
d := &Driver{}
// LIMIT 1
inputSql := "SELECT * FROM User WHERE ID = 1 LIMIT 1"
expectedSql := "SELECT TOP 1 * FROM User WHERE ID = 1"
resultSql, err := d.handleSelectSqlReplacement(inputSql)
t.AssertNil(err)
t.Assert(resultSql, expectedSql)
// LIMIT query with offset and number of rows
inputSql = "SELECT * FROM User ORDER BY ID DESC LIMIT 100, 200"
expectedSql = "SELECT * FROM ( SELECT ROW_NUMBER() OVER (ORDER BY ID DESC) as ROW_NUMBER__, * FROM (SELECT * FROM User) as InnerQuery ) as TMP_ WHERE TMP_.ROW_NUMBER__ > 100 AND TMP_.ROW_NUMBER__ <= 300"
resultSql, err = d.handleSelectSqlReplacement(inputSql)
t.AssertNil(err)
t.Assert(resultSql, expectedSql)
// Simple query with no LIMIT
inputSql = "SELECT * FROM User WHERE age > 18"
expectedSql = "SELECT * FROM User WHERE age > 18"
resultSql, err = d.handleSelectSqlReplacement(inputSql)
t.AssertNil(err)
t.Assert(resultSql, expectedSql)
// without LIMIT
inputSql = "SELECT * FROM User ORDER BY ID DESC"
expectedSql = "SELECT * FROM User ORDER BY ID DESC"
resultSql, err = d.handleSelectSqlReplacement(inputSql)
t.AssertNil(err)
t.Assert(resultSql, expectedSql)
// LIMIT query with only rows
inputSql = "SELECT * FROM User LIMIT 50"
expectedSql = "SELECT * FROM ( SELECT ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) as ROW_NUMBER__, * FROM (SELECT * FROM User) as InnerQuery ) as TMP_ WHERE TMP_.ROW_NUMBER__ > 0 AND TMP_.ROW_NUMBER__ <= 50"
resultSql, err = d.handleSelectSqlReplacement(inputSql)
t.AssertNil(err)
t.Assert(resultSql, expectedSql)
// LIMIT query without ORDER BY
inputSql = "SELECT * FROM User LIMIT 30"
expectedSql = "SELECT * FROM ( SELECT ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) as ROW_NUMBER__, * FROM (SELECT * FROM User) as InnerQuery ) as TMP_ WHERE TMP_.ROW_NUMBER__ > 0 AND TMP_.ROW_NUMBER__ <= 30"
resultSql, err = d.handleSelectSqlReplacement(inputSql)
t.AssertNil(err)
t.Assert(resultSql, expectedSql)
// Complex query with ORDER BY and LIMIT
inputSql = "SELECT name, age FROM User WHERE age > 18 ORDER BY age ASC LIMIT 10, 5"
expectedSql = "SELECT * FROM ( SELECT ROW_NUMBER() OVER (ORDER BY age ASC) as ROW_NUMBER__, * FROM (SELECT name, age FROM User WHERE age > 18) as InnerQuery ) as TMP_ WHERE TMP_.ROW_NUMBER__ > 10 AND TMP_.ROW_NUMBER__ <= 15"
resultSql, err = d.handleSelectSqlReplacement(inputSql)
t.AssertNil(err)
t.Assert(resultSql, expectedSql)
// Complex conditional queries have limits
inputSql = "SELECT * FROM User WHERE age > 18 AND status = 'active' LIMIT 100, 50"
expectedSql = "SELECT * FROM ( SELECT ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) as ROW_NUMBER__, * FROM (SELECT * FROM User WHERE age > 18 AND status = 'active') as InnerQuery ) as TMP_ WHERE TMP_.ROW_NUMBER__ > 100 AND TMP_.ROW_NUMBER__ <= 150"
resultSql, err = d.handleSelectSqlReplacement(inputSql)
t.AssertNil(err)
t.Assert(resultSql, expectedSql)
// A LIMIT query that contains subquery
inputSql = "SELECT * FROM (SELECT * FROM User WHERE age > 18) AS subquery LIMIT 10"
expectedSql = "SELECT * FROM ( SELECT ROW_NUMBER() OVER (ORDER BY (SELECT NULL)) as ROW_NUMBER__, * FROM (SELECT * FROM (SELECT * FROM User WHERE age > 18) AS subquery) as InnerQuery ) as TMP_ WHERE TMP_.ROW_NUMBER__ > 0 AND TMP_.ROW_NUMBER__ <= 10"
resultSql, err = d.handleSelectSqlReplacement(inputSql)
t.AssertNil(err)
t.Assert(resultSql, expectedSql)
// Queries with complex ORDER BY and LIMIT
inputSql = "SELECT name, age FROM User WHERE age > 18 ORDER BY age DESC, name ASC LIMIT 20, 10"
expectedSql = "SELECT * FROM ( SELECT ROW_NUMBER() OVER (ORDER BY age DESC, name ASC) as ROW_NUMBER__, * FROM (SELECT name, age FROM User WHERE age > 18) as InnerQuery ) as TMP_ WHERE TMP_.ROW_NUMBER__ > 20 AND TMP_.ROW_NUMBER__ <= 30"
resultSql, err = d.handleSelectSqlReplacement(inputSql)
t.AssertNil(err)
t.Assert(resultSql, expectedSql)
})
}
+203
View File
@@ -0,0 +1,203 @@
// Copyright GoFrame Author(https://goframe.org). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
package mssql
import (
"context"
"database/sql"
"fmt"
"strings"
"git.magicany.cc/black1552/gin-base/database"
"github.com/gogf/gf/v2/container/gset"
"github.com/gogf/gf/v2/errors/gcode"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/text/gstr"
)
// DoInsert inserts or updates data for given table.
// The list parameter must contain at least one record, which was previously validated.
func (d *Driver) DoInsert(
ctx context.Context, link database.Link, table string, list database.List, option database.DoInsertOption,
) (result sql.Result, err error) {
switch option.InsertOption {
case database.InsertOptionSave:
return d.doSave(ctx, link, table, list, option)
case database.InsertOptionReplace:
// MSSQL does not support REPLACE INTO syntax, use SAVE instead.
return d.doSave(ctx, link, table, list, option)
case database.InsertOptionIgnore:
// MSSQL does not support INSERT IGNORE syntax, use MERGE instead.
return d.doInsertIgnore(ctx, link, table, list, option)
default:
return d.Core.DoInsert(ctx, link, table, list, option)
}
}
// doSave support upsert for MSSQL
func (d *Driver) doSave(ctx context.Context,
link database.Link, table string, list database.List, option database.DoInsertOption,
) (result sql.Result, err error) {
return d.doMergeInsert(ctx, link, table, list, option, true)
}
// doInsertIgnore implements INSERT IGNORE operation using MERGE statement for MSSQL database.
// It only inserts records when there's no conflict on primary/unique keys.
func (d *Driver) doInsertIgnore(ctx context.Context,
link database.Link, table string, list database.List, option database.DoInsertOption,
) (result sql.Result, err error) {
return d.doMergeInsert(ctx, link, table, list, option, false)
}
// doMergeInsert implements MERGE-based insert operations for MSSQL database.
// When withUpdate is true, it performs upsert (insert or update).
// When withUpdate is false, it performs insert ignore (insert only when no conflict).
func (d *Driver) doMergeInsert(
ctx context.Context,
link database.Link, table string, list database.List, option database.DoInsertOption, withUpdate bool,
) (result sql.Result, err error) {
// If OnConflict is not specified, automatically get the primary key of the table
conflictKeys := option.OnConflict
if len(conflictKeys) == 0 {
primaryKeys, err := d.Core.GetPrimaryKeys(ctx, table)
if err != nil {
return nil, gerror.WrapCode(
gcode.CodeInternalError,
err,
`failed to get primary keys for table`,
)
}
foundPrimaryKey := false
for _, primaryKey := range primaryKeys {
for dataKey := range list[0] {
if strings.EqualFold(dataKey, primaryKey) {
foundPrimaryKey = true
break
}
}
if foundPrimaryKey {
break
}
}
if !foundPrimaryKey {
return nil, gerror.NewCodef(
gcode.CodeMissingParameter,
`Replace/Save/InsertIgnore operation requires conflict detection: `+
`either specify OnConflict() columns or ensure table '%s' has a primary key in the data`,
table,
)
}
// TODO consider composite primary keys.
conflictKeys = primaryKeys
}
var (
one = list[0]
oneLen = len(one)
charL, charR = d.GetChars()
conflictKeySet = gset.NewStrSet(false)
// queryHolders: Handle data with Holder that need to be merged
// queryValues: Handle data that need to be merged
// insertKeys: Handle valid keys that need to be inserted
// insertValues: Handle values that need to be inserted
// updateValues: Handle values that need to be updated (only when withUpdate=true)
queryHolders = make([]string, oneLen)
queryValues = make([]any, oneLen)
insertKeys = make([]string, oneLen)
insertValues = make([]string, oneLen)
updateValues []string
)
// conflictKeys slice type conv to set type
for _, conflictKey := range conflictKeys {
conflictKeySet.Add(gstr.ToUpper(conflictKey))
}
index := 0
for key, value := range one {
queryHolders[index] = "?"
queryValues[index] = value
insertKeys[index] = charL + key + charR
insertValues[index] = "T2." + charL + key + charR
// Build updateValues only when withUpdate is true
// Filter conflict keys and soft created fields from updateValues
if withUpdate && !(conflictKeySet.Contains(key) || d.Core.IsSoftCreatedFieldName(key)) {
updateValues = append(
updateValues,
fmt.Sprintf(`T1.%s = T2.%s`, charL+key+charR, charL+key+charR),
)
}
index++
}
var (
batchResult = new(database.SqlResult)
sqlStr = parseSqlForMerge(table, queryHolders, insertKeys, insertValues, updateValues, conflictKeys)
)
r, err := d.DoExec(ctx, link, sqlStr, queryValues...)
if err != nil {
return r, err
}
if n, err := r.RowsAffected(); err != nil {
return r, err
} else {
batchResult.Result = r
batchResult.Affected += n
}
return batchResult, nil
}
// parseSqlForMerge generates MERGE statement for MSSQL database.
// When updateValues is empty, it only inserts (INSERT IGNORE behavior).
// When updateValues is provided, it performs upsert (INSERT or UPDATE).
// Examples:
// - INSERT IGNORE: MERGE INTO table T1 USING (...) T2 ON (...) WHEN NOT MATCHED THEN INSERT(...) VALUES (...)
// - UPSERT: MERGE INTO table T1 USING (...) T2 ON (...) WHEN NOT MATCHED THEN INSERT(...) VALUES (...) WHEN MATCHED THEN UPDATE SET ...
func parseSqlForMerge(table string,
queryHolders, insertKeys, insertValues, updateValues, duplicateKey []string,
) (sqlStr string) {
var (
queryHolderStr = strings.Join(queryHolders, ",")
insertKeyStr = strings.Join(insertKeys, ",")
insertValueStr = strings.Join(insertValues, ",")
duplicateKeyStr string
)
// Build ON condition
for index, keys := range duplicateKey {
if index != 0 {
duplicateKeyStr += " AND "
}
duplicateKeyStr += fmt.Sprintf("T1.%s = T2.%s", keys, keys)
}
// Build SQL based on whether UPDATE is needed
pattern := gstr.Trim(
`MERGE INTO %s T1 USING (VALUES(%s)) T2 (%s) ON (%s) WHEN NOT MATCHED THEN INSERT(%s) VALUES (%s)`,
)
if len(updateValues) > 0 {
// Upsert: INSERT or UPDATE
pattern += gstr.Trim(` WHEN MATCHED THEN UPDATE SET %s`)
return fmt.Sprintf(
pattern+";",
table,
queryHolderStr,
insertKeyStr,
duplicateKeyStr,
insertKeyStr,
insertValueStr,
strings.Join(updateValues, ","),
)
}
// Insert Ignore: INSERT only
return fmt.Sprintf(pattern+";", table, queryHolderStr, insertKeyStr, duplicateKeyStr, insertKeyStr, insertValueStr)
}
+62
View File
@@ -0,0 +1,62 @@
// Copyright GoFrame Author(https://goframe.org). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
package mssql
import (
"database/sql"
"fmt"
"git.magicany.cc/black1552/gin-base/database"
"github.com/gogf/gf/v2/errors/gcode"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/text/gstr"
)
// Open creates and returns an underlying sql.DB object for mssql.
func (d *Driver) Open(config *database.ConfigNode) (db *sql.DB, err error) {
source, err := configNodeToSource(config)
if err != nil {
return nil, err
}
underlyingDriverName := "sqlserver"
if db, err = sql.Open(underlyingDriverName, source); err != nil {
err = gerror.WrapCodef(
gcode.CodeDbOperationError, err,
`sql.Open failed for driver "%s" by source "%s"`, underlyingDriverName, source,
)
return nil, err
}
return
}
func configNodeToSource(config *database.ConfigNode) (string, error) {
var source string
source = fmt.Sprintf(
"user id=%s;password=%s;server=%s;encrypt=disable",
config.User, config.Pass, config.Host,
)
if config.Name != "" {
source = fmt.Sprintf("%s;database=%s", source, config.Name)
}
if config.Port != "" {
source = fmt.Sprintf("%s;port=%s", source, config.Port)
}
if config.Extra != "" {
extraMap, err := gstr.Parse(config.Extra)
if err != nil {
return "", gerror.WrapCodef(
gcode.CodeInvalidParameter,
err,
`invalid extra configuration: %s`, config.Extra,
)
}
for k, v := range extraMap {
source += fmt.Sprintf(`;%s=%s`, k, v)
}
}
return source, nil
}
+22
View File
@@ -0,0 +1,22 @@
// Copyright GoFrame Author(https://goframe.org). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
package mssql
// Result instance of sql.Result
type Result struct {
lastInsertId int64
rowsAffected int64
err error
}
func (r *Result) LastInsertId() (int64, error) {
return r.lastInsertId, r.err
}
func (r *Result) RowsAffected() (int64, error) {
return r.rowsAffected, r.err
}
@@ -0,0 +1,88 @@
// Copyright GoFrame Author(https://goframe.org). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
package mssql
import (
"context"
"fmt"
"git.magicany.cc/black1552/gin-base/database"
"github.com/gogf/gf/v2/util/gutil"
)
var (
tableFieldsSqlTmp = `
SELECT
c.name AS Field,
CASE
WHEN t.name IN ('datetime', 'datetime2', 'smalldatetime', 'date', 'time', 'text', 'ntext', 'image', 'xml') THEN t.name
WHEN t.name IN ('decimal', 'numeric') THEN t.name + '(' + CAST(c.precision AS varchar(20)) + ',' + CAST(c.scale AS varchar(20)) + ')'
WHEN t.name IN ('char', 'varchar', 'binary', 'varbinary') THEN t.name + '(' + CASE WHEN c.max_length = -1 THEN 'max' ELSE CAST(c.max_length AS varchar(20)) END + ')'
WHEN t.name IN ('nchar', 'nvarchar') THEN t.name + '(' + CASE WHEN c.max_length = -1 THEN 'max' ELSE CAST(c.max_length/2 AS varchar(20)) END + ')'
ELSE t.name
END AS Type,
CASE WHEN c.is_nullable = 1 THEN 'YES' ELSE 'NO' END AS [Null],
CASE WHEN pk.column_id IS NOT NULL THEN 'PRI' ELSE '' END AS [Key],
CASE WHEN c.is_identity = 1 THEN 'IDENTITY' ELSE '' END AS Extra,
ISNULL(dc.definition, '') AS [Default],
ISNULL(CAST(ep.value AS nvarchar(max)), '') AS [Comment]
FROM sys.columns c
INNER JOIN sys.objects o ON c.object_id = o.object_id AND o.type = 'U' AND o.is_ms_shipped = 0
INNER JOIN sys.types t ON c.user_type_id = t.user_type_id
LEFT JOIN sys.default_constraints dc ON c.default_object_id = dc.object_id
LEFT JOIN sys.extended_properties ep ON c.object_id = ep.major_id AND c.column_id = ep.minor_id AND ep.name = 'MS_Description'
LEFT JOIN (
SELECT ic.object_id, ic.column_id
FROM sys.index_columns ic
INNER JOIN sys.indexes i ON ic.object_id = i.object_id AND ic.index_id = i.index_id
WHERE i.is_primary_key = 1
) pk ON c.object_id = pk.object_id AND c.column_id = pk.column_id
WHERE o.name = '%s'
ORDER BY c.column_id
`
)
func init() {
var err error
tableFieldsSqlTmp, err = database.FormatMultiLineSqlToSingle(tableFieldsSqlTmp)
if err != nil {
panic(err)
}
}
// TableFields retrieves and returns the fields' information of specified table of current schema.
//
// Also see DriverMysql.TableFields.
func (d *Driver) TableFields(ctx context.Context, table string, schema ...string) (fields map[string]*database.TableField, err error) {
var (
result database.Result
link database.Link
usedSchema = gutil.GetOrDefaultStr(d.GetSchema(), schema...)
)
if link, err = d.SlaveLink(usedSchema); err != nil {
return nil, err
}
structureSql := fmt.Sprintf(tableFieldsSqlTmp, table)
result, err = d.DoSelect(ctx, link, structureSql)
if err != nil {
return nil, err
}
fields = make(map[string]*database.TableField)
for i, m := range result {
fields[m["Field"].String()] = &database.TableField{
Index: i,
Name: m["Field"].String(),
Type: m["Type"].String(),
Null: m["Null"].Bool(),
Key: m["Key"].String(),
Default: m["Default"].Val(),
Extra: m["Extra"].String(),
Comment: m["Comment"].String(),
}
}
return fields, nil
}
+38
View File
@@ -0,0 +1,38 @@
// Copyright GoFrame Author(https://goframe.org). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://github.com/gogf/gf.
package mssql
import (
"context"
"git.magicany.cc/black1552/gin-base/database"
)
const (
tablesSqlTmp = `SELECT name FROM sys.objects WHERE type='U' AND is_ms_shipped = 0 ORDER BY name`
)
// Tables retrieves and returns the tables of current schema.
// It's mainly used in cli tool chain for automatically generating the models.
func (d *Driver) Tables(ctx context.Context, schema ...string) (tables []string, err error) {
var result database.Result
link, err := d.SlaveLink(schema...)
if err != nil {
return nil, err
}
result, err = d.DoSelect(ctx, link, tablesSqlTmp)
if err != nil {
return
}
for _, m := range result {
for _, v := range m {
tables = append(tables, v.String())
}
}
return
}