cgimport/sqlite/sqldb.go

package sqlite

import (
	"context"
	"database/sql"
	"fmt"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"time"

	"git.wecise.com/wecise/cgimport/sqlite/schema"
	"github.com/spf13/cast"
	"github.com/wecisecode/util/cmap"
	"github.com/wecisecode/util/merrs"
	"github.com/wecisecode/util/msgpack"
	"github.com/wecisecode/util/rc"
	"modernc.org/sqlite"
	sqlite3 "modernc.org/sqlite/lib"
)

type Status int

const (
	Uninited Status = iota // 未初始化，不可读，不可写
	Initing                // 正在初始化，不可读，不可写
	Inited                 // 初始化完成，可读，可写
	Closed                 // 已关闭，不可读，不可写
)

func (status Status) String() string {
	switch status {
	case Uninited:
		return "Uninited"
	case Initing:
		return "Initing"
	case Inited:
		return "Inited"
	case Closed:
		return "Closed"
	}
	return "unkown"
}

var ReNonWordChars = regexp.MustCompile(`\W`)

// 一个 SQLDB
// 对应一个读写表
// 或
// 对应一个视图，关联多个只读表
type SQLDB struct {
	db           *sql.DB
	keyspace     string
	dbname       string
	dbsourcename string
	//
	bufferedCommandsMutex sync.Mutex
	bufferedCommands      []*Command
	chstatus              chan Status
	chiniting             chan error
	// table info
	infomutex  sync.RWMutex
	tableinfo  *schema.TableInfo
	insertStmt *Stmt
	upsertStmt *Stmt
	datainfo   *DataInfo
	statinfo   *StatInfo
	//
	reorg_proc ReorgProc
	reorg_rc   *rc.RoutinesController
}

type ReorgProc struct {
	sync.Mutex
	statinfo_update func()
	datainfo_update func()
	expire_clear    func()
}

var dbs = cmap.New[string, *SQLDB]()

func NewSQLDB(keyspace string, dbname string, inmemory bool) (sdb *SQLDB, err error) {
	dbkey := fmt.Sprint("K:", keyspace, "N:", dbname, "M:", inmemory)
	sdb, err = dbs.GetWithNew(dbkey, func() (*SQLDB, error) {
		return openSQLDB(keyspace, dbname, inmemory, false)
	})
	if err != nil {
		return
	}
	if sdb == nil {
		err = merrs.NewError("NewSQLDB " + dbkey + " error")
		return
	}
	return sdb, nil
}

var DBPath = "/opt/matrix/var/cgimport/"

