eventservice: split large scanned transactions#5511
Conversation
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
do-not-merge/needs-linked-issue
release-note
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughThis PR adds row-level scan resumption and large-transaction spill handling across eventstore, eventservice, and event broker paths. It also adds dispatcher big-txn metrics, an integration test for large transaction splitting, a failpoint-triggered dispatcher reset, and a go.mod dependency reorder. ChangesLarge Transaction Split & Spill
Dispatcher batch-reset failpoint
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant EventBroker
participant EventScanner
participant DispatcherStat
participant EventStore
participant LargeTxnSpill
EventBroker->>DispatcherStat: getDataRange()/getScanTaskDataRange()
DispatcherStat-->>EventBroker: DataRange (RowLevelScanPosition)
EventBroker->>EventScanner: scan(dataRange)
EventScanner->>EventStore: GetIterator(dataRange)
EventStore-->>EventScanner: NextWithScanPosition()
EventScanner->>LargeTxnSpill: appendInsert(row)
EventScanner-->>EventBroker: events, scanProgress
EventBroker->>DispatcherStat: updateScanRangeWithPosition(progress)
EventBroker->>EventBroker: persist resume state
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
ti-chi-bot
Bot
added
the
size/XXL
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces support for row-level resume tokens and large transaction spilling in the event service. It adds a new EventIteratorWithScanPosition interface, updates the event store iterator to track and return opaque scan positions, and implements disk-based spilling for large transactions via largeTxnInsertSpill and largeTxnScanState. Review feedback identifies two key issues in large_txn_spill.go: a potential resource leak where a failed Close() in Cleanup() skips deleting the temporary file, and a loose record length check that could lead to OOM panics if the spill file is corrupted.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
Signed-off-by: dongmen <414110582@qq.com>
# Conflicts: # pkg/eventservice/event_scanner.go
Signed-off-by: dongmen <414110582@qq.com>
Signed-off-by: dongmen <414110582@qq.com>
Signed-off-by: dongmen <414110582@qq.com>
Introduce pkg/spill as a shared temp-file record store with chunked append, handle reads, sequential reads, and cleanup. Refactor eventservice large transaction insert spill to reuse it while keeping RawKVEntry encoding local.
Signed-off-by: dongmen <414110582@qq.com>
# Conflicts: # pkg/eventservice/event_broker_test.go
ti-chi-bot
Bot
removed
the
do-not-merge/work-in-progress
|
/test all |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pkg/eventservice/event_scanner.go (1)
336-340: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winSet
session.progressbefore returning herepkg/eventservice/event_scanner.go:336-340—interruptScan()only appends events, butscan()returnssess.progress; this branch can emit a zero-value checkpoint and break resume handling.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/eventservice/event_scanner.go` around lines 336 - 340, The interrupt path in scan() currently returns before session.progress is updated, so the checkpoint can remain stale or zero-valued. In the branch that calls interruptScan() in event_scanner.go, assign session.progress from the current rawEvent state before returning, using the same progress fields that scan() later emits so resumed scans continue from the correct checkpoint.
🧹 Nitpick comments (6)
tests/integration_tests/large_txn_split/run.sh (2)
13-14: 🔒 Security & Privacy | 🔵 Trivial | 💤 Low valueUnquoted
rm -rf $WORK_DIRflagged by static analysis.Static analysis flags the unquoted expansion in
rm -rf $WORK_DIRas a potential path-injection/word-splitting risk.WORK_DIRis test-harness controlled here, but quoting is a trivial, no-risk hardening matching the also-flagged SC2086 hints on lines 14-25, 80, 98, 114-117.🔧 Proposed quoting fix
- rm -rf $WORK_DIR && mkdir -p $WORK_DIR + rm -rf -- "$WORK_DIR" && mkdir -p "$WORK_DIR"🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration_tests/large_txn_split/run.sh` around lines 13 - 14, The shell script has an unquoted variable expansion in the `prepare` function that triggers static analysis warnings; update the `rm -rf $WORK_DIR` usage to quote `WORK_DIR` consistently with the other SC2086-related command invocations in this script. Apply the same quoting pattern anywhere `WORK_DIR` is expanded in `prepare` and the related test helpers so the shell treats it as a single path argument.Source: Linters/SAST tools
98-98: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
prepare $*passes unused arguments.
prepare()(lines 13-32) takes no parameters and never references$1/$@, soprepare $*is misleading; also flagged by shellcheck (SC2048) for missing quoting.🔧 Proposed cleanup
- prepare $* + prepare🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration_tests/large_txn_split/run.sh` at line 98, The `prepare $*` call in `run.sh` is passing unused positional arguments to `prepare()`, which takes no parameters and does not reference them. Update the call site near `prepare` to invoke it without forwarding `$*`, and keep the existing `prepare` function signature as-is unless you actually intend to consume arguments.Source: Linters/SAST tools
pkg/eventservice/dispatcher_stat.go (1)
234-252: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winCopy scan-position slices at the state boundary.
common.ScanPositionis a mutable[]byte; storing and returning the same backing array lets later caller mutations corrupt the dispatcher’s resume cursor.Suggested fix
func (a *dispatcherStat) storeLastScannedPosition(position common.ScanPosition) { if len(position) == 0 { a.lastScannedPosition.Store(common.ScanPosition{}) return } - // ScanPosition is produced as an immutable snapshot by the scanner path. - a.lastScannedPosition.Store(position) + a.lastScannedPosition.Store(append(common.ScanPosition(nil), position...)) } func (a *dispatcherStat) getLastScannedPosition() common.ScanPosition { value := a.lastScannedPosition.Load() if value == nil { @@ if len(position) == 0 { return nil } - return position + return append(common.ScanPosition(nil), position...) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/eventservice/dispatcher_stat.go` around lines 234 - 252, The dispatcherStat scan-position handling is leaking a mutable common.ScanPosition slice across the state boundary, so caller mutations can corrupt the stored resume cursor. Update storeLastScannedPosition and getLastScannedPosition to make defensive copies when storing into lastScannedPosition and when returning from it, rather than keeping or exposing the same backing array. Use the dispatcherStat methods and the common.ScanPosition type to locate the fix.logservice/eventstore/format.go (1)
102-108: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueSuccessor-byte technique is correct.
Manually rebuilding
prefix + position + 0x00to produce a strict successor key matches the same append-0x00 pattern Pebble's own default comparer uses forImmediateSuccessor. Confirmed via Pebble source. Logic checks out against thepebble_test.goassertions.Minor optional note: this duplicates the uniqueID/tableID prefix-writing already done in
encodeTxnCommitTsBoundaryKeyTo(line 83-87 area). Could factor out a smallwriteKeyPrefix(buf, uniqueID, tableID)helper shared by both call sites to avoid drift if the prefix layout changes.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/eventstore/format.go` around lines 102 - 108, The successor-byte logic in encodeRowLevelScanPositionLowerBound is fine, but the uniqueID/tableID prefix writing is duplicated from encodeTxnCommitTsBoundaryKeyTo. Extract the shared prefix construction into a small helper such as a writeKeyPrefix function and use it in both places so the key layout stays consistent if the prefix format changes.pkg/eventservice/event_service_test.go (1)
384-389: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueAdd a length guard to avoid a panic on malformed positions.
binary.BigEndian.Uint64will panic ifpositionis non-empty but shorter than 8 bytes. Currently only internally-produced 8-byte positions reach this function, but a defensive check would avoid future breakage if the mock is reused with externally-supplied positions.🛡️ Suggested guard
func decodeMockScanPosition(position common.ScanPosition) int { - if len(position) == 0 { + if len(position) != 8 { return -1 } return int(binary.BigEndian.Uint64(position)) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/eventservice/event_service_test.go` around lines 384 - 389, Add a defensive length check in decodeMockScanPosition so it only calls binary.BigEndian.Uint64 when common.ScanPosition has at least 8 bytes; otherwise return the sentinel value already used for empty input. Keep the existing behavior for valid 8-byte positions, and make the guard explicit in this helper so future callers of decodeMockScanPosition do not panic on malformed or externally supplied positions.pkg/eventservice/event_broker.go (1)
1127-1132: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate cleanup+log pattern across three call sites.
The cleanup-and-warn block is repeated verbatim (differing only in the log message) in
removeDispatcher,resetDispatcher's initial path, and its CAS-retry loop. Consider extracting a small helper, e.g.logLargeTxnCleanupErr(stat, dispatcherInfo, context).♻️ Suggested helper
func (c *eventBroker) cleanupLargeTxnStateLogged(stat *dispatcherStat, dispatcherInfo DispatcherInfo, action string) { if err := stat.cleanupLargeTxnState(); err != nil { log.Warn("cleanup large txn state failed", zap.String("action", action), zap.Stringer("changefeedID", dispatcherInfo.GetChangefeedID()), zap.Stringer("dispatcherID", dispatcherInfo.GetID()), zap.Error(err)) } }Also applies to: 1206-1211, 1262-1267
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/eventservice/event_broker.go` around lines 1127 - 1132, The cleanup-and-warn block around stat.cleanupLargeTxnState() is duplicated in removeDispatcher and both resetDispatcher paths, so extract it into a small helper on eventBroker (for example cleanupLargeTxnStateLogged or logLargeTxnCleanupErr) and call that from each site. Have the helper accept the dispatcher state plus dispatcherInfo and an action/context string, then centralize the log.Warn call there while preserving the existing changefeedID, dispatcherID, and error fields.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pkg/common/table_span.go`:
- Around line 62-65: Update the DataRange equality logic in the table span
comparison so it also includes LastScannedTxnStartTs alongside Span,
CommitTsStart, CommitTsEnd, and RowLevelScanPosition. The issue is in the
equality method that currently returns true for ranges with different
transaction resume cursors when RowLevelScanPosition is empty; add
LastScannedTxnStartTs to the same comparison in the DataRange/TableSpan equality
path so distinct iterator lower bounds no longer compare equal.
In `@pkg/eventservice/dispatcher_stat.go`:
- Around line 271-279: The big transaction metric state is being replaced in the
dispatcher without first emitting the currently pending metric, which drops data
when a new (startTs, commitTs) pair arrives. Update the logic in the
bigTxnMetricState handling block to flush/emit the existing state before
assigning a new bigTxnMetricState, using the same code path that records the
metric so the pending value is not lost.
In `@pkg/eventservice/large_txn_spill.go`:
- Around line 30-48: newLargeTxnInsertSpill and largeTxnInsertSpill.Append are
creating ad-hoc errors with errors.New instead of using the repository’s
predefined spill-file error. Update these paths to return the existing
spill-related error type (for example cerror.ErrSpillFileOp) and preserve the
specific context for empty dir, record file creation failure, and nil entry
cases. Use the symbols newLargeTxnInsertSpill, Append, and cerror.ErrSpillFileOp
to locate and replace the generic error creation consistently.
In `@pkg/eventservice/large_txn_state.go`:
- Around line 61-65: The large-txn state checks in large_txn_state.go are
returning ad-hoc errors instead of the repository’s predefined error types.
Update the mismatch/cleanup/wrong-phase paths in the relevant large txn state
helpers (including the state comparison block and the related cases in the same
file) to return the existing shared errors used by large_txn_spill.go, and
follow the error-handling guidance in docs/agents/error-handling.md so callers
can reliably match these conditions.
In `@pkg/spill/record_file.go`:
- Around line 129-137: The deferred close on the freshly opened spill file
handle in the spill file open path ignores a possible error, which is what the
CI check is flagging. Update the `recordFile` open logic so the `os.Open` handle
is closed with error handling rather than a bare `defer file.Close()`, and make
sure any close failure is handled or logged consistently with the existing error
wrapping used in `cerror.WrapError` and the `spill` file access flow.
- Around line 16-22: The import in record_file.go aliases
github.com/pingcap/ticdc/pkg/errors as cerror even though there is no local
package name conflict, so it should be imported unaliased to satisfy the
importas rule. Update the import block in the RecordFile code accordingly, then
rename every cerror usage in this file to the package name errors, including the
error constants and helpers used in the RecordFile methods.
---
Outside diff comments:
In `@pkg/eventservice/event_scanner.go`:
- Around line 336-340: The interrupt path in scan() currently returns before
session.progress is updated, so the checkpoint can remain stale or zero-valued.
In the branch that calls interruptScan() in event_scanner.go, assign
session.progress from the current rawEvent state before returning, using the
same progress fields that scan() later emits so resumed scans continue from the
correct checkpoint.
---
Nitpick comments:
In `@logservice/eventstore/format.go`:
- Around line 102-108: The successor-byte logic in
encodeRowLevelScanPositionLowerBound is fine, but the uniqueID/tableID prefix
writing is duplicated from encodeTxnCommitTsBoundaryKeyTo. Extract the shared
prefix construction into a small helper such as a writeKeyPrefix function and
use it in both places so the key layout stays consistent if the prefix format
changes.
In `@pkg/eventservice/dispatcher_stat.go`:
- Around line 234-252: The dispatcherStat scan-position handling is leaking a
mutable common.ScanPosition slice across the state boundary, so caller mutations
can corrupt the stored resume cursor. Update storeLastScannedPosition and
getLastScannedPosition to make defensive copies when storing into
lastScannedPosition and when returning from it, rather than keeping or exposing
the same backing array. Use the dispatcherStat methods and the
common.ScanPosition type to locate the fix.
In `@pkg/eventservice/event_broker.go`:
- Around line 1127-1132: The cleanup-and-warn block around
stat.cleanupLargeTxnState() is duplicated in removeDispatcher and both
resetDispatcher paths, so extract it into a small helper on eventBroker (for
example cleanupLargeTxnStateLogged or logLargeTxnCleanupErr) and call that from
each site. Have the helper accept the dispatcher state plus dispatcherInfo and
an action/context string, then centralize the log.Warn call there while
preserving the existing changefeedID, dispatcherID, and error fields.
In `@pkg/eventservice/event_service_test.go`:
- Around line 384-389: Add a defensive length check in decodeMockScanPosition so
it only calls binary.BigEndian.Uint64 when common.ScanPosition has at least 8
bytes; otherwise return the sentinel value already used for empty input. Keep
the existing behavior for valid 8-byte positions, and make the guard explicit in
this helper so future callers of decodeMockScanPosition do not panic on
malformed or externally supplied positions.
In `@tests/integration_tests/large_txn_split/run.sh`:
- Around line 13-14: The shell script has an unquoted variable expansion in the
`prepare` function that triggers static analysis warnings; update the `rm -rf
$WORK_DIR` usage to quote `WORK_DIR` consistently with the other SC2086-related
command invocations in this script. Apply the same quoting pattern anywhere
`WORK_DIR` is expanded in `prepare` and the related test helpers so the shell
treats it as a single path argument.
- Line 98: The `prepare $*` call in `run.sh` is passing unused positional
arguments to `prepare()`, which takes no parameters and does not reference them.
Update the call site near `prepare` to invoke it without forwarding `$*`, and
keep the existing `prepare` function signature as-is unless you actually intend
to consume arguments.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 4d2dd6fe-80ac-444b-83dd-e765ee0225f8
📒 Files selected for processing (30)
downstreamadapter/eventcollector/dispatcher_stat.godownstreamadapter/eventcollector/dispatcher_stat_test.gogo.modlogservice/eventstore/event_store.gologservice/eventstore/event_store_test.gologservice/eventstore/format.gologservice/eventstore/pebble_test.gopkg/common/table_span.gopkg/config/debug.gopkg/config/server_config_test.gopkg/errors/error.gopkg/eventservice/dispatcher_stat.gopkg/eventservice/dispatcher_stat_test.gopkg/eventservice/event_broker.gopkg/eventservice/event_broker_test.gopkg/eventservice/event_scanner.gopkg/eventservice/event_scanner_benchmark_test.gopkg/eventservice/event_scanner_test.gopkg/eventservice/event_service_test.gopkg/eventservice/large_txn_spill.gopkg/eventservice/large_txn_spill_test.gopkg/eventservice/large_txn_state.gopkg/eventservice/metrics_collector.gopkg/metrics/event_service.gopkg/spill/record_file.gopkg/spill/record_file_test.gotests/integration_tests/large_txn_split/conf/diff_config.tomltests/integration_tests/large_txn_split/conf/server.tomltests/integration_tests/large_txn_split/run.shtests/integration_tests/run_light_it_in_ci.sh
| return d.Span.Equal(other.Span) && | ||
| d.CommitTsStart == other.CommitTsStart && | ||
| d.CommitTsEnd == other.CommitTsEnd && | ||
| bytes.Equal(d.RowLevelScanPosition, other.RowLevelScanPosition) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Include LastScannedTxnStartTs in range equality.
LastScannedTxnStartTs changes the iterator lower bound when RowLevelScanPosition is empty, so two DataRanges with different transaction-level resume cursors can currently compare equal.
Suggested fix
return d.Span.Equal(other.Span) &&
d.CommitTsStart == other.CommitTsStart &&
d.CommitTsEnd == other.CommitTsEnd &&
- bytes.Equal(d.RowLevelScanPosition, other.RowLevelScanPosition)
+ d.LastScannedTxnStartTs == other.LastScannedTxnStartTs &&
+ bytes.Equal(d.RowLevelScanPosition, other.RowLevelScanPosition)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| return d.Span.Equal(other.Span) && | |
| d.CommitTsStart == other.CommitTsStart && | |
| d.CommitTsEnd == other.CommitTsEnd && | |
| bytes.Equal(d.RowLevelScanPosition, other.RowLevelScanPosition) | |
| return d.Span.Equal(other.Span) && | |
| d.CommitTsStart == other.CommitTsStart && | |
| d.CommitTsEnd == other.CommitTsEnd && | |
| d.LastScannedTxnStartTs == other.LastScannedTxnStartTs && | |
| bytes.Equal(d.RowLevelScanPosition, other.RowLevelScanPosition) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/common/table_span.go` around lines 62 - 65, Update the DataRange equality
logic in the table span comparison so it also includes LastScannedTxnStartTs
alongside Span, CommitTsStart, CommitTsEnd, and RowLevelScanPosition. The issue
is in the equality method that currently returns true for ranges with different
transaction resume cursors when RowLevelScanPosition is empty; add
LastScannedTxnStartTs to the same comparison in the DataRange/TableSpan equality
path so distinct iterator lower bounds no longer compare equal.
| if a.bigTxnMetricState == nil || | ||
| a.bigTxnMetricState.startTs != startTs || | ||
| a.bigTxnMetricState.commitTs != commitTs { | ||
| a.bigTxnMetricState = &bigTxnMetricState{ | ||
| startTs: startTs, | ||
| commitTs: commitTs, | ||
| largeTxnThresholdInBytes: largeTxnThresholdInBytes, | ||
| } | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Flush the pending metric before replacing it.
When a different (startTs, commitTs) arrives, the existing bigTxnMetricState is overwritten without being emitted.
Suggested fix
if a.bigTxnMetricState == nil ||
a.bigTxnMetricState.startTs != startTs ||
a.bigTxnMetricState.commitTs != commitTs {
+ a.finishPendingBigTxnMetricBefore(startTs, commitTs)
a.bigTxnMetricState = &bigTxnMetricState{
startTs: startTs,
commitTs: commitTs,
largeTxnThresholdInBytes: largeTxnThresholdInBytes,📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if a.bigTxnMetricState == nil || | |
| a.bigTxnMetricState.startTs != startTs || | |
| a.bigTxnMetricState.commitTs != commitTs { | |
| a.bigTxnMetricState = &bigTxnMetricState{ | |
| startTs: startTs, | |
| commitTs: commitTs, | |
| largeTxnThresholdInBytes: largeTxnThresholdInBytes, | |
| } | |
| } | |
| if a.bigTxnMetricState == nil || | |
| a.bigTxnMetricState.startTs != startTs || | |
| a.bigTxnMetricState.commitTs != commitTs { | |
| a.finishPendingBigTxnMetricBefore(startTs, commitTs) | |
| a.bigTxnMetricState = &bigTxnMetricState{ | |
| startTs: startTs, | |
| commitTs: commitTs, | |
| largeTxnThresholdInBytes: largeTxnThresholdInBytes, | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/eventservice/dispatcher_stat.go` around lines 271 - 279, The big
transaction metric state is being replaced in the dispatcher without first
emitting the currently pending metric, which drops data when a new (startTs,
commitTs) pair arrives. Update the logic in the bigTxnMetricState handling block
to flush/emit the existing state before assigning a new bigTxnMetricState, using
the same code path that records the metric so the pending value is not lost.
| func newLargeTxnInsertSpill(dir string) (*largeTxnInsertSpill, error) { | ||
| if dir == "" { | ||
| return nil, errors.New("large txn spill dir is empty") | ||
| } | ||
| file, err := recordspill.NewRecordFile(dir, "eventservice-large-txn-insert-*.spill") | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return &largeTxnInsertSpill{ | ||
| path: file.Path(), | ||
| file: file, | ||
| }, nil | ||
| } | ||
|
|
||
| func (s *largeTxnInsertSpill) Append(entry *common.RawKVEntry) error { | ||
| if entry == nil { | ||
| return errors.New("cannot append nil RawKVEntry to large txn spill") | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Use predefined repository errors instead of ad-hoc errors.New.
newLargeTxnInsertSpill and Append raise generic errors via errors.New(...) rather than a predefined repository error (e.g. reusing cerror.ErrSpillFileOp, which already exists for this exact spill-file domain). As per coding guidelines, "Use predefined repository errors; see docs/agents/error-handling.md before changing error creation, wrapping, or propagation."
🧰 Tools
🪛 golangci-lint (2.12.2)
[error] 30-30: : # github.com/pingcap/ticdc/cmd/kafka-consumer [github.com/pingcap/ticdc/cmd/kafka-consumer.test]
cmd/kafka-consumer/consumer.go:30:22: undefined: kafka.ConfigMap
cmd/kafka-consumer/consumer.go:39:22: undefined: kafka.NewAdminClient
cmd/kafka-consumer/consumer.go:57:24: undefined: kafka.Error
cmd/kafka-consumer/consumer.go:58:62: undefined: kafka.ErrTransport
cmd/kafka-consumer/consumer.go:90:16: undefined: kafka.Consumer
cmd/kafka-consumer/writer.go:45:24: undefined: kafka.Offset
cmd/kafka-consumer/writer.go:59:79: undefined: kafka.Offset
cmd/kafka-consumer/writer.go:317:67: undefined: kafka.Message
cmd/kafka-consumer/writer.go:554:84: undefined: kafka.Offset
cmd/kafka-consumer/writer.go:581:97: undefined: kafka.Offset
cmd/kafka-consumer/consumer.go:58:62: too many errors
(typecheck)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/eventservice/large_txn_spill.go` around lines 30 - 48,
newLargeTxnInsertSpill and largeTxnInsertSpill.Append are creating ad-hoc errors
with errors.New instead of using the repository’s predefined spill-file error.
Update these paths to return the existing spill-related error type (for example
cerror.ErrSpillFileOp) and preserve the specific context for empty dir, record
file creation failure, and nil entry cases. Use the symbols
newLargeTxnInsertSpill, Append, and cerror.ErrSpillFileOp to locate and replace
the generic error creation consistently.
Source: Coding guidelines
| if state.startTs != startTs || state.commitTs != commitTs || state.tableID != tableID { | ||
| return nil, errors.Errorf( | ||
| "large txn spill state mismatch, existing start-ts: %d, commit-ts: %d, table-id: %d, new start-ts: %d, commit-ts: %d, table-id: %d", | ||
| state.startTs, state.commitTs, state.tableID, startTs, commitTs, tableID) | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Use predefined repository errors instead of ad-hoc errors.New/errors.Errorf.
Similar to large_txn_spill.go, this file raises generic errors (state mismatch, cleaned-up state, wrong phase) rather than predefined repository errors. As per coding guidelines, "Use predefined repository errors; see docs/agents/error-handling.md before changing error creation, wrapping, or propagation."
Also applies to: 124-158
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/eventservice/large_txn_state.go` around lines 61 - 65, The large-txn
state checks in large_txn_state.go are returning ad-hoc errors instead of the
repository’s predefined error types. Update the mismatch/cleanup/wrong-phase
paths in the relevant large txn state helpers (including the state comparison
block and the related cases in the same file) to return the existing shared
errors used by large_txn_spill.go, and follow the error-handling guidance in
docs/agents/error-handling.md so callers can reliably match these conditions.
Source: Coding guidelines
Signed-off-by: dongmen <414110582@qq.com>
|
@asddongmen: The following test failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
1 participant
What problem does this PR solve?
Issue Number: close #5596
EventBroker currently scans a whole large upstream transaction into memory before it can send DML to the event collector. Very large transactions can therefore cause OOM, especially when UK-changing updates cache deferred insert rows in memory.
What is changed and how it works?
This PR adds row-level eventstore scan resume tokens and uses them in EventBroker scanning so
transaction-atomicity=nonecan emit bounded fragments from one large transaction without sending resolved-ts for that commit-ts early.It also spills deferred insert halves of UK-changing updates to local disk, drains them after the original delete phase, and cleans spill state on reset/remove. EventCollector reset semantics remain checkpoint-ts based.
Check List
Tests
Questions
Will it cause performance regression or break compatibility?
No compatibility break is expected. The split path is only enabled when transaction atomicity allows splitting. It reduces EventBroker peak memory for large transactions, with local spill I/O for UK-changing update insert halves.
Do you need to update user documentation, design documentation or monitoring documentation?
No user documentation update is required for this internal scan behavior change.
Release note
Summary by CodeRabbit
large-txn-threshold-in-byteswith a 1MB default.