consumer: use dml message instead of dml event - #5590
Conversation
ti-chi-bot
Bot
added
do-not-merge/work-in-progress
|
Skipping CI for Draft Pull Request. |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
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)
📝 WalkthroughWalkthroughDML decoding now returns deferred ChangesDML message pipeline
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 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 refactors the consumer and codec components to use a deferred DMLMessage abstraction instead of assembling DMLEvent objects immediately, which optimizes memory usage and performance. Additionally, the Canal-JSON decoder is updated to track DDL commit timestamps for more accurate table schema caching across column changes. The code review feedback suggests several defensive programming enhancements to prevent potential nil pointer dereferences and nil map panics, specifically by checking for nil messages in consumer.go and event_group.go, and ensuring the ddlCommitTs map is properly initialized and checked in canal_json_decoder.go.
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.
| row := decoder.NextDMLMessage().ToDMLEvent() | ||
| row.PhysicalTableID = tableID | ||
| c.appendRow2Group(row, fileIdx.EnableTableAcrossNodes) | ||
| c.appendMessage2Group(common.NewDMLMessageFromEvent(row), fileIdx.EnableTableAcrossNodes) |
There was a problem hiding this comment.
For defensive programming, we should check if decoder.NextDMLMessage() is nil before calling .ToDMLEvent() to avoid any potential nil pointer dereference panic.
dmlMsg := decoder.NextDMLMessage()
if dmlMsg == nil {
log.Panic("DML message is nil, it's not expected")
}
row := dmlMsg.ToDMLEvent()
row.PhysicalTableID = tableID
c.appendMessage2Group(common.NewDMLMessageFromEvent(row), fileIdx.EnableTableAcrossNodes)References
- Defensive programming: always check for nil pointers before dereferencing or calling methods on objects that can be nil.
| func (d *decoder) addDDLCommitTs(schema, table string, commitTs uint64) { | ||
| if schema == "" || table == "" || commitTs == 0 { | ||
| return | ||
| } |
There was a problem hiding this comment.
If d.ddlCommitTs is not initialized (e.g., if the decoder is created manually in tests or other packages without calling NewDecoder), calling d.ddlCommitTs[key] will panic. We should defensively initialize it if it is nil.
func (d *decoder) addDDLCommitTs(schema, table string, commitTs uint64) {
if schema == "" || table == "" || commitTs == 0 {
return
}
if d.ddlCommitTs == nil {
d.ddlCommitTs = make(map[tableNameKey][]uint64)
}
key := tableNameKey{schema: schema, table: table}References
- Defensive programming: ensure maps are initialized before writing to them to avoid nil map panics.
| func (d *decoder) getDDLCommitTs(schema, table string, commitTs uint64) uint64 { | ||
| if commitTs == 0 { | ||
| return 0 | ||
| } | ||
|
|
||
| commitTsList := d.ddlCommitTs[tableNameKey{schema: schema, table: table}] |
There was a problem hiding this comment.
If d.ddlCommitTs is nil, accessing it will panic. We should defensively check if d.ddlCommitTs is nil before querying it.
| func (d *decoder) getDDLCommitTs(schema, table string, commitTs uint64) uint64 { | |
| if commitTs == 0 { | |
| return 0 | |
| } | |
| commitTsList := d.ddlCommitTs[tableNameKey{schema: schema, table: table}] | |
| func (d *decoder) getDDLCommitTs(schema, table string, commitTs uint64) uint64 { | |
| if commitTs == 0 || d.ddlCommitTs == nil { | |
| return 0 | |
| } | |
| commitTsList := d.ddlCommitTs[tableNameKey{schema: schema, table: table}] |
References
- Defensive programming: check for nil maps before reading from them to avoid nil pointer panics.
| func AppendOrMergeDMLEvent(events []*commonEvent.DMLEvent, row *commonEvent.DMLEvent) []*commonEvent.DMLEvent { | ||
| var lastDMLEvent *commonEvent.DMLEvent | ||
| if len(events) > 0 { | ||
| lastDMLEvent = events[len(events)-1] | ||
| } | ||
|
|
||
| if lastDMLEvent == nil || lastDMLEvent.GetCommitTs() < row.GetCommitTs() { |
There was a problem hiding this comment.
If row is nil, calling row.GetCommitTs() will panic with a nil pointer dereference. We should defensively check if row is nil at the beginning of AppendOrMergeDMLEvent.
func AppendOrMergeDMLEvent(events []*commonEvent.DMLEvent, row *commonEvent.DMLEvent) []*commonEvent.DMLEvent {
if row == nil {
return events
}
var lastDMLEvent *commonEvent.DMLEvent
if len(events) > 0 {
lastDMLEvent = events[len(events)-1]
}
if lastDMLEvent == nil || lastDMLEvent.GetCommitTs() < row.GetCommitTs() {References
- Defensive programming: check for nil arguments before calling methods on them to avoid nil pointer panics.
|
/test kafka |
|
/test kafka |
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: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pkg/sink/codec/open/decoder.go (1)
195-271: 🚀 Performance & Scalability | 🟠 Major | ⚡ Quick winOpen decoder decompresses and decodes each DML value twice per message.
NextDMLMessagecallsrowTypeFromDMLValue(value)(Line 207) which decompresses anddecode()s the payload just to classifyRowType. The deferred closure then callsdecodeDMLMessage(&key, value)(Line 211), which decompresses and decodes the same bytes again from scratch to build the event. Unlike the avro/canal-json decoders in this PR, which decode once up front and reuse the parsed structures inside the deferred closure, this decoder pays for decompression + deserialization twice for every non-claim-check row on the hot decode path.Consider caching the decoded
*messageRowfromrowTypeFromDMLValueand passing it directly into the closure instead of re-decoding:⚡ Illustrative fix: decode the row once and reuse it
func (b *decoder) NextDMLMessage() *common.DMLMessage { if b.nextKey.Type != common.MessageTypeRow { log.Panic("message type is not row", zap.Any("messageType", b.nextKey.Type)) } key := *b.nextKey value := b.nextDMLValue() b.nextKey = nil rowType := commonType.RowTypeInsert + var decodedRow *messageRow if key.ClaimCheckLocation == "" { - rowType = b.rowTypeFromDMLValue(value) + decodedRow = b.decodeDMLValue(value) + rowType = rowTypeFromMessageRow(decodedRow) } tableID := tableIDAllocator.Allocate(key.Schema, key.Table) return common.NewDMLMessage(tableID, key.Schema, key.Table, key.Ts, rowType, func() *commonEvent.DMLEvent { - return b.decodeDMLMessage(&key, value) + return b.decodeDMLMessageWithRow(&key, decodedRow, value) }) } -func (b *decoder) rowTypeFromDMLValue(value []byte) commonType.RowType { - value, err := common.Decompress(b.config.LargeMessageHandle.LargeMessageHandleCompression, value) - ... - nextRow := new(messageRow) - nextRow.decode(value) - return rowTypeFromMessageRow(nextRow) -} +func (b *decoder) decodeDMLValue(value []byte) *messageRow { + value, err := common.Decompress(b.config.LargeMessageHandle.LargeMessageHandleCompression, value) + if err != nil { + log.Panic("decompress failed", ...) + } + row := new(messageRow) + row.decode(value) + return row +} -func (b *decoder) decodeDMLMessage(key *messageKey, value []byte) *commonEvent.DMLEvent { - value, err := common.Decompress(b.config.LargeMessageHandle.LargeMessageHandleCompression, value) - ... - nextRow := new(messageRow) - nextRow.decode(value) - - ctx := context.Background() +func (b *decoder) decodeDMLMessageWithRow(key *messageKey, nextRow *messageRow, value []byte) *commonEvent.DMLEvent { + ctx := context.Background() if key.ClaimCheckLocation != "" { return b.assembleEventFromClaimCheckStorage(ctx, key) } if key.OnlyHandleKey && b.upstreamTiDB != nil { return b.assembleHandleKeyOnlyDMLEvent(ctx, key, nextRow) } return b.assembleDMLEvent(key, nextRow) }(claim-check messages already skip the eager decode today, so
value/decode-on-demand can stay for that branch only)🤖 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/sink/codec/open/decoder.go` around lines 195 - 271, Update NextDMLMessage, rowTypeFromDMLValue, and decodeDMLMessage to decode non-claim-check DML payloads only once: cache the parsed *messageRow used to determine rowType and pass it into the deferred event-building path. Preserve claim-check handling by skipping eager value decoding and retaining its existing on-demand behavior.
🧹 Nitpick comments (2)
cmd/util/event_group.go (1)
60-68: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueManual slice-insert works correctly.
The append-nil/copy/shift-right pattern here is a correct manual re-implementation of
slices.Insert. Not a bug, just a minor style note: the standard-libraryslices.Insert(g.messages, i, message)would be more concise and equally correct if theslicesdependency removal wasn't intentional.🤖 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 `@cmd/util/event_group.go` around lines 60 - 68, No code change is required: the manual insertion in the force branch of the surrounding event-group method is correct. Optionally replace the append/copy/assignment sequence with slices.Insert using the computed index, but only if restoring or retaining the slices dependency is intentional.pkg/sink/codec/open/decoder.go (1)
337-339: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueDeferred closure mutates the captured
*messageKey(key.OnlyHandleKey = false).
assembleHandleKeyOnlyDMLEventmutateskey(the same pointer captured byNextDMLMessage's closure via&key). This is safe as long asToDMLEvent()is invoked exactly once per message; if it's ever invoked twice (retry, re-flush, etc.), the second call would skip the handle-key-only branch it already skipped, producing a different/incorrect result silently. No current caller appears to invoke it twice, so this is speculative, but worth a defensive note given the whole design leans on closures capturing mutable pointers.🤖 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/sink/codec/open/decoder.go` around lines 337 - 339, Remove the mutation of the captured messageKey in assembleHandleKeyOnlyDMLEvent by avoiding changes to key.OnlyHandleKey; preserve the method’s handle-key-only assembly behavior while ensuring repeated ToDMLEvent invocations produce the same result.
🤖 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/sink/codec/debezium/decoder.go`:
- Around line 172-173: The Debezium decoder must eagerly register allocated
table IDs before deferred DML event assembly. In
pkg/sink/codec/debezium/decoder.go:172-173, update the allocation flow around
Allocate to immediately call AddBlockTableID, and remove the duplicate
registration in queryTableInfoFromPayload. Apply the same immediate registration
after Allocate in pkg/sink/codec/canal/canal_json_txn_decoder.go:111-114 so both
decoders preserve accurate blocked-table partition mapping.
---
Outside diff comments:
In `@pkg/sink/codec/open/decoder.go`:
- Around line 195-271: Update NextDMLMessage, rowTypeFromDMLValue, and
decodeDMLMessage to decode non-claim-check DML payloads only once: cache the
parsed *messageRow used to determine rowType and pass it into the deferred
event-building path. Preserve claim-check handling by skipping eager value
decoding and retaining its existing on-demand behavior.
---
Nitpick comments:
In `@cmd/util/event_group.go`:
- Around line 60-68: No code change is required: the manual insertion in the
force branch of the surrounding event-group method is correct. Optionally
replace the append/copy/assignment sequence with slices.Insert using the
computed index, but only if restoring or retaining the slices dependency is
intentional.
In `@pkg/sink/codec/open/decoder.go`:
- Around line 337-339: Remove the mutation of the captured messageKey in
assembleHandleKeyOnlyDMLEvent by avoiding changes to key.OnlyHandleKey; preserve
the method’s handle-key-only assembly behavior while ensuring repeated
ToDMLEvent invocations produce the same result.
🪄 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: 75d0c5be-ec56-4822-be7f-74da0c52d7e1
📒 Files selected for processing (28)
cmd/kafka-consumer/writer.gocmd/kafka-consumer/writer_test.gocmd/pulsar-consumer/consumer.gocmd/pulsar-consumer/writer.gocmd/pulsar-consumer/writer_test.gocmd/storage-consumer/consumer.gocmd/util/event_group.gocmd/util/event_group_test.gopkg/sink/codec/avro/avro_test.gopkg/sink/codec/avro/decoder.gopkg/sink/codec/avro/encoder_test.gopkg/sink/codec/canal/canal_json_decoder.gopkg/sink/codec/canal/canal_json_encoder_test.gopkg/sink/codec/canal/canal_json_test.gopkg/sink/codec/canal/canal_json_txn_decoder.gopkg/sink/codec/common/decoder.gopkg/sink/codec/common/table_info_cache.gopkg/sink/codec/csv/csv_decoder.gopkg/sink/codec/csv/csv_decoder_test.gopkg/sink/codec/debezium/avro_decoder.gopkg/sink/codec/debezium/avro_test.gopkg/sink/codec/debezium/debezium_test.gopkg/sink/codec/debezium/decoder.gopkg/sink/codec/open/decoder.gopkg/sink/codec/open/encoder_test.gopkg/sink/codec/simple/decoder.gopkg/sink/codec/simple/decoder_test.gopkg/sink/codec/simple/encoder_test.go
| tableID := tableIDAllocator.Allocate(schemaName, tableName) | ||
| d.clear() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | 🏗️ Heavy lift
Eagerly register allocated table IDs to prevent out-of-order DDL/DML execution.
Both decoders allocate a table ID for the incoming DML message but fail to eagerly register it with the allocator's blocked list. Because event assembly (ToDMLEvent) is now deferred, delaying or omitting AddBlockTableID means an interleaving DDL event will not include these physical table IDs in its ddl.GetBlockedTables() list. Consequently, the consumer will mistakenly skip the pre-DDL flush for these buffered DMLs, violating execution order and leading to schema errors (such as the "Unknown column" downstream failure).
pkg/sink/codec/debezium/decoder.go#L172-L173: AddtableIDAllocator.AddBlockTableID(schemaName, tableName, tableID)immediately after allocating thetableID. (You can safely remove the duplicate registration insidequeryTableInfoFromPayloadsince it's now covered eagerly).pkg/sink/codec/canal/canal_json_txn_decoder.go#L111-L114: AddtableIDAllocator.AddBlockTableID(schemaName, tableName, tableID)immediately after allocating thetableIDhere as well to preserve accurate partition mapping on flush.
📍 Affects 2 files
pkg/sink/codec/debezium/decoder.go#L172-L173(this comment)pkg/sink/codec/canal/canal_json_txn_decoder.go#L111-L114
🤖 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/sink/codec/debezium/decoder.go` around lines 172 - 173, The Debezium
decoder must eagerly register allocated table IDs before deferred DML event
assembly. In pkg/sink/codec/debezium/decoder.go:172-173, update the allocation
flow around Allocate to immediately call AddBlockTableID, and remove the
duplicate registration in queryTableInfoFromPayload. Apply the same immediate
registration after Allocate in
pkg/sink/codec/canal/canal_json_txn_decoder.go:111-114 so both decoders preserve
accurate blocked-table partition mapping.
|
/test all |
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
pkg/sink/codec/avro/decoder.go (2)
269-284: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winPrevent native panics on missing
connect.parametersortidb_type.If
connect.parametersortidb_typeis absent in the decoded Avro schema (e.g., if the schema was not generated with TiDB extensions), type assertions like.(map[string]any)or.(string)will fail onnilinterfaces. This causes a native Go panic, bypassing standard error handling and abruptly crashing the consumer when processing malformed or unexpected external schemas.Safely assert these types and return an error instead.
🛡️ Proposed fix to add safe type assertions
var holder map[string]any switch ty := field["type"].(type) { case []any: if m, ok := ty[0].(map[string]any); ok { - holder = m["connect.parameters"].(map[string]any) + if params, ok := m["connect.parameters"].(map[string]any); ok { + holder = params + } } else if m, ok := ty[1].(map[string]any); ok { - holder = m["connect.parameters"].(map[string]any) + if params, ok := m["connect.parameters"].(map[string]any); ok { + holder = params + } } else { log.Panic("type info is anything else", zap.Any("typeInfo", field["type"])) } case map[string]any: - holder = ty["connect.parameters"].(map[string]any) + if params, ok := ty["connect.parameters"].(map[string]any); ok { + holder = params + } default: log.Panic("type info is anything else", zap.Any("typeInfo", field["type"])) } + + if holder == nil { + return nil, errors.ErrCodecDecode.GenWithStack("connect.parameters not found or invalid in schema field") + } - tidbType := holder["tidb_type"].(string) + tidbType, ok := holder["tidb_type"].(string) + if !ok { + return nil, errors.ErrCodecDecode.GenWithStack("tidb_type not found or invalid in connect.parameters") + }🤖 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/sink/codec/avro/decoder.go` around lines 269 - 284, Update the schema type handling around the field decoder’s holder extraction and tidbType assignment to safely validate connect.parameters as map[string]any and tidb_type as string instead of using unchecked assertions. On missing or invalid values, return the decoder’s standard error rather than triggering a native panic; preserve the existing handling for valid map and array schema forms.
307-307: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winPrevent out-of-bounds panic on namespace split.
If the Avro schema's
namespacedoes not contain a dot (e.g.,"myschema"instead of"myschema.mytable"),strings.Split(namespace, ".")[1]will panic with an index out-of-bounds error. Additionally, ifnamespaceornameis absent, the type assertion.(string)will also trigger a native panic, crashing the consumer.Safely assert these types, validate the namespace format, and return an error instead of crashing.
🛠️ Proposed fix to handle namespace extraction safely
Update the function to return an error safely:
-func schemaAndTableName(schema map[string]any) (string, string) { - namespace := schema["namespace"].(string) - return strings.Split(namespace, ".")[1], schema["name"].(string) -} +func schemaAndTableName(schema map[string]any) (string, string, error) { + ns, ok := schema["namespace"].(string) + if !ok { + return "", "", errors.ErrCodecDecode.GenWithStack("schema namespace must be a string") + } + parts := strings.Split(ns, ".") + if len(parts) < 2 { + return "", "", errors.ErrCodecDecode.GenWithStack("schema namespace must contain a dot, got: %s", ns) + } + name, ok := schema["name"].(string) + if !ok { + return "", "", errors.ErrCodecDecode.GenWithStack("schema name must be a string") + } + return parts[1], name, nil +}And handle the error in
assembleEvent:- schemaName, tableName := schemaAndTableName(schema) + schemaName, tableName, err := schemaAndTableName(schema) + if err != nil { + return nil, err + }Also applies to: 338-341
🤖 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/sink/codec/avro/decoder.go` at line 307, Update schemaAndTableName to safely validate namespace and name type assertions, require a dotted namespace before extracting schema and table components, and return an error for missing or malformed values instead of panicking. Update assembleEvent to handle and propagate the returned error from schemaAndTableName.
🤖 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/sink/codec/avro/decoder.go`:
- Around line 249-252: In pkg/sink/codec/avro/decoder.go, replace the four
errors.New calls in the schema field parsing and value/commit timestamp lookup
paths with errors.ErrCodecDecode.GenWithStack: use “schema fields should be an
array” for the fields assertion at lines 249-252, retain “schema field should be
a map” at lines 258-261, include colName in “value not found for column: %s” at
lines 288-290, and use “commit ts not found” at lines 312-314.
---
Outside diff comments:
In `@pkg/sink/codec/avro/decoder.go`:
- Around line 269-284: Update the schema type handling around the field
decoder’s holder extraction and tidbType assignment to safely validate
connect.parameters as map[string]any and tidb_type as string instead of using
unchecked assertions. On missing or invalid values, return the decoder’s
standard error rather than triggering a native panic; preserve the existing
handling for valid map and array schema forms.
- Line 307: Update schemaAndTableName to safely validate namespace and name type
assertions, require a dotted namespace before extracting schema and table
components, and return an error for missing or malformed values instead of
panicking. Update assembleEvent to handle and propagate the returned error from
schemaAndTableName.
🪄 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: 631ee8d7-f08b-4163-ab38-ef5a4eca28a3
📒 Files selected for processing (5)
cmd/kafka-consumer/writer.gocmd/pulsar-consumer/writer.gocmd/storage-consumer/consumer.gopkg/sink/codec/avro/decoder.gopkg/sink/codec/open/encoder_test.go
💤 Files with no reviewable changes (1)
- pkg/sink/codec/open/encoder_test.go
🚧 Files skipped from review as they are similar to previous changes (3)
- cmd/storage-consumer/consumer.go
- cmd/kafka-consumer/writer.go
- cmd/pulsar-consumer/writer.go
| fields, ok := schema["fields"].([]any) | ||
| if !ok { | ||
| return nil, errors.New("schema fields should be a map") | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Use predefined repository errors instead of errors.New.
The use of errors.New violates the coding guidelines that require using predefined repository errors for error creation. Additionally, the error message for the fields type assertion inaccurately expects a "map" when checking for a slice ([]any). Use errors.ErrCodecDecode.GenWithStack consistently as seen elsewhere in this file.
pkg/sink/codec/avro/decoder.go#L249-L252: replaceerrors.New("schema fields should be a map")witherrors.ErrCodecDecode.GenWithStack("schema fields should be an array").pkg/sink/codec/avro/decoder.go#L258-L261: replaceerrors.New("schema field should be a map")witherrors.ErrCodecDecode.GenWithStack("schema field should be a map").pkg/sink/codec/avro/decoder.go#L288-L290: replaceerrors.New("value not found")witherrors.ErrCodecDecode.GenWithStack("value not found for column: %s", colName).pkg/sink/codec/avro/decoder.go#L312-L314: replaceerrors.New("commit ts not found")witherrors.ErrCodecDecode.GenWithStack("commit ts not found").
📍 Affects 1 file
pkg/sink/codec/avro/decoder.go#L249-L252(this comment)pkg/sink/codec/avro/decoder.go#L258-L261pkg/sink/codec/avro/decoder.go#L288-L290pkg/sink/codec/avro/decoder.go#L312-L314
🤖 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/sink/codec/avro/decoder.go` around lines 249 - 252, In
pkg/sink/codec/avro/decoder.go, replace the four errors.New calls in the schema
field parsing and value/commit timestamp lookup paths with
errors.ErrCodecDecode.GenWithStack: use “schema fields should be an array” for
the fields assertion at lines 249-252, retain “schema field should be a map” at
lines 258-261, include colName in “value not found for column: %s” at lines
288-290, and use “commit ts not found” at lines 312-314.
Source: Coding guidelines
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pkg/sink/codec/debezium/decoder.go (1)
278-288: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winMove blocked-table registration into the decode path
tableIDAllocator.Allocate(...)already happens inNextDMLMessage; onlyAddBlockTableID(...)is deferred here throughqueryTableInfoFromPayload. Register the table ID before clearing decoder state so interleaved DDLs don’t miss this table inGetBlockedTables().🤖 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/sink/codec/debezium/decoder.go` around lines 278 - 288, Move the AddBlockTableID registration from queryTableInfoFromPayload into the decode path in NextDMLMessage, immediately after Allocate and before decoder state is cleared. Remove the deferred registration from queryTableInfoFromPayload while preserving the existing table ID and table metadata construction.
🧹 Nitpick comments (2)
pkg/sink/codec/debezium/decoder.go (2)
336-337: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valuePreallocate the map with the expected capacity.
Since the maximum number of columns is known, preallocating the map can reduce reallocations during assembly.
⚡ Proposed fix
func assembleColumnData(data map[string]any, columns []*timodel.ColumnInfo, timeZone *time.Location) map[string]any { - result := make(map[string]any, 0) + result := make(map[string]any, len(columns))🤖 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/sink/codec/debezium/decoder.go` around lines 336 - 337, Update assembleColumnData to initialize its result map with capacity len(columns), using the known column count instead of zero while preserving the existing assembly behavior.
336-337: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valuePreallocate the map with the expected capacity.
Since the maximum number of columns is known, preallocating the map can reduce unnecessary memory allocations and rehashing during column data assembly.
⚡ Proposed fix
func assembleColumnData(data map[string]any, columns []*timodel.ColumnInfo, timeZone *time.Location) map[string]any { - result := make(map[string]any, 0) + result := make(map[string]any, len(columns))🤖 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/sink/codec/debezium/decoder.go` around lines 336 - 337, Update assembleColumnData to initialize its result map with columns’ expected capacity rather than zero, using len(columns) as the map size hint while preserving the existing assembly behavior.
🤖 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/sink/codec/debezium/decoder.go`:
- Around line 527-530: Update the error returned in the schema validation block
of the decoder to say “decode schema failed” instead of “decode payload failed,”
while preserving the existing data context and control flow.
- Around line 527-530: Update the error returned by the schema type assertion in
the decoder flow to say “decode schema failed” instead of “decode payload
failed,” while preserving the existing data context and control flow.
---
Outside diff comments:
In `@pkg/sink/codec/debezium/decoder.go`:
- Around line 278-288: Move the AddBlockTableID registration from
queryTableInfoFromPayload into the decode path in NextDMLMessage, immediately
after Allocate and before decoder state is cleared. Remove the deferred
registration from queryTableInfoFromPayload while preserving the existing table
ID and table metadata construction.
---
Nitpick comments:
In `@pkg/sink/codec/debezium/decoder.go`:
- Around line 336-337: Update assembleColumnData to initialize its result map
with capacity len(columns), using the known column count instead of zero while
preserving the existing assembly behavior.
- Around line 336-337: Update assembleColumnData to initialize its result map
with columns’ expected capacity rather than zero, using len(columns) as the map
size hint while preserving the existing assembly behavior.
🪄 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: 944278d2-ba93-4ebc-8167-e4e6d9f99517
📒 Files selected for processing (3)
pkg/sink/codec/canal/canal_json_test.gopkg/sink/codec/debezium/decoder.gopkg/sink/codec/simple/decoder_test.go
🚧 Files skipped from review as they are similar to previous changes (2)
- pkg/sink/codec/simple/decoder_test.go
- pkg/sink/codec/canal/canal_json_test.go
| schema, ok := v["schema"].(map[string]any) | ||
| if !ok { | ||
| return nil, nil, fmt.Errorf("decode payload failed, data: %+v", v) | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Fix the error message to accurately reflect the failure.
The error message indicates that decoding the payload failed, but this block is actually validating the schema. Update the message to "decode schema failed" to avoid confusion during debugging.
✏️ Proposed fix
schema, ok := v["schema"].(map[string]any)
if !ok {
- return nil, nil, fmt.Errorf("decode payload failed, data: %+v", v)
+ return nil, nil, fmt.Errorf("decode schema failed, data: %+v", v)
}📝 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.
| schema, ok := v["schema"].(map[string]any) | |
| if !ok { | |
| return nil, nil, fmt.Errorf("decode payload failed, data: %+v", v) | |
| } | |
| schema, ok := v["schema"].(map[string]any) | |
| if !ok { | |
| return nil, nil, fmt.Errorf("decode schema failed, data: %+v", v) | |
| } |
🤖 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/sink/codec/debezium/decoder.go` around lines 527 - 530, Update the error
returned in the schema validation block of the decoder to say “decode schema
failed” instead of “decode payload failed,” while preserving the existing data
context and control flow.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Fix the error message to accurately reflect the failure.
The error message indicates that decoding the payload failed, but this block is actually validating the schema. Update the message to "decode schema failed" to avoid confusion during debugging.
✏️ Proposed fix
schema, ok := v["schema"].(map[string]any)
if !ok {
- return nil, nil, fmt.Errorf("decode payload failed, data: %+v", v)
+ return nil, nil, fmt.Errorf("decode schema failed, data: %+v", v)
}📝 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.
| schema, ok := v["schema"].(map[string]any) | |
| if !ok { | |
| return nil, nil, fmt.Errorf("decode payload failed, data: %+v", v) | |
| } | |
| schema, ok := v["schema"].(map[string]any) | |
| if !ok { | |
| return nil, nil, fmt.Errorf("decode schema failed, data: %+v", v) | |
| } |
🤖 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/sink/codec/debezium/decoder.go` around lines 527 - 530, Update the error
returned by the schema type assertion in the decoder flow to say “decode schema
failed” instead of “decode payload failed,” while preserving the existing data
context and control flow.
|
/test all |
|
@wk989898: 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. |
ti-chi-bot
Bot
added
needs-1-more-lgtm
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: 3AceShowHand, lidezhu The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
[LGTM Timeline notifier]Timeline:
|
wk989898
added
the
do-not-merge/hold
wk989898
removed
the
do-not-merge/hold
|
/cherry-pick release-nextgen-202603 |
wk989898
added
the
needs-cherry-pick-release-8.5
|
@wk989898: new pull request created to branch DetailsIn response to this:
Instructions 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 ti-community-infra/tichi repository. |
|
In response to a cherrypick label: new pull request created to branch |
* This is an automated cherry-pick of #5590 Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io> * update Signed-off-by: wk989898 <nhsmwk@gmail.com> * update Signed-off-by: wk989898 <nhsmwk@gmail.com> * fix Signed-off-by: wk989898 <nhsmwk@gmail.com> --------- Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io> Signed-off-by: wk989898 <nhsmwk@gmail.com> Co-authored-by: nhsmw <nhsmwk@gmail.com>
4 participants
What problem does this PR solve?
Issue Number: close #5587
What is changed and how it works?
DML decoding now returns DMLMessage instead of eagerly building DMLEvent. NextDMLMessage() only exposes metadata such as tableID, schema, table, rowType, and commitTs.
DMLEvent construction is deferred until flush time. Event groups store DMLMessage and sort/resolve by commitTs. Only when DML is actually flushed does the consumer call message.ToDMLEvent().
Simple protocol cached DML now also uses DMLMessage. If a simple DML arrives before its table info, it is cached as the raw message. After the DDL/table info arrives, the cache releases DMLMessage, not prebuilt DMLEvent.
The deferred toDMLEvent callbacks were made safe for delayed/asynchronous use. They no longer restore data into decoder cursor fields such as d.msg, d.keyPayload, or d.valuePayload. Instead, they convert using data captured by the DMLMessage itself.
Message payloads are no longer mutated during conversion. Simple and canal-json column formatting now returns new maps instead of modifying the original message maps in place.
Shared codec caches were protected. The fake table ID allocator and canal-json table-info/DDL commit-ts caches now use locks so deferred conversion cannot race with later decoder/cache updates.
Kafka partition validation is still preserved. messageWithPartitionCheck wraps DMLMessage.ToDMLEvent() so the partition check runs only when the DML is actually converted at flush time.
Check List
Tests
Questions
Will it cause performance regression or break compatibility?
Do you need to update user documentation, design documentation or monitoring documentation?
Release note
Summary by CodeRabbit
New Features
Bug Fixes
Refactor