func openSQLDB(keyspace string, dbname string, inmemory bool, autorebuilding bool) (sqldb *SQLDB, err error) {
	var status Status = Uninited
	dbsourcename := "file:"
	if inmemory {
		dbsourcename += dbname + ":memory:?mode=memory&cache=shared"
	} else {
		dbfname := filepath.Join(DBPath, keyspace, dbname)
		e := os.MkdirAll(filepath.Dir(dbfname), os.ModePerm)
		if e != nil {
			err = merrs.NewError(e)
			return nil, err
		}
		fi, e := os.Stat(dbfname)
		if e != nil && !os.IsNotExist(e) {
			err = merrs.NewError(e)
			return nil, err
		}
		if fi != nil {
			if autorebuilding {
				e := os.Remove(dbfname)
				if e != nil {
					err = merrs.NewError(e)
					return nil, err
				}
				status = Uninited
			} else {
				status = Inited
			}
		}
		dbsourcename += dbfname + "?cache=shared"
	}
	db, e := sql.Open("sqlite", dbsourcename)
	if e != nil {
		err = merrs.NewError(e)
		return nil, err
	}
	if !inmemory {
		_, e = db.Exec("PRAGMA temp_store=memory") // 0=1=file 2=memory
		if e != nil {
			err = merrs.NewError(e)
			return nil, err
		}
		_, e = db.Exec("PRAGMA journal_mode=memory") // DELETE TRUNCATE PERSIST MEMORY OFF
		if e != nil {
			err = merrs.NewError(e)
			return nil, err
		}
		_, e = db.Exec("PRAGMA synchronous=off") // 0=off 1=normal 2=full
		if e != nil {
			err = merrs.NewError(e)
			return nil, err
		}
		// _, e = db.Exec("PRAGMA auto_vacuum=none") // 0=none 1=full
		// if e != nil {
		// 	err = merrs.NewError(e)
		// 	return nil
		// }
	}
	// _, e = db.Exec("PRAGMA count_changes=1")
	// if e != nil {
	// 	err = merrs.NewError(e)
	// 	return nil
	// }
	_, e = db.Exec("PRAGMA case_sensitive_like=1")
	if e != nil {
		err = merrs.NewError(e)
		return nil, err
	}
	sdb := &SQLDB{
		db:               db,
		keyspace:         keyspace,
		dbname:           dbname,
		dbsourcename:     dbsourcename,
		bufferedCommands: []*Command{},
		chstatus:         make(chan Status, 1),
		chiniting:        make(chan error, 1),
		reorg_rc:         rc.NewRoutinesController("", 1),
	}
	sdb.chiniting <- nil
	if status == Inited {
		e := sdb.loadTableInfo()
		if e != nil {
			if !merrs.UninitedError.Contains(e) {
				err = merrs.NewError(e)
				return nil, err
			}
			if !autorebuilding {
				sdb.db.Close()
				return openSQLDB(keyspace, dbname, inmemory, true)
			}
			status = Uninited
		}
	}
	sdb.chstatus <- status
	return sdb, nil
}

func (me *SQLDB) DBName() string {
	return me.dbname
}

func (me *SQLDB) SourceName() string {
	return me.dbsourcename
}

func (me *SQLDB) Reset() {
	status := <-me.chstatus
	if status == Initing {
		e := <-me.chiniting // 等待初始化完成
		me.chiniting <- e
	}
	me.chstatus <- Uninited
}

func (me *SQLDB) Close() error {
	status := <-me.chstatus
	if status == Initing {
		e := <-me.chiniting // 等待初始化完成
		me.chiniting <- e
	}
	me.chstatus <- Closed
	if status == Closed {
		return nil
	}
	return me.db.Close()
}

func (me *SQLDB) Status() Status {
	select {
	case status := <-me.chstatus:
		me.chstatus <- status
		return status
	}
}

func (me *SQLDB) SetMaxAttached() (err error) {
	// set SQLITE_MAX_ATTACHED = 125
	conn, e := me.db.Conn(context.Background())
	if e != nil {
		return merrs.NormalError.NewCause(e)
	}
	defer conn.Close()
	sqlite.Limit(conn, sqlite3.SQLITE_MAX_ATTACHED, 125)
	return nil
}

func (me *SQLDB) AttacheDB(clsdb *SQLDB, dbname string) error {
	dbsourcename := clsdb.SourceName()
	sql := "ATTACH DATABASE '" + dbsourcename + "' as " + dbname
	logger.Trace(sql)
	_, e := me.exec(sql)
	return e
}

func (me *SQLDB) DBList() (dbs map[string]string, err error) {
	sql := "PRAGMA database_list"
	rows, e := me.queryMaps(sql)
	if e != nil {
		return nil, merrs.NormalError.NewWith("", []error{e}, merrs.SSMaps{{"sql": sql}}, 1)
	}
	dbs = map[string]string{}
	for _, m := range rows {
		dbs[cast.ToString(m["name"])] = cast.ToString(m["file"])
	}
	return
}

