sink: add before field for avro protocol - #5154
Conversation
ti-chi-bot
Bot
added
the
release-note
📝 WalkthroughWalkthroughThis PR adds optional Avro before-value support for insert, update, and delete events. The option propagates through API and internal configuration. The encoder writes ChangesAvro Before-Value Feature
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant SinkConfig
participant AvroEncoder
participant Kafka
participant AvroDecoder
SinkConfig->>AvroEncoder: enable AvroIncludeBeforeValue
AvroEncoder->>Kafka: publish event with ticdcBefore
Kafka->>AvroDecoder: deliver Avro event
AvroDecoder->>AvroDecoder: decode before image and assemble DML event
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches 💡 2⚔️ Resolve merge conflicts 💡
🛠️ Fix failing CI checks 💡
🧪 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 the AvroIncludeBeforeValue configuration option, allowing Avro-encoded update and delete events to include their pre-row ('before') values under the _ticdc_before field. The changes span across configuration definitions, the Avro encoder/decoder implementations, and integration tests. Feedback on the changes highlights two key issues in the decoder: first, a suggestion to dynamically check the _tidb_op field in the decoded map to determine delete events rather than relying on the decoder's static configuration; second, a potential out-of-bounds panic when splitting the schema namespace if it does not contain a dot.
| if d.config.AvroIncludeBeforeValue { | ||
| isDelete = valueMap[tidbOp] == deleteOperation | ||
| } |
There was a problem hiding this comment.
Instead of relying on the decoder's local configuration d.config.AvroIncludeBeforeValue to determine if the event is a delete, it is much more robust to dynamically check for the presence of the _tidb_op field in the decoded valueMap. This prevents decoding failures or mismatches if the decoder's configuration does not perfectly align with the producer's configuration.
| if d.config.AvroIncludeBeforeValue { | |
| isDelete = valueMap[tidbOp] == deleteOperation | |
| } | |
| if op, ok := valueMap[tidbOp].(string); ok { | |
| isDelete = op == deleteOperation | |
| } |
| namespace := schema["namespace"].(string) | ||
| schemaName := strings.Split(namespace, ".")[1] |
There was a problem hiding this comment.
If the namespace string does not contain a dot (e.g., if the schema or keyspace is empty), strings.Split(namespace, ".")[1] will panic with an out-of-bounds index. It is safer to check the length of the split parts before accessing the index.
namespace := schema["namespace"].(string)
parts := strings.Split(namespace, ".")
var schemaName string
if len(parts) > 1 {
schemaName = parts[1]
} else {
schemaName = namespace
}
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
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/avro/decoder.go (1)
118-132: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winSet the outer message commit timestamp for value-bearing deletes.
For a delete with
tidbOp == deleteOperation,hasValueis true anddeleteCommitTsis zero. Lines 120-123 then skiptidbCommitTsbecauseisDeleteis true.common.NewDMLMessagereceives commit timestamp zero, althoughassembleEventsets the correct timestamp later. This can break message ordering and downstream timestamp handling for Avro deletes withbeforedata.Use
hasValueto decide when to readtidbCommitTs.Proposed fix
- if commitTs == 0 && !isDelete { + if commitTs == 0 && hasValue { commitTs = uint64(valueMap[tidbCommitTs].(int64)) }🤖 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 118 - 132, Update the commit timestamp selection in the decoder flow around decodeDMLPayload and NewDMLMessage to read valueMap[tidbCommitTs] whenever hasValue is true and deleteCommitTs is zero, including value-bearing deletes. Preserve deleteCommitTs when it is present, and ensure the corrected commitTs is passed to common.NewDMLMessage.
🤖 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.
Outside diff comments:
In `@pkg/sink/codec/avro/decoder.go`:
- Around line 118-132: Update the commit timestamp selection in the decoder flow
around decodeDMLPayload and NewDMLMessage to read valueMap[tidbCommitTs]
whenever hasValue is true and deleteCommitTs is zero, including value-bearing
deletes. Preserve deleteCommitTs when it is present, and ensure the corrected
commitTs is passed to common.NewDMLMessage.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: ea545c8e-7404-4e4c-b784-12b7b156c715
📒 Files selected for processing (11)
api/v2/model.goapi/v2/model_test.gopkg/config/sink.gopkg/sink/codec/avro/arvo.gopkg/sink/codec/avro/avro_test.gopkg/sink/codec/avro/decoder.gopkg/sink/codec/avro/helper.gopkg/sink/codec/common/config.gopkg/sink/codec/common/config_test.gotests/integration_tests/avro_basic/data/data.sqltests/integration_tests/avro_basic/run.sh
🚧 Files skipped from review as they are similar to previous changes (8)
- pkg/sink/codec/avro/helper.go
- pkg/config/sink.go
- api/v2/model_test.go
- tests/integration_tests/avro_basic/data/data.sql
- tests/integration_tests/avro_basic/run.sh
- api/v2/model.go
- pkg/sink/codec/common/config.go
- pkg/sink/codec/avro/arvo.go
Signed-off-by: wk989898 <nhsmwk@gmail.com>
|
/test pull-error-log-review |
|
[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 |
ti-chi-bot
Bot
added
approved
and removed
needs-1-more-lgtm
[LGTM Timeline notifier]Timeline:
|
|
/retest |
3 participants
What problem does this PR solve?
Issue Number: close #5153
What is changed and how it works?
This change adds the avro-include-before-value option to include the previous row in Avro messages.
When enabled:
The decoder reads _tidb_op from the message instead of relying on its local configuration and reconstructs the corresponding insert, update, or delete event. The change also adds API/TOML configuration support and related unit and integration tests.
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
Summary by CodeRabbit
New Features
Bug Fixes
Tests