update Handle function for presetIDStrategy

Author: pol-mfilipowicz-viant

go.mod

@@ -1,8 +1,6 @@
 module github.com/viant/sqlx
 
-go 1.21.6
-
-toolchain go1.21.12
+go 1.22.0
 
 require (
 	github.com/aerospike/aerospike-client-go v4.5.2+incompatible
@@ -36,13 +34,10 @@ require (
 	github.com/aerospike/aerospike-client-go/v6 v6.15.1 // indirect
 	github.com/andybalholm/brotli v1.0.4 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.0-20210816181553-5444fa50b93d // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-errors/errors v1.5.1 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
-	github.com/goccy/go-json v0.9.11 // indirect
-	github.com/golang-jwt/jwt/v4 v4.4.1 // indirect
 	github.com/golang-sql/civil v0.0.0-20190719163853-cb61b32ac6fe // indirect
 	github.com/golang-sql/sqlexp v0.1.0 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
@@ -51,19 +46,12 @@ require (
 	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
 	github.com/googleapis/gax-go/v2 v2.12.0 // indirect
 	github.com/klauspost/compress v1.15.9 // indirect
-	github.com/lestrrat-go/backoff/v2 v2.0.8 // indirect
-	github.com/lestrrat-go/blackmagic v1.0.0 // indirect
-	github.com/lestrrat-go/httpcc v1.0.1 // indirect
-	github.com/lestrrat-go/iter v1.0.1 // indirect
-	github.com/lestrrat-go/jwx v1.2.25 // indirect
-	github.com/lestrrat-go/option v1.0.0 // indirect
 	github.com/mattn/go-runewidth v0.0.9 // indirect
 	github.com/olekukonko/tablewriter v0.0.5 // indirect
 	github.com/onsi/ginkgo v1.16.5 // indirect
 	github.com/pierrec/lz4/v4 v4.1.15 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/segmentio/encoding v0.3.5 // indirect
-	github.com/viant/scy v0.3.2-0.20220825213848-acc5c59cde78 // indirect
 	github.com/viant/sqlparser v0.7.4 // indirect
 	github.com/viant/x v0.3.0 // indirect
 	github.com/yuin/gopher-lua v1.1.1 // indirect

io/insert/record_numeric.go

@@ -3,15 +3,16 @@ package insert
 import (
 	"context"
 	"fmt"
+	"reflect"
+	"sync"
+
 	"github.com/viant/sqlx"
 	"github.com/viant/sqlx/io"
 	"github.com/viant/sqlx/metadata"
 	"github.com/viant/sqlx/metadata/info"
 	"github.com/viant/sqlx/metadata/info/dialect"
 	"github.com/viant/sqlx/metadata/sink"
 	"github.com/viant/sqlx/option"
-	"reflect"
-	"sync"
 )
 
 type numericSequencer struct {
@@ -68,7 +69,7 @@ func (n *numericSequencer) nextSequence(ctx context.Context, sess *session, reco
 		return nil, nil
 	case dialect.PresetIDStrategyUndefined:
 		n.shallPresetIdentities = false
-		n.updateSequence(ctx, n.getSequenceName(sess), recordCount)
+		n.updateSequencer(ctx, n.getSequenceName(sess), recordCount)
 		return nil, nil
 	case dialect.PresetIDWithMax:
 		options = append(options, n.maxIDSQLBuilder(sess))
@@ -89,32 +90,26 @@ func (n *numericSequencer) nextSequence(ctx context.Context, sess *session, reco
 	return n.sequence, nil
 }
 
-func (n *numericSequencer) transientDMLBuilder(sess *session, record interface{}, batchRecordBuffer []interface{}, recordCount int64) func(*sink.Sequence) (*sqlx.SQL, error) {
-	return func(sequence *sink.Sequence) (*sqlx.SQL, error) {
+func (n *numericSequencer) transientDMLBuilder(sess *session, record interface{}, batchRecordBuffer []interface{}, recordCount int64) func(*sink.Sequence) (*sqlx.SQL, int64, error) {
+	return func(sequence *sink.Sequence) (*sqlx.SQL, int64, error) {
 		resetAutoincrementQuery := sess.Builder.Build(record, option.BatchSize(1))
 		resetAutoincrementQuery = sess.Dialect.EnsurePlaceholders(resetAutoincrementQuery)
 		sess.binder(record, batchRecordBuffer, 0, len(sess.columns))
 
 		values := make([]interface{}, len(sess.columns))
 		copy(values, batchRecordBuffer[0:len(sess.columns)-1]) // don't copy ID pointer (last position in slice)
 
-		oldValue := sequence.Value
-		var passedValue int64
-
-		switch recordCount {
-		default:
-			sequence.Value = sequence.NextValue(recordCount)
-			if diff := sequence.Value - oldValue; diff < recordCount {
-				return nil, fmt.Errorf("new next value for sequenceName %d is too small, expected >= %d but had ", sequence.Value, oldValue+recordCount)
-			}
-			passedValue = sequence.Value - sequence.IncrementBy // decreasing is required for transient insert approach
+		maxIdValue, err := sequence.ComputeNextForTransient(recordCount)
+		if err != nil {
+			return nil, 0, err
 		}
-		values[len(sess.columns)-1] = &passedValue
+
+		values[len(sess.columns)-1] = &maxIdValue
 		resetAutoincrementSQL := &sqlx.SQL{
 			Query: resetAutoincrementQuery,
 			Args:  values,
 		}
-		return resetAutoincrementSQL, nil
+		return resetAutoincrementSQL, recordCount, nil
 	}
 }
 
@@ -149,50 +144,41 @@ func (n *numericSequencer) getColumn() io.Column {
 }
 
 func (n *numericSequencer) prepareSequenceIfNeeded(ctx context.Context, sess *session, record interface{}, columnValue *interface{}, recordCount int, identitiesBatched []interface{}, options []option.Option) error {
-	// presetting sequence only once reserves (if implemented) sequence values on db only one time
-	if n.detectedPreset {
-		return nil
-	}
-
-	if recordCount == 0 {
+	// presetting sequence only once reserves (if implemented) sequence values on db only one time per batch operation
+	if recordCount <= 0 {
 		return nil
 	}
 
 	n.muxPreset.Lock()
+	defer n.muxPreset.Unlock()
+
 	if n.detectedPreset {
-		n.muxPreset.Unlock()
 		return nil
 	}
 
+	if columnValue == nil {
+		return fmt.Errorf("columnValue is nil")
+	}
+
 	isColumnZeroValue := isZero(*columnValue)
 	n.shallPresetIdentities = isColumnZeroValue
 
 	if isColumnZeroValue {
-		var err error
-		n.sequence, err = n.nextSequence(ctx, sess, record, identitiesBatched, recordCount, options)
+		seq, err := n.nextSequence(ctx, sess, record, identitiesBatched, recordCount, options)
 		if err != nil {
-			n.muxPreset.Unlock()
 			return err
 		}
-		//} else { // n.sequence should be nil (it's not in use), and it's important in afterFlush func
-		//	n.updateSequence(ctx, n.getSequenceName(sess), recordCount)
+		n.sequence = seq
 	}
 
 	if n.sequence != nil && n.shallPresetIdentities && n.sequenceValue == nil {
-		var seqValue int64
-
-		switch recordCount {
-		case 1: // TODO not proved that miss some edge cases, if does then only default case should be used
-			seqValue = n.sequence.Value - n.sequence.IncrementBy
-		default:
-			seqValue = n.sequence.MinValue(int64(recordCount))
-		}
-
+		seqValue := n.sequence.MinValue(int64(recordCount))
 		n.sequenceValue = &seqValue
 	}
 
-	n.detectedPreset = true // detectPreset must be here to avoid sending 0 in preset mode to db
-	n.muxPreset.Unlock()
+	// detectPreset must be set at the end to avoid sending 0 in preset mode to db
+	n.detectedPreset = true
+
 	return nil
 }
 
@@ -237,7 +223,7 @@ func (n *numericSequencer) afterFlush(ctx context.Context, values []interface{},
 		}
 		//in case there is a batch insert, we need to check if last inserted ID is the same as the sequence value
 		//if so we can safely update the identities with the new sequence value within the batch
-		n.updateSequence(ctx, n.sequence.Name, int(rowsAffected))
+		n.updateSequencer(ctx, n.sequence.Name, int(rowsAffected))
 		sequenceValue = n.sequence.Value
 		expectedNextInsertID := (1 + rowsAffected) * inceremntBy
 		if expectedNextInsertID != sequenceValue { //race condition during batch insert, skip updating IDs
@@ -272,7 +258,7 @@ func isZero(value interface{}) bool {
 	}
 }
 
-func (n *numericSequencer) updateSequence(ctx context.Context, sequenceName string, recordCount int) {
+func (n *numericSequencer) updateSequencer(ctx context.Context, sequenceName string, recordCount int) {
 	meta := metadata.New()
 	options := []option.Option{option.NewArgs(n.session.info.Catalog, n.session.info.Schema, sequenceName), n.session.Dialect}
 

io/insert/service.go

@@ -48,7 +48,7 @@ func New(ctx context.Context, db *sql.DB, tableName string, options ...option.Op
 	return inserter, nil
 }
 
-// NextSequence resets next updateSequence
+// NextSequence resets next updateSequencer
 func (s *Service) NextSequence(ctx context.Context, any interface{}, recordCount int, options ...option.Option) (*sink.Sequence, error) {
 	valueAt, count, err := io.Values(any)
 	if err != nil {
@@ -153,6 +153,22 @@ func (s *Service) NewSession(ctx context.Context, record interface{}, db *sql.DB
 	defer s.mux.Unlock()
 	rType := reflect.TypeOf(record)
 	if sess := s.cachedSession; sess != nil && sess.rType == rType && sess.batchSize == batchSize {
+		// Reset per-call state on cached numeric sequencers before returning
+		// to avoid carrying over detected preset state between sessions.
+		for _, updater := range s.cachedSession.recordUpdaters {
+			if asNumeric, ok := updater.(*numericSequencer); ok {
+				// reset preset/sequence flags under locks
+				asNumeric.muxPreset.Lock()
+				asNumeric.muxSequenceValue.Lock()
+				asNumeric.detectedPreset = false
+				asNumeric.sequence = nil
+				asNumeric.shallPresetIdentities = true
+				asNumeric.sequenceValue = nil
+				asNumeric.muxSequenceValue.Unlock()
+				asNumeric.muxPreset.Unlock()
+			}
+		}
+
 		if db == nil {
 			db = sess.db
 		}

metadata/product/mysql/sequence/handler.go

@@ -4,11 +4,12 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"github.com/viant/sqlx/metadata/sink"
-	"github.com/viant/sqlx/option"
 	"os"
 	"strconv"
 	"strings"
+
+	"github.com/viant/sqlx/metadata/sink"
+	"github.com/viant/sqlx/option"
 )
 
 const autoincrementAssignment = "AUTO_INCREMENT="
@@ -53,7 +54,7 @@ func updateSequence(ctx context.Context, db *sql.DB, sequence *sink.Sequence, tx
 		seqValueFragment := DDL[index+len(autoincrementAssignment):]
 		debugSequencer := os.Getenv("DEBUG_SEQUENCER") == "true"
 		if debugSequencer {
-			fmt.Printf("sequencer: %v, value: %v\n", sequence.Name, seqValueFragment)
+			fmt.Printf("u sequencer: %v, value: %v\n", sequence.Name, seqValueFragment)
 		}
 		if index := strings.Index(seqValueFragment, " "); index != -1 {
 			seqValueFragment = seqValueFragment[:index]
@@ -102,10 +103,14 @@ func ensureIncrementsAndValues(ctx context.Context, db *sql.DB, sequence *sink.S
 		sequence.MaxValue = MaxSeqValue
 	}
 
-	debugSequencer := os.Getenv("DEBUG_SEQUENCER") == "true"
-	if debugSequencer {
-		fmt.Printf("sequencer: %v, start: %v, value: %v, max: %v, increment: %v\n", sequence.Name, sequence.StartValue, sequence.Value, sequence.MaxValue, sequence.IncrementBy)
+	if sequence.DataType == "" {
+		sequence.DataType = "int"
 	}
+
+	//debugSequencer := os.Getenv("DEBUG_SEQUENCER") == "true"
+	//if debugSequencer {
+	//	fmt.Printf("e sequencer_ptr: %p sequencer: %v, start: %v, value: %v, max: %v, increment: %v\n", &sequence, sequence.Name, sequence.StartValue, sequence.Value, sequence.MaxValue, sequence.IncrementBy)
+	//}
 	return nil
 }