func (me *SQLDB) InitTable(tabledefine *schema.TableDefine, force bool) (err error) {
	me.infomutex.Lock()
	defer me.infomutex.Unlock()
	status := <-me.chstatus
	if status == Inited && !force {
		// 初始化已完成，直接返回
		me.chstatus <- status
		return nil
	}
	if status == Initing {
		me.chstatus <- status
		e := <-me.chiniting // 等待初始化完成
		me.chiniting <- e
		return e
	}
	<-me.chiniting
	me.chstatus <- Initing
	// 操作计数
	// PushCount(time.Now(), 1, me.keyspace, "renew", tabledefine.TableName)
	e := me.renewTable(tabledefine)
	if e != nil {
		err = e
		me.chiniting <- err
		return
	}
	logger.Debug("Init LocalDB", me.dbsourcename)
	// 执行初始化期间产生的更新操作
	_, e = me.execBufferedCommands()
	if e != nil {
		err = merrs.NewError(e)
		me.chiniting <- err
		return
	}
	me.chiniting <- nil
	status = <-me.chstatus
	me.chstatus <- Inited
	// 执行新产生的更新操作
	_, e = me.execBufferedCommands()
	if e != nil {
		err = merrs.NewError(e)
		return
	}
	return nil
}

func (me *SQLDB) loadTableInfo() (err error) {
	rows, e := me.queryMaps("SELECT name FROM sqlite_master WHERE type='table'")
	if e != nil {
		return e
	}
	if len(rows) < 2 {
		return merrs.UninitedError.New("uninited")
	}

	// 通过 sqlite_master 和 PRAGMA table_info 获得的信息有限
	// 通过自定义的 schema 表，可以获得自己保存的信息
	rows, e = me.queryMaps("SELECT key,value FROM __table_info__")
	if e != nil {
		return e
	}
	if len(rows) < 3 {
		return merrs.UninitedError.New("uninited")
	}
	mrows := map[string]any{}
	for _, row := range rows {
		mrows[row["key"].(string)] = row["value"]
	}
	// tableinfo
	v_tableinfo := mrows["tableinfo"]
	if v_tableinfo == nil {
		return merrs.UninitedError.New("not found tableinfo")
	}
	s_tableinfo, ok := v_tableinfo.(string)
	if !ok {
		return merrs.UninitedError.New("tableinfo type inconsistent") // 类型不一致
	}
	load_tableinfo := &schema.TableInfo{}
	e = msgpack.Decode([]byte(s_tableinfo), load_tableinfo)
	if e != nil {
		err = merrs.NewError(e, merrs.SSMap{"tablename": load_tableinfo.TableName})
		return
	}
	if load_tableinfo.Version != schema.TableInfoVersion {
		return merrs.UninitedError.New("uninited load_tableinfo.version=" + load_tableinfo.Version)
	}
	me.tableinfo = load_tableinfo
	// datainfo
	v_datainfo := mrows["datainfo"]
	if v_datainfo == nil {
		return merrs.UninitedError.New("not found datainfo")
	}
	s_datainfo, ok := v_datainfo.(string)
	if !ok {
		return merrs.UninitedError.New("datainfo type inconsistent") // 类型不一致
	}
	me.datainfo = &DataInfo{}
	msgpack.Decode([]byte(s_datainfo), me.datainfo)
	// statinfo
	v_statinfo := mrows["statinfo"]
	if v_statinfo == nil {
		return merrs.UninitedError.New("not found statinfo")
	}
	s_statinfo, ok := v_statinfo.(string)
	if !ok {
		return merrs.UninitedError.New("statinfo type inconsistent") // 类型不一致
	}
	me.statinfo = &StatInfo{}
	msgpack.Decode([]byte(s_statinfo), me.statinfo)
	if me.statinfo.CreateTime.Equal(time.Time{}) {
		return merrs.UninitedError.New("uninited")
	}
	//
	insertstmt, upsertstmt, e := me.initInsertStmt(false)
	if e != nil {
		err = merrs.NewError(e, merrs.SSMap{"tablename": me.tableinfo.TableName})
		return
	}
	me.insertStmt = insertstmt
	me.upsertStmt = upsertstmt
	logger.Debug("Load LocalDB", me.dbsourcename)
	return
}

