handle Restore Duplicate Entry

[qingxinhome]

User description

What type of PR is this?

• API-change
• BUG
• Improvement
• Documentation
• Feature
• Test and CI
• Code Refactoring

Which issue(s) this PR fixes:

issue #16100

What this PR does / why we need it:

SQL执行时将事务WriteOffset与当前语句绑定，解决读数据万圣节问题

MO Checkin Regression test susccess:
https://github.com/matrixorigin/ci-test/actions/runs/9362961560
https://github.com/matrixorigin/ci-test/actions/runs/9379340928

---

PR Type

Bug fix, Enhancement

---

Description

• Added txnOffset and fromSnapshot parameters to various methods across multiple files to handle transaction offsets and snapshot reads.
• Updated mock methods in engine_mock.go to include the new parameters.
• Added TxnOffset field to Scope and Compile structs and initialized it in NewCompile.
• Updated NewReader calls in scope.go to pass the TxnOffset.
• Modified moTableColMaxMinImpl to use the new Ranges method signature.
• Enhanced filter.go functions to include txnOffset and fromSnapshot parameters.
• Added txnOffset field to PartitionReader and updated its prepare method.
• Included txnOffset parameter in newBlockMergeReader and loadDeletes methods.
• Updated Ranges and related methods in txn_table.go to handle txnOffset and fromSnapshot.
• Added txnOffset field to blockMergeReader.
• Enhanced memory engine reader methods to include txnOffset parameter.
• Updated Relation interface to include txnOffset and fromSnapshot parameters in NewReader and Ranges methods.

---

Changes walkthrough 📝

Relevant files

Enhancement

12 files engine_mock.go Add new parameter to NewReader and Ranges methods in mocks pkg/frontend/test/engine_mock.go Added a new parameter arg5 to the NewReader and Ranges methods. Updated mock methods to include the new parameter. +8/-8      compile.go Initialize and use TxnOffset in compile methods                    pkg/sql/compile/compile.go Added TxnOffset initialization in NewCompile. Updated compileTableScanWithNode and expandRanges to use TxnOffset. +9/-3      scope.go Pass TxnOffset to NewReader in scope                                          pkg/sql/compile/scope.go Added TxnOffset parameter to NewReader calls. +24/-7    types.go Add TxnOffset field to Scope and Compile structs                  pkg/sql/compile/types.go Added TxnOffset field to Scope and Compile structs. +4/-0      func_mo.go Update moTableColMaxMinImpl to use new Ranges method signature pkg/sql/plan/function/func_mo.go Updated moTableColMaxMinImpl to use the new Ranges method signature. +2/-1      filter.go Add txnOffset and fromSnapshot parameters to filter functions pkg/vm/engine/disttae/filter.go Added txnOffset and fromSnapshot parameters to TryFastFilterBlocks and ExecuteBlockFilter. +7/-1      partition_reader.go Add txnOffset to PartitionReader and update prepare method pkg/vm/engine/disttae/partition_reader.go Added txnOffset field to PartitionReader. Updated prepare method to use txnOffset. +4/-3      reader.go Add txnOffset parameter to block merge reader methods        pkg/vm/engine/disttae/reader.go Added txnOffset parameter to newBlockMergeReader and loadDeletes. +4/-2      txn_table.go Add txnOffset and fromSnapshot to Ranges and related methods pkg/vm/engine/disttae/txn_table.go Added txnOffset and fromSnapshot parameters to Ranges and related methods. Updated NewReader and newMergeReader to use txnOffset. +47/-17  types.go Add txnOffset field to blockMergeReader                                    pkg/vm/engine/disttae/types.go Added txnOffset field to blockMergeReader. +2/-1      table_reader.go Add txnOffset parameter to memory engine reader methods    pkg/vm/engine/memoryengine/table_reader.go Added txnOffset parameter to NewReader and Ranges methods. +2/-2      types.go Update Relation interface with txnOffset and fromSnapshot parameters pkg/vm/engine/types.go Updated Relation interface to include txnOffset and fromSnapshot parameters in NewReader and Ranges methods. +4/-2

---

💡 PR-Agent usage:
Comment /help on the PR to get a list of all available PR-Agent tools and their descriptions

[codiumai-pr-agent-pro]

PR-Agent was enabled for this repository. To continue using it, please link your git user with your CodiumAI identity here.

PR Review 🔍

⏱️ Estimated effort to review [1-5]	4, because the PR involves multiple changes across various files and methods, adding new parameters to handle transaction offsets and snapshot reads. The changes are spread across different layers of the application, including mock interfaces, SQL compilation processes, and engine behaviors, which increases the complexity and the need for a thorough review to ensure consistency and correctness.
🧪 Relevant tests	No
⚡ Possible issues	Possible Bug: The addition of txnOffset and fromSnapshot parameters to various methods without corresponding updates in all calling locations could lead to runtime errors if not all usages have been correctly updated.
	Performance Concern: The modifications in transaction handling and snapshot reads might impact performance. It's crucial to benchmark these changes, especially in high-load scenarios, to ensure that the performance remains acceptable.
🔒 Security concerns	No

[codiumai-pr-agent-pro]

PR-Agent was enabled for this repository. To continue using it, please link your git user with your CodiumAI identity here.

PR Code Suggestions ✨

Category	Suggestion	Score
Best practice	Add a context cancellation check to handle timeouts or cancellations in long-running operations Consider adding a timeout or context cancellation check within the Ranges function to ensure that long-running operations can be interrupted if the context is canceled or times out. pkg/vm/engine/disttae/txn_table.go [633-636] func (tbl *txnTable) Ranges(ctx context.Context, exprs []*plan.Expr, txnOffset int, fromSnapshot bool) (ranges engine.Ranges, err error) { start := time.Now() seq := tbl.db.op.NextSequence() + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: + } trace.GetService().AddTxnDurationAction( Suggestion importance[1-10]: 8 Why: Adding a context cancellation check is crucial for long-running operations to handle timeouts or cancellations gracefully, improving the robustness and reliability of the function.	8
Possible bug	Add nil checks for tbl.proc and tbl.proc.Load() to prevent potential nil pointer dereferences To avoid potential nil pointer dereferences, ensure that tbl.proc and tbl.proc.Load() are not nil before using them in the Ranges function. pkg/vm/engine/disttae/txn_table.go [684-686] -tbl.proc.Load(), -txnOffset, -fromSnapshot, +if tbl.proc != nil && tbl.proc.Load() != nil { + tbl.proc.Load(), + txnOffset, + fromSnapshot, +} else { + return nil, errors.New("process is nil") +} Suggestion importance[1-10]: 7 Why: Ensuring that tbl.proc and tbl.proc.Load() are not nil before usage is a valid safety check to prevent runtime errors from nil pointer dereferences.	7
Maintainability	Use named parameters for the NewReader function to improve readability To improve readability and maintainability, consider using named parameters for the NewReader function. This will make it clear what each argument represents, especially since there are many boolean and integer parameters. pkg/sql/compile/scope.go [520-525] -readers, err = s.DataSource.Rel.NewReader(c.ctx, - scanUsedCpuNumber, - s.DataSource.FilterExpr, - s.NodeInfo.Data, - len(s.DataSource.OrderBy) > 0, - s.TxnOffset) +readers, err = s.DataSource.Rel.NewReader( + ctx: c.ctx, + parallel: scanUsedCpuNumber, + expr: s.DataSource.FilterExpr, + data: s.NodeInfo.Data, + hasOrderBy: len(s.DataSource.OrderBy) > 0, + txnOffset: s.TxnOffset) Suggestion importance[1-10]: 7 Why: Using named parameters enhances code readability and maintainability, especially when the function has multiple parameters of similar types.	7
	Use a constant for the default TxnOffset value to improve code readability Instead of setting TxnOffset to -1 when TxnOperator is nil, consider using a constant or a sentinel value to improve code readability and maintainability. pkg/sql/compile/compile.go [140] -c.TxnOffset = -1 +const DefaultTxnOffset = -1 +... +c.TxnOffset = DefaultTxnOffset Suggestion importance[1-10]: 5 Why: Using a constant for the default TxnOffset value enhances code readability and maintainability, although it's a relatively minor improvement.	5
Performance	Cache the result of GetSnapshotWriteOffset to avoid redundant calls and improve performance To improve performance, consider caching the result of tbl.getTxn().GetSnapshotWriteOffset() if it is called multiple times within the same function. pkg/vm/engine/disttae/txn_table.go [916-919] writeOffset := txnOffset if fromSnapshot { - writeOffset = tbl.getTxn().GetSnapshotWriteOffset() + snapshotWriteOffset := tbl.getTxn().GetSnapshotWriteOffset() + writeOffset = snapshotWriteOffset } Suggestion importance[1-10]: 6 Why: Caching the result of GetSnapshotWriteOffset can improve performance by reducing redundant calls, especially if the method is computationally expensive or called multiple times.	6
	Use a sync.Pool for the readers slice in the NewReader function to reduce allocations To improve performance, consider using a sync.Pool for the readers slice to reduce allocations in the NewReader function. pkg/vm/engine/memoryengine/table_reader.go [44-47] +var readerPool = sync.Pool{ + New: func() interface{} { + return make([]engine.Reader, 0, 10) + }, +} + func (t *Table) NewReader(ctx context.Context, parallel int, expr *plan.Expr, bytes []byte, _ bool, txnOffset int) (readers []engine.Reader, err error) { - readers = make([]engine.Reader, parallel) + readers = readerPool.Get().([]engine.Reader) + defer readerPool.Put(readers[:0]) + readers = readers[:parallel] shardIDs := ShardIdSlice(bytes) Suggestion importance[1-10]: 5 Why: Using sync.Pool could improve performance by reducing allocations, but the impact depends on the frequency and context of NewReader calls which isn't detailed in the suggestion.	5
Possible issue	Add validation checks for txnOffset and fromSnapshot parameters in the TryFastFilterBlocks function To ensure that the txnOffset and fromSnapshot parameters are always passed correctly, consider adding validation checks at the beginning of the TryFastFilterBlocks function. pkg/vm/engine/disttae/filter.go [992-999] func TryFastFilterBlocks( ctx context.Context, tbl *txnTable, txnOffset int, fromSnapshot bool, snapshotTS timestamp.Timestamp, tableDef *plan.TableDef, exprs []*plan.Expr, ) error { + if txnOffset < 0 { + return fmt.Errorf("invalid txnOffset: %d", txnOffset) + } + if !fromSnapshot { + return fmt.Errorf("fromSnapshot must be true") + } Suggestion importance[1-10]: 6 Why: Adding validation checks can prevent runtime errors and ensure data integrity, although the suggestion assumes specific validation rules without context from the existing codebase.	6
	Use a mutex to protect access to the txnOffset field in the PartitionReader struct To avoid potential issues with concurrent access, consider using a mutex to protect access to the txnOffset field in the PartitionReader struct. pkg/vm/engine/disttae/partition_reader.go [36-42] type PartitionReader struct { table *txnTable txnOffset int + txnOffsetMu sync.Mutex prepared bool inserts []*batch.Batch deletes map[types.Rowid]uint8 } +func (p *PartitionReader) setTxnOffset(offset int) { + p.txnOffsetMu.Lock() + defer p.txnOffsetMu.Unlock() + p.txnOffset = offset +} + +func (p *PartitionReader) getTxnOffset() int { + p.txnOffsetMu.Lock() + defer p.txnOffsetMu.Unlock() + return p.txnOffset +} + Suggestion importance[1-10]: 4 Why: While adding a mutex could prevent concurrent access issues, the suggestion does not provide evidence that txnOffset is accessed concurrently, which makes the necessity of this change unclear.	4