upstream/icingadb
1
0
Fork 0
mirror of https://github.com/Icinga/icingadb.git synced 2026-06-04 22:32:57 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

DB#Upsert(): allow to update not all columns

Browse source
This commit is contained in:
Alexander A. Klimov 2021-03-24 12:59:26 +01:00
parent ebfabaffc2
commit 4e87ca6de3
3 changed files with 27 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
pkg/contracts/contracts.go
View file

@ -60,3 +60,9 @@ type Waiter interface {
type Initer interface {
Init() // Init initializes the object.
}
// Upserter implements the Upsert method,
// which returns a part of the object for ON DUPLICATE KEY UPDATE.
type Upserter interface {
Upsert() interface{} // Upsert partitions the object.
}

27
pkg/icingadb/db.go
View file

@ -86,21 +86,29 @@ func (db DB) BuildUpdateStmt(update interface{}) string {
)
}
func (db DB) BuildUpsertStmt(subject interface{}) string {
columns := db.BuildColumns(subject)
set := make([]string, 0, len(columns))
func (db DB) BuildUpsertStmt(subject interface{}) (stmt string, placeholders int) {
insertColumns := db.BuildColumns(subject)
var updateColumns []string
for _, col := range columns {
if upserter, ok := subject.(contracts.Upserter); ok {
updateColumns = db.BuildColumns(upserter.Upsert())
} else {
updateColumns = insertColumns
}
set := make([]string, 0, len(updateColumns))
for _, col := range updateColumns {
set = append(set, fmt.Sprintf("%s = :%s", col, col))
}
return fmt.Sprintf(
`INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s`,
utils.Key(utils.Name(subject), '_'),
strings.Join(columns, ","),
fmt.Sprintf(":%s", strings.Join(columns, ",:")),
strings.Join(insertColumns, ","),
fmt.Sprintf(":%s", strings.Join(insertColumns, ",:")),
strings.Join(set, ","),
)
), len(insertColumns) + len(updateColumns)
}
func (db DB) BulkExec(ctx context.Context, query string, count int, concurrent int, args []interface{}) error {
@ -358,9 +366,8 @@ func (db DB) Upsert(ctx context.Context, entities <-chan contracts.Entity, succe
return err
}
return db.NamedBulkExec(
ctx, db.BuildUpsertStmt(first), 1<<15/len(db.BuildColumns(first))/2, 1<<3, forward, succeeded,
)
stmt, placeholders := db.BuildUpsertStmt(first)
return db.NamedBulkExec(ctx, stmt, 1<<15/placeholders, 1<<3, forward, succeeded)
}
func (db DB) Update(ctx context.Context, entities <-chan contracts.Entity) error {

5
pkg/icingadb/ha.go
View file

@ -186,7 +186,10 @@ func (h *HA) realize(s *icingaredisv1.IcingaStatus, t *types.UnixMilli) error {
Icinga2FlapDetectionEnabled: s.FlapDetectionEnabled,
Icinga2PerformanceDataEnabled: s.PerformanceDataEnabled,
}
_, err = tx.NamedExecContext(ctx, h.db.BuildUpsertStmt(i), i)
stmt, _ := h.db.BuildUpsertStmt(i)
_, err = tx.NamedExecContext(ctx, stmt, i)
if err != nil {
cancel()
if !utils.IsDeadlock(err) {