func (me *SQLDB) renewTable(tabledefine *schema.TableDefine) (err error) {
	tableinfo, createfieldtypes := tabledefine.TableInfo()
	if me.tableinfo != nil {
		// 简化结构，一个 DB 里只保留一个表或视图，可通过视图关联多个表
		err = me.dropTable(me.tableinfo.TableName)
		if err != nil {
			return
		}
	}
	e := me.newTable(tableinfo, createfieldtypes)
	if e != nil {
		err = merrs.NewError(e, merrs.SSMap{"tablename": tableinfo.TableName})
		return
	}
	me.tableinfo = tableinfo
	insertstmt, upsertstmt, e := me.initInsertStmt(true)
	if e != nil {
		err = merrs.NewError(e, merrs.SSMap{"tablename": tableinfo.TableName})
		return
	}
	me.insertStmt = insertstmt
	me.upsertStmt = upsertstmt
	return
}

func (me *SQLDB) dropTable(tablename string) error {
	dropSql := fmt.Sprintf(`DROP TABLE IF EXISTS %s`, tablename)
	// logger.Tracef("Drop table %s sql: %s", tablename, dropSql)
	_, e := me.exec(dropSql)
	if e != nil {
		return merrs.NewError(e, merrs.SSMap{"tablename": tablename})
	}
	return nil
}

func (me *SQLDB) newTable(tableinfo *schema.TableInfo,
	createfieldtypes []string,
) (err error) {
	tablename := tableinfo.TableName
	colums := strings.Join(createfieldtypes, `,`)
	if len(tableinfo.IDKeys) > 0 {
		colums = colums + `,` + `PRIMARY KEY("` + strings.Join(tableinfo.IDKeys, `","`) + `")`
	}
	createSql := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (%s)`, tablename, colums)
	// logger.Tracef("Create table %s sql: %s", tablename, createSql)
	if _, e := me.exec(createSql); e != nil {
		err = merrs.NewError(fmt.Sprintf("sqlite create table %s error", tablename), e)
		return
	}
	for idxName, idxfields := range tableinfo.Indexes {
		var indexSql string
		if len(idxfields) > 1 {
			indexSql = fmt.Sprintf(`CREATE INDEX IF NOT EXISTS %s on %s ("%s")`, idxName, tablename, strings.Join(idxfields, `", "`))
		} else {
			indexSql = fmt.Sprintf(`CREATE INDEX IF NOT EXISTS %s on %s ("%s")`, idxName, tablename, idxfields[0])
		}
		// logger.Tracef("Create table %s index sql: %s", tablename, indexSql)
		if _, e := me.exec(indexSql); e != nil {
			err = merrs.NewError(fmt.Sprintf("sqlite create table index error"), e)
			return
		}
	}
	if _, e := me.exec(`CREATE TABLE IF NOT EXISTS __table_info__ (key TEXT PRIMARY KEY, value TEXT)`); e != nil {
		err = merrs.NewError(fmt.Sprintf("sqlite create table %s error", tablename), e)
		return
	}
	if _, e := me.exec(`DELETE FROM __table_info__`); e != nil {
		err = merrs.NewError(fmt.Sprintf("sqlite create table %s error", tablename), e)
		return
	}
	tableinfobs, e := msgpack.Encode(tableinfo)
	if e != nil {
		err = merrs.NewError(fmt.Sprintf("sqlite create table %s error", tablename), e)
		return
	}
	if _, e := me.exec(`INSERT INTO __table_info__ (key, value) values (?,?)`, "tableinfo", string(tableinfobs)); e != nil {
		err = merrs.NewError(fmt.Sprintf("sqlite create table %s error", tablename), e)
		return
	}
	me.datainfo = &DataInfo{}
	datainfobs, e := msgpack.Encode(me.datainfo)
	if e != nil {
		err = merrs.NewError(fmt.Sprintf("sqlite create table %s error", tablename), e)
		return
	}
	if _, e := me.exec(`INSERT INTO __table_info__ (key, value) values (?,?)`, "datainfo", string(datainfobs)); e != nil {
		err = merrs.NewError(fmt.Sprintf("sqlite create table %s error", tablename), e)
		return
	}
	me.statinfo = &StatInfo{
		CreateTime: time.Now(),
	}
	statinfobs, e := msgpack.Encode(me.statinfo)
	if e != nil {
		err = merrs.NewError(fmt.Sprintf("sqlite create table %s error", tablename), e)
		return
	}
	if _, e := me.exec(`INSERT INTO __table_info__ (key, value) values (?,?)`, "statinfo", string(statinfobs)); e != nil {
		err = merrs.NewError(fmt.Sprintf("sqlite create table %s error", tablename), e)
		return
	}
	return nil
}

func (me *SQLDB) initInsertStmt(initvaluetypes bool) (insertstmt, upsertstmt *Stmt, err error) {
	qmarks := []string{}
	for range me.tableinfo.FieldNames {
		qmarks = append(qmarks, "?")
	}
	upsertSql := fmt.Sprintf(`BEGIN TRANSACTION;
DELETE FROM "`+me.tableinfo.TableName+`" WHERE "%s"=?;
INSERT INTO "`+me.tableinfo.TableName+`" ("%s") VALUES (%s);
COMMIT;`,
		strings.Join(me.tableinfo.IDKeys, `"=? and "`), strings.Join(me.tableinfo.FieldNames, `","`), strings.Join(qmarks, ","))
	upsertstmt, err = me.prepare(upsertSql)
	if err != nil {
		return
	}
	insertSql := fmt.Sprintf(`INSERT INTO "`+me.tableinfo.TableName+`" ("%s") VALUES (%s)`,
		strings.Join(me.tableinfo.FieldNames, `","`), strings.Join(qmarks, ","))
	insertstmt, err = me.prepare(insertSql)
	if err != nil {
		return
	}
	if initvaluetypes {
		if len(me.tableinfo.DefValues) != len(me.tableinfo.FieldNames) {
			err = merrs.NewError(fmt.Sprintf("sqlite prepare insert table %s error", me.tableinfo.TableName))
			return
		}
		params := []any{}
		for _, k := range me.tableinfo.IDKeys {
			for i, f := range me.tableinfo.FieldNames {
				if f == k {
					params = append(params, me.tableinfo.DefValues[i])
				}
			}
		}
		params = append(params, me.tableinfo.DefValues...)
		// 首次插入数据的类型决定该字段的实际存储类型
		_, err = insertstmt.Exec(params...)
		if err != nil {
			return
		}
		_, err = me.delete(nil)
		if err != nil {
			return
		}
	}
	return
}

func (me *SQLDB) setLastQueryTime() {
	me.statinfo.LastQueryTime = time.Now()
	me.reorg_proc.Lock()
	defer me.reorg_proc.Unlock()
	me.reorg_proc.statinfo_update = me.updateStatInfo
	me.reorg()
}

func (me *SQLDB) setLastUpdateTime() {
	me.statinfo.LastUpdateTime = time.Now()
	me.reorg_proc.Lock()
	defer me.reorg_proc.Unlock()
	me.reorg_proc.statinfo_update = me.updateStatInfo
	me.reorg()
}

func (me *SQLDB) updateStatInfo() {
	statinfobs, e := msgpack.Encode(me.statinfo)
	if e != nil {
		logger.Error("UpdateStatInfo Error", e)
	}
	sql := "UPDATE __table_info__ SET value=? WHERE key=?"
	_, e = me.exec(sql, "statinfo", string(statinfobs))
	if e != nil {
		logger.Error("UpdateStatInfo Error", e)
	}
}

func (me *SQLDB) clearExpireData() {
	// Format("2006-01-02 15:04:05")
	deleteSql := fmt.Sprintf(`DELETE FROM %s WHERE `+me.tableinfo.LatField+`<?`, me.tableinfo.TableName)
	n, e := me.exec(deleteSql, time.Now().Add(-me.tableinfo.Ttl).UnixMilli())
	if e != nil {
		logger.Error("ClearExpireData Error", e)
	}
	logger.Debug("ClearExpireData", n)
}

func (me *SQLDB) ClearExpireData() {
	if me.tableinfo.Ttl > 0 {
		me.reorg_proc.Lock()
		defer me.reorg_proc.Unlock()
		me.reorg_proc.expire_clear = me.clearExpireData
		me.reorg()
	}
}

func (me *SQLDB) reorg() {
	me.reorg_rc.CallLast2Only(func() {
		me.reorg_proc.Lock()
		datainfo_update := me.reorg_proc.datainfo_update
		statinfo_update := me.reorg_proc.statinfo_update
		expire_clear := me.reorg_proc.expire_clear
		me.reorg_proc.datainfo_update = nil
		me.reorg_proc.statinfo_update = nil
		me.reorg_proc.expire_clear = nil
		me.reorg_proc.Unlock()
		if datainfo_update != nil {
			datainfo_update()
		}
		if statinfo_update != nil {
			statinfo_update()
		}
		if expire_clear != nil {
			expire_clear()
		}
		time.Sleep(1 * time.Second)
	})
}

func (me *SQLDB) execBufferedCommands() (count int, err error) {

	for {
		var uc *Command
		me.bufferedCommandsMutex.Lock()
		if len(me.bufferedCommands) > 0 {
			uc = me.bufferedCommands[0]
			me.bufferedCommands = me.bufferedCommands[1:]
		}
		me.bufferedCommandsMutex.Unlock()
		if uc == nil {
			return
		}
		n, e := me.execCommand(uc)
		if e != nil {
			err = e
			if me.Status() != Inited {
				// 重新初始化
				me.bufferedCommandsMutex.Lock()
				me.bufferedCommands = append(me.bufferedCommands, uc)
				me.bufferedCommandsMutex.Unlock()
				// err = nil
				err = merrs.NewError(merrs.SSMaps{
					{
						"me.Status()": fmt.Sprint(me.Status()),
					}}, e)
				logger.Error(err)
				err = nil
			}
			return
		}
		count += n
	}
}

func (me *SQLDB) pushCommand(uc *Command) (n int, err error) {
	select {
	case status := <-me.chstatus:
		me.chstatus <- status
		switch status {
		case Uninited, Initing:
			me.bufferedCommandsMutex.Lock()
			me.bufferedCommands = append(me.bufferedCommands, uc)
			me.bufferedCommandsMutex.Unlock()
			return 1, nil
		case Inited:
			me.bufferedCommandsMutex.Lock()
			me.bufferedCommands = append(me.bufferedCommands, uc)
			me.bufferedCommandsMutex.Unlock()
			n, err = me.execBufferedCommands()
			return
		case Closed:
			return
		}
	}
	return
}

func (me *SQLDB) execCommand(uc *Command) (n int, err error) {
	switch uc.CmdType {
	case UCTInsert:
		return me.insert(uc.Data, uc.OverwriteExists)
	case UCTUpdate:
		return me.update(uc.Data, uc.Conds)
	case UCTDelete:
		return me.delete(uc.Conds)
	}
	panic(merrs.NewError(fmt.Sprint("unsupport UpdateCommandType ", uc.CmdType), nil))
}

func (me *SQLDB) Prepare(sql string) (*Stmt, error) {
	return me.prepare(sql)
}

func (me *SQLDB) prepare(sql string) (*Stmt, error) {
	stmt, e := me.db.Prepare(sql)
	if e != nil {
		return nil, merrs.NewError(e, merrs.SSMaps{{"sql": sql}})
	}
	return &Stmt{Stmt: stmt}, nil
}

func (me *SQLDB) exec(sql string, args ...any) (n int, err error) {
	rslt, e := me.db.Exec(sql, args...)
	if e != nil {
		return 0, merrs.NewError(e, merrs.SSMaps{{"sql": sql}, {"args": fmt.Sprintf("%v", args)}})
	}
	if rslt == nil {
		return 0, merrs.NewError("no result", merrs.SSMaps{{"sql": sql}, {"args": fmt.Sprintf("%v", args)}})
	}
	rn, e := rslt.RowsAffected()
	if e != nil {
		return 0, merrs.NewError(e, merrs.SSMaps{{"sql": sql}, {"args": fmt.Sprintf("%v", args)}})
	}
	n = int(rn)
	return
}

func (me *SQLDB) Select(ctx context.Context, sql string, args ...any) (iter *Iter, err error) {
	var status Status
	select {
	case status = <-me.chstatus:
		me.chstatus <- status
		if status == Closed {
			return nil, merrs.ClosedError.New("closed")
		}
		if status == Uninited || status == Initing {
			return nil, merrs.UninitedError.New("uninited")
		}
	}
	iter, err = me.query(ctx, sql, args...)
	me.setLastQueryTime()
	// PushCount(time.Now(), 1, me.keyspace, "select", me.tableinfo.TableName)
	return
}

func (me *SQLDB) queryMaps(sql string, args ...any) (rows []map[string]any, err error) {
	iter, e := me.query(nil, sql, args...)
	if e != nil {
		return nil, e
	}
	return iter.AllMaps()
}

func (me *SQLDB) query(ctx context.Context, sql string, args ...any) (iter *Iter, err error) {
	defer func() {
		x := recover()
		if x != nil {
			switch xv := x.(type) {
			case error:
				err = xv
			default:
				err = merrs.NewError(fmt.Sprint(x))
			}
		}
	}()
	if ctx == nil {
		ctx = context.Background()
	}
	rows, e := me.db.QueryContext(ctx, sql, args...)
	if e != nil {
		return nil, merrs.NewError(e, merrs.SSMaps{{"sql": sql}, {"args": fmt.Sprintf("%v", args)}})
	}
	coltypes, e := rows.ColumnTypes()
	if e != nil {
		return nil, merrs.NewError(e, merrs.SSMaps{{"sql": sql}, {"args": fmt.Sprintf("%v", args)}})
	}
	iter = &Iter{ctx: ctx, rows: rows, cols: coltypes}
	return
}

func (me *SQLDB) ids(data map[string]any) (idmap map[string]any) {
	idmap = map[string]any{}
	for _, k := range me.tableinfo.IDKeys {
		idmap[k] = data[k]
	}
	return
}

func (me *SQLDB) exists(conds map[string]any) (exists bool, err error) {
	values := []any{}
	existsSql := fmt.Sprintf(`SELECT * FROM %s`, me.tableinfo.TableName)
	if len(conds) > 0 {
		existsSql += " WHERE "
		keys := []string{}
		for k, v := range conds {
			keys = append(keys, k+"=?")
			values = append(values, v)
		}
		existsSql += strings.Join(keys, " and ")
	}
	existsSql += " LIMIT 1"
	logger.Tracef("Exists sql: %s, params=%v", existsSql, values)
	iter, e := me.Select(nil, existsSql, values...)
	if e != nil {
		return false, merrs.NewError(e)
	}
	defer iter.Close()
	if iter.rows.Next() {
		return true, nil
	}
	return false, nil
}

func (me *SQLDB) Insert(
	data map[string]any,
	overwriteexists bool,
) (
	n int,
	err error,
) {
	return me.pushCommand(&Command{CmdType: UCTInsert, Data: data, OverwriteExists: overwriteexists})
}

func (me *SQLDB) insert(
	data map[string]any,
	overwriteexists bool,
) (
	n int,
	err error,
) {
	encodedata := map[string]any{}
	for _, field := range me.tableinfo.FieldNames {
		ftype := me.tableinfo.MapNameFields[field].Type
		v, _ := data[field]
		if field == me.tableinfo.LatField {
			v = time.Now().UnixMilli()
		} else {
			v = SQLValueEncode(ftype, v)
		}
		encodedata[field] = v
	}
	values := []interface{}{}
	upsertStmt := me.insertStmt
	if overwriteexists {
		// 覆盖已有数据，需要指定主键值
		upsertStmt = me.upsertStmt
		for _, field := range me.tableinfo.IDKeys {
			values = append(values, encodedata[field])
		}
	}
	for _, field := range me.tableinfo.FieldNames {
		values = append(values, encodedata[field])
	}
	logger.Tracef("Insert %s data: values=%v", me.tableinfo.TableName, values)
	rslt, err := upsertStmt.Exec(values...)
	if err != nil {
		if strings.Index(err.Error(), "UNIQUE constraint") >= 0 {
			return 0, merrs.ExistError.NewError(err,
				merrs.SSMaps{
					{"TableName": me.tableinfo.TableName},
					{"Values": fmt.Sprint(values)},
					{"Status": fmt.Sprint(me.Status())},
				})
		}
		return 0, merrs.NewError(err,
			merrs.SSMaps{
				{"TableName": me.tableinfo.TableName},
				{"Values": fmt.Sprint(values)},
				{"Status": fmt.Sprint(me.Status())},
			})
	}
	if rslt == nil {
		return 0, merrs.NewError(fmt.Sprintf("sqlite insert table %s no result", me.tableinfo.TableName))
	}
	rn, err := rslt.RowsAffected()
	if err != nil {
		return 0, merrs.NewError(err,
			merrs.SSMaps{
				{"TableName": me.tableinfo.TableName},
				{"Values": fmt.Sprint(values)},
				{"Status": fmt.Sprint(me.Status())},
			})
	}
	n = int(rn)
	if n > 0 {
		me.setLastUpdateTime()
	}
	// PushCount(time.Now(), n, me.keyspace, "insert", me.tableinfo.TableName)
	return
}

func (me *SQLDB) Update(
	data map[string]any,
	conds map[string]any,
) (
	n int,
	err error,
) {
	return me.pushCommand(&Command{CmdType: UCTUpdate, Data: data, Conds: conds})
}

func (me *SQLDB) update(
	data map[string]any,
	conds map[string]any,
) (
	n int,
	err error,
) {
	updatefields := []string{}
	condfields := []string{}
	values := []interface{}{}
	for k, v := range data {
		updatefields = append(updatefields, k)
		values = append(values, v)
	}
	updatefields = append(updatefields, me.tableinfo.LatField)
	values = append(values, time.Now().UnixMilli())
	for k, v := range conds {
		condfields = append(condfields, k)
		values = append(values, v)
	}
	if len(values) == 0 {
		return 0, nil
	}
	updateSql := fmt.Sprintf(`UPDATE %s SET "%s"=?`, me.tableinfo.TableName, strings.Join(updatefields, `"=?,"`))
	if len(condfields) > 0 {
		updateSql += fmt.Sprintf(` WHERE "%s"=?`, strings.Join(condfields, `"=? and "`))
	}
	logger.Tracef("Update sql: %s, params=%v", updateSql, values)
	n, err = me.exec(updateSql, values...)
	if err != nil {
		return 0, merrs.NewError(fmt.Sprintf("sqlite update table %s error", me.tableinfo.TableName), err)
	}
	if n > 0 {
		me.setLastUpdateTime()
	}
	// PushCount(time.Now(), n, me.keyspace, "update", me.tableinfo.TableName)
	return
}

func (me *SQLDB) Delete(
	conds map[string]any,
) (
	n int,
	err error,
) {
	return me.pushCommand(&Command{CmdType: UCTDelete, Conds: conds})
}

func (me *SQLDB) delete(
	conds map[string]any,
) (
	n int,
	err error,
) {
	values := []interface{}{}
	deleteSql := fmt.Sprintf(`DELETE FROM %s`, me.tableinfo.TableName)
	if len(conds) > 0 {
		fields := []string{}
		for k, v := range conds {
			fields = append(fields, `"`+k+`"=?`)
			values = append(values, v)
		}
		deleteSql += " WHERE " + strings.Join(fields, " and ")
	}
	logger.Tracef("Delete sql: %s, params=%v", deleteSql, values)
	n, err = me.exec(deleteSql, values...)
	if err != nil {
		return 0, merrs.NewError(fmt.Sprintf("sqlite delete table %s error", me.tableinfo.TableName), err)
	}
	if n > 0 {
		me.setLastUpdateTime()
	}
	// PushCount(time.Now(), n, me.keyspace, "delete", me.tableinfo.TableName)
	return
}