fix: UpdateStmt doesn't update the statement correctly in sqlx/bulkinserter.go (#3607)

core/stores/sqlx/bulkinserter.go

@@ -4,6 +4,7 @@ import (
 	"database/sql"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/zeromicro/go-zero/core/executors"
@@ -30,6 +31,7 @@ type (
 		executor *executors.PeriodicalExecutor
 		inserter *dbInserter
 		stmt     bulkStmt
+		lock     sync.RWMutex // guards stmt
 	}
 
 	bulkStmt struct {
@@ -65,6 +67,9 @@ func (bi *BulkInserter) Flush() {
 
 // Insert inserts given args.
 func (bi *BulkInserter) Insert(args ...any) error {
+	bi.lock.RLock()
+	defer bi.lock.RUnlock()
+
 	value, err := format(bi.stmt.valueFormat, args...)
 	if err != nil {
 		return err
@@ -95,6 +100,11 @@ func (bi *BulkInserter) UpdateStmt(stmt string) error {
 		return err
 	}
 
+	bi.lock.Lock()
+	defer bi.lock.Unlock()
+
+	// with write lock, it doesn't matter what's the order of setting bi.stmt and calling flush.
+	bi.stmt = bkStmt
 	bi.executor.Flush()
 	bi.executor.Sync(func() {
 		bi.inserter.stmt = bkStmt