From 8d077e2a88f4e740d8f5c169845bd7d5dffa2bd2 Mon Sep 17 00:00:00 2001 From: Jian Qiu Date: Fri, 23 Sep 2022 14:09:00 +0800 Subject: [PATCH] fix: splitable duplicate records (#15032) Co-authored-by: Qiu Jian --- go.mod | 2 +- go.sum | 4 +- pkg/util/splitable/insert.go | 64 ++++++++++++++------ pkg/util/splitable/metadata.go | 9 ++- pkg/util/splitable/purge.go | 14 ++++- pkg/util/splitable/splitable.go | 30 +++++++++ vendor/modules.txt | 2 +- vendor/yunion.io/x/sqlchemy/backends.go | 2 + vendor/yunion.io/x/sqlchemy/backends_base.go | 4 ++ vendor/yunion.io/x/sqlchemy/sync.go | 21 +++++++ vendor/yunion.io/x/sqlchemy/table.go | 3 + 11 files changed, 127 insertions(+), 28 deletions(-) diff --git a/go.mod b/go.mod index b7d48f1ef7..a0f45083c5 100644 --- a/go.mod +++ b/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 ) diff --git a/go.sum b/go.sum index c008048c61..44ab24a7d1 100644 --- a/go.sum +++ b/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= diff --git a/pkg/util/splitable/insert.go b/pkg/util/splitable/insert.go index a87db039ad..18bdb448fc 100644 --- a/pkg/util/splitable/insert.go +++ b/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) } diff --git a/pkg/util/splitable/metadata.go b/pkg/util/splitable/metadata.go index 2fbcb5f8b0..896dafdbca 100644 --- a/pkg/util/splitable/metadata.go +++ b/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 } diff --git a/pkg/util/splitable/purge.go b/pkg/util/splitable/purge.go index 7f33fb984a..9bbf6f35e5 100644 --- a/pkg/util/splitable/purge.go +++ b/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") } diff --git a/pkg/util/splitable/splitable.go b/pkg/util/splitable/splitable.go index 0b34c761f5..58d7701278 100644 --- a/pkg/util/splitable/splitable.go +++ b/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 diff --git a/vendor/modules.txt b/vendor/modules.txt index 10daedb42d..0457f7fe88 100644 --- a/vendor/modules.txt +++ b/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 diff --git a/vendor/yunion.io/x/sqlchemy/backends.go b/vendor/yunion.io/x/sqlchemy/backends.go index 3033c998fb..fc29d4464a 100644 --- a/vendor/yunion.io/x/sqlchemy/backends.go +++ b/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 diff --git a/vendor/yunion.io/x/sqlchemy/backends_base.go b/vendor/yunion.io/x/sqlchemy/backends_base.go index 0205092754..c7a126b729 100644 --- a/vendor/yunion.io/x/sqlchemy/backends_base.go +++ b/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 } diff --git a/vendor/yunion.io/x/sqlchemy/sync.go b/vendor/yunion.io/x/sqlchemy/sync.go index fcc26b5ab8..5858a14879 100644 --- a/vendor/yunion.io/x/sqlchemy/sync.go +++ b/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 diff --git a/vendor/yunion.io/x/sqlchemy/table.go b/vendor/yunion.io/x/sqlchemy/table.go index 5c043a913d..2ae5fbcbd2 100644 --- a/vendor/yunion.io/x/sqlchemy/table.go +++ b/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