fix: splitable duplicate records (#15032)

Co-authored-by: Qiu Jian <qiujian@yunionyun.com>

go.mod

@@ -105,7 +105,7 @@ require (
 	yunion.io/x/ovsdb v0.0.0-20200526071744-27bf0940cbc7
 	yunion.io/x/pkg v1.0.1-0.20220819080140-55b265a8658e
 	yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e
-	yunion.io/x/sqlchemy v1.1.2-0.20220815121507-5d4f1362bdf7
+	yunion.io/x/sqlchemy v1.1.2-0.20220923040147-ddd3e96868af
 	yunion.io/x/structarg v0.0.0-20220312084958-9c6c79c7d1c6
 )
 

go.sum

@@ -1169,7 +1169,7 @@ yunion.io/x/pkg v1.0.1-0.20220819080140-55b265a8658e/go.mod h1:nYOMflWG1wQC6301r
 yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e h1:v+EzIadodSwkdZ/7bremd7J8J50Cise/HCylsOJngmo=
 yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e/go.mod h1:0iFKpOs1y4lbCxeOmq3Xx/0AcQoewVPwj62eRluioEo=
 yunion.io/x/sqlchemy v1.0.1/go.mod h1:FTdwPdGhMgh4E+UFXc9klI1Ok34fMuybTT+jLhOaIjI=
-yunion.io/x/sqlchemy v1.1.2-0.20220815121507-5d4f1362bdf7 h1:KVO10d7utfv6oj5BA5MGyT3T/fSnYhwIGm04ucLomLw=
-yunion.io/x/sqlchemy v1.1.2-0.20220815121507-5d4f1362bdf7/go.mod h1:xXVfJLIxxTRTnPEaVIIoRZXqWyO8+takIOYhc+zO5KE=
+yunion.io/x/sqlchemy v1.1.2-0.20220923040147-ddd3e96868af h1:WjaARkzdxBQBD6j5I1MDMlJ0Z3WOHzEp5bTQl+TScaU=
+yunion.io/x/sqlchemy v1.1.2-0.20220923040147-ddd3e96868af/go.mod h1:xXVfJLIxxTRTnPEaVIIoRZXqWyO8+takIOYhc+zO5KE=
 yunion.io/x/structarg v0.0.0-20220312084958-9c6c79c7d1c6 h1:WuWXhY3DvhdRTzWCJ/kwt3Ss6KIq7+KqJwb+esvNGwU=
 yunion.io/x/structarg v0.0.0-20220312084958-9c6c79c7d1c6/go.mod h1:EP6NSv2C0zzqBDTKumv8hPWLb3XvgMZDHQRfyuOrQng=

pkg/util/splitable/insert.go

@@ -15,6 +15,7 @@
 package splitable
 
 import (
+	"database/sql"
 	"fmt"
 	"reflect"
 	"time"
@@ -26,17 +27,26 @@ import (
 	"yunion.io/x/sqlchemy"
 )
 
-func (t *SSplitTableSpec) Insert(dt interface{}) error {
-	metas, err := t.GetTableMetas()
-	if err != nil {
-		return errors.Wrap(err, "GetTableMeta")
+func (t *SSplitTableSpec) getLastTableSpecWithLock(lastDate time.Time) (*sqlchemy.STableSpec, error) {
+	t.lastTableLock.Lock()
+	defer t.lastTableLock.Unlock()
+	now := time.Now()
+	if t.lastTableSpec != nil && !t.lastTableExpire.IsZero() && t.lastTableExpire.Before(now) {
+		return t.lastTableSpec, nil
 	}
-	var lastDate time.Time
-	vs := reflectutils.FetchAllStructFieldValueSet(reflect.Indirect(reflect.ValueOf(dt)))
-	if lastDateV, ok := vs.GetValue(t.dateField); !ok {
-		return errors.Wrap(errors.ErrInvalidStatus, "no dateField found")
-	} else {
-		lastDate = lastDateV.Interface().(time.Time)
+	lastTableSpec, err := t.getLastTableSpec(lastDate)
+	if err != nil {
+		return nil, errors.Wrap(err, "getLastTableSpec")
+	}
+	t.lastTableSpec = lastTableSpec
+	t.lastTableExpire = now.Add(time.Hour * lastTableSpecExpireHours)
+	return t.lastTableSpec, nil
+}
+
+func (t *SSplitTableSpec) getLastTableSpec(lastDate time.Time) (*sqlchemy.STableSpec, error) {
+	lastMeta, err := t.getTableLastMeta()
+	if err != nil && errors.Cause(err) != sql.ErrNoRows {
+		return nil, errors.Wrap(err, "GetTableMeta")
 	}
 
 	var lastRecIndex int64
@@ -44,10 +54,9 @@ func (t *SSplitTableSpec) Insert(dt interface{}) error {
 	var lastTableSpec *sqlchemy.STableSpec
 
 	newMeta := false
-	if len(metas) > 0 {
-		lastMeta := metas[len(metas)-1]
+	if lastMeta != nil {
 		if !lastMeta.StartDate.IsZero() && lastDate.Sub(lastMeta.StartDate) > t.maxDuration {
-			lastTable := t.GetTableSpec(lastMeta)
+			lastTable := t.GetTableSpec(*lastMeta)
 			ti := lastTable.Instance()
 			q := ti.Query(sqlchemy.MAX("last_index", ti.Field(t.indexField)), sqlchemy.MAX("last_date", ti.Field(t.dateField)), sqlchemy.COUNT("total"))
 			r := q.Row()
@@ -55,34 +64,34 @@ func (t *SSplitTableSpec) Insert(dt interface{}) error {
 			var total uint64
 			err := r.Scan(&lastRecIndex, &lastRecDateStr, &total)
 			if err != nil {
-				return errors.Wrap(err, "scan lastRecIndex and lastRecDate")
+				return nil, errors.Wrap(err, "scan lastRecIndex and lastRecDate")
 			}
 			log.Debugf("lastRecDateStr: %s", lastRecDateStr)
 			lastRecDate, _ = timeutils.ParseTimeStr(lastRecDateStr)
 			// seal last meta
-			_, err = t.metaSpec.Update(&lastMeta, func() error {
+			_, err = t.metaSpec.Update(lastMeta, func() error {
 				lastMeta.End = lastRecIndex
 				lastMeta.EndDate = lastRecDate
 				lastMeta.Count = total
 				return nil
 			})
 			if err != nil {
-				return errors.Wrap(err, "Update last meta")
+				return nil, errors.Wrap(err, "Update last meta")
 			}
 			newMeta = true
 		} else {
 			if lastMeta.StartDate.IsZero() {
 				indexCol := t.tableSpec.ColumnSpec(t.indexField)
-				_, err = t.metaSpec.Update(&lastMeta, func() error {
+				_, err = t.metaSpec.Update(lastMeta, func() error {
 					lastMeta.Start = indexCol.AutoIncrementOffset()
 					lastMeta.StartDate = lastDate
 					return nil
 				})
 				if err != nil {
-					return errors.Wrap(err, "Update last meta")
+					return nil, errors.Wrap(err, "Update last meta")
 				}
 			}
-			lastTableSpec = t.GetTableSpec(lastMeta)
+			lastTableSpec = t.GetTableSpec(*lastMeta)
 		}
 	} else {
 		newMeta = true
@@ -90,9 +99,24 @@ func (t *SSplitTableSpec) Insert(dt interface{}) error {
 	if newMeta {
 		lastTableSpec, err = t.newTable(lastRecIndex, lastDate)
 		if err != nil {
-			return errors.Wrap(err, "newTable")
+			return nil, errors.Wrap(err, "newTable")
 		}
 	}
+	return lastTableSpec, nil
+}
+
+func (t *SSplitTableSpec) Insert(dt interface{}) error {
+	var lastDate time.Time
+	vs := reflectutils.FetchAllStructFieldValueSet(reflect.Indirect(reflect.ValueOf(dt)))
+	if lastDateV, ok := vs.GetValue(t.dateField); !ok {
+		return errors.Wrap(errors.ErrInvalidStatus, "no dateField found")
+	} else {
+		lastDate = lastDateV.Interface().(time.Time)
+	}
+	lastTableSpec, err := t.getLastTableSpecWithLock(lastDate)
+	if err != nil {
+		return errors.Wrap(err, "getLastTableSpec")
+	}
 	return lastTableSpec.Insert(dt)
 }
 

pkg/util/splitable/metadata.go

@@ -35,10 +35,17 @@ type STableMetadata struct {
 	CreatedAt time.Time `nullable:"false" created_at:"true"`
 }
 
+func (spec *SSplitTableSpec) getTableLastMeta() (*STableMetadata, error) {
+	return spec.GetTableMetaByTime(time.Time{})
+}
+
 func (spec *SSplitTableSpec) GetTableMetaByTime(recordTime time.Time) (*STableMetadata, error) {
 	q := spec.metaSpec.Query().Desc("id").IsFalse("deleted")
+	if !recordTime.IsZero() {
+		q = q.LE("start_date", recordTime)
+	}
 	meta := new(STableMetadata)
-	err := q.LE("start_date", recordTime).First(meta)
+	err := q.First(meta)
 	return meta, err
 }
 

pkg/util/splitable/purge.go

@@ -21,7 +21,6 @@ import (
 	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"
 	"yunion.io/x/pkg/utils"
-	"yunion.io/x/sqlchemy"
 )
 
 func (t *SSplitTableSpec) Purge(tables []string) ([]string, error) {
@@ -35,12 +34,21 @@ func (t *SSplitTableSpec) Purge(tables []string) ([]string, error) {
 	if t.maxSegments >= len(metas) && len(tables) == 0 {
 		return nil, nil
 	}
+	metaMax := len(metas)
+	if len(tables) == 0 {
+		// keep maxSegments if no table specified
+		metaMax -= t.maxSegments
+	} else {
+		// cannot delete the last one
+		metaMax -= 1
+	}
 	ret := []string{}
-	for i := 0; i < len(metas); i += 1 {
+	for i := 0; i < metaMax; i += 1 {
 		if len(tables) == 0 || utils.IsInStringArray(metas[i].Table, tables) {
 			dropSQL := fmt.Sprintf("DROP TABLE `%s`", metas[i].Table)
 			log.Infof("Ready to drop table: %s", dropSQL)
-			_, err := sqlchemy.Exec(dropSQL)
+			tblSpec := t.GetTableSpec(metas[i])
+			err := tblSpec.Drop()
 			if err != nil {
 				return ret, errors.Wrap(err, "sqlchemy.Exec")
 			}

pkg/util/splitable/splitable.go

@@ -18,6 +18,7 @@ import (
 	"database/sql"
 	"fmt"
 	"reflect"
+	"sync"
 	"time"
 
 	"yunion.io/x/pkg/errors"
@@ -33,8 +34,16 @@ type SSplitTableSpec struct {
 	metaSpec    *sqlchemy.STableSpec
 	maxDuration time.Duration
 	maxSegments int
+
+	lastTableSpec   *sqlchemy.STableSpec
+	lastTableLock   *sync.Mutex
+	lastTableExpire time.Time
 }
 
+const (
+	lastTableSpecExpireHours = 1 // 1 hour
+)
+
 func (t *SSplitTableSpec) DataType() reflect.Type {
 	return t.tableSpec.DataType()
 }
@@ -112,6 +121,25 @@ func (t *SSplitTableSpec) Fetch(dt interface{}) error {
 	return sql.ErrNoRows
 }
 
+func (t *SSplitTableSpec) Drop() error {
+	metas, err := t.GetTableMetas()
+	if err != nil {
+		return errors.Wrap(err, "GetTableMetas")
+	}
+	for _, meta := range metas {
+		ts := t.GetTableSpec(meta)
+		err := ts.Drop()
+		if err != nil {
+			return errors.Wrapf(err, "Drop %s", meta.Table)
+		}
+	}
+	err = t.metaSpec.Drop()
+	if err != nil {
+		return errors.Wrap(err, "Drop Meta")
+	}
+	return nil
+}
+
 func NewSplitTableSpec(s interface{}, name string, indexField string, dateField string, maxDuration time.Duration, maxSegments int, dbName sqlchemy.DBName) (*SSplitTableSpec, error) {
 	spec := sqlchemy.NewTableSpecFromStructWithDBName(s, name, dbName)
 	/*indexCol := spec.ColumnSpec(indexField)
@@ -142,6 +170,8 @@ func NewSplitTableSpec(s interface{}, name string, indexField string, dateField
 		metaSpec:    metaSpec,
 		maxDuration: maxDuration,
 		maxSegments: maxSegments,
+
+		lastTableLock: &sync.Mutex{},
 	}
 
 	return sts, nil

vendor/modules.txt

@@ -1473,7 +1473,7 @@ yunion.io/x/pkg/utils
 # yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e
 ## explicit; go 1.12
 yunion.io/x/s3cli
-# yunion.io/x/sqlchemy v1.1.2-0.20220815121507-5d4f1362bdf7
+# yunion.io/x/sqlchemy v1.1.2-0.20220923040147-ddd3e96868af
 ## explicit; go 1.17
 yunion.io/x/sqlchemy
 yunion.io/x/sqlchemy/backends

vendor/yunion.io/x/sqlchemy/backends.go

@@ -61,6 +61,8 @@ type IBackend interface {
 	UnionDistinctString() string
 	// support mixed insert vars
 	SupportMixedInsertVariables() bool
+	// Drop table
+	DropTableSQL(table string) string
 
 	// Capability
 

vendor/yunion.io/x/sqlchemy/backends_base.go

@@ -68,6 +68,10 @@ func (bb *SBaseBackend) UnionDistinctString() string {
 	return "UNION"
 }
 
+func (bb *SBaseBackend) DropTableSQL(table string) string {
+	return fmt.Sprintf("DROP TABLE `%s`", table)
+}
+
 func (bb *SBaseBackend) SupportMixedInsertVariables() bool {
 	return true
 }

vendor/yunion.io/x/sqlchemy/sync.go

@@ -144,6 +144,27 @@ func (ts *STableSpec) Exists() bool {
 	return in
 }
 
+// Drop drop table
+func (ts *STableSpec) Drop() error {
+	if !ts.Exists() {
+		return nil
+	}
+	db := ts.Database()
+	if db == nil {
+		panic("DropForeignKeySQL empty database")
+	}
+	if db.backend == nil {
+		panic("DropForeignKeySQL empty backend")
+	}
+	sql := db.backend.DropTableSQL(ts.name)
+	_, err := db.Exec(sql)
+	if err != nil {
+		log.Errorf("exec sql error %s: %s", sql, err)
+		return errors.Wrap(err, "Exec")
+	}
+	return nil
+}
+
 type STableChanges struct {
 	// indexes
 	RemoveIndexes []STableIndex

vendor/yunion.io/x/sqlchemy/table.go

@@ -77,6 +77,9 @@ type ITableSpec interface {
 
 	// Database returns the database of this table
 	Database() *SDatabase
+
+	// Drop drops table
+	Drop() error
 }
 
 // STableSpec defines the table specification, which implements ITableSpec