fix(cdc): use native LSN tuple for change deduplication - #249
Conversation
Replace content-based FNV-1a hash with native CDC unique identifier for row deduplication. - Remove ComputeChangeID (FNV-1a hash over txid, table, data, lsn, operation) - Add ComputeNativeID using __ + __ + __ tuple - The native CDC key is guaranteed unique by SQL Server CDC engine, eliminating potential hash collisions from content-based approach - Add SeqVal field to Change struct to capture sequence value within transaction
WalkthroughThis PR implements CDC change deduplication by replacing content-based hash IDs with SQL Server CDC's native row-level unique identifier (LSN + SeqVal + Operation). The capturer now extracts the Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. 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 |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 4 issues, and left some high level feedback:
- The native CDC ID formatting logic (
LSN:SeqVal:Op) is duplicated betweenComputeNativeIDand the integration test (fmt.Sprintf("%s:%s:%d", ...)); consider routing all ID construction throughComputeNativeID(or a shared helper) to avoid format drift. - In
GetChanges,SeqValdefaults tonilif__$seqvalis missing or cannot be decoded, which then flows intoComputeNativeID; consider validating thatSeqValis non-empty and failing or logging explicitly when it’s missing to avoid generating ambiguous IDs.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The native CDC ID formatting logic (`LSN:SeqVal:Op`) is duplicated between `ComputeNativeID` and the integration test (`fmt.Sprintf("%s:%s:%d", ...)`); consider routing all ID construction through `ComputeNativeID` (or a shared helper) to avoid format drift.
- In `GetChanges`, `SeqVal` defaults to `nil` if `__$seqval` is missing or cannot be decoded, which then flows into `ComputeNativeID`; consider validating that `SeqVal` is non-empty and failing or logging explicitly when it’s missing to avoid generating ambiguous IDs.
## Individual Comments
### Comment 1
<location path="internal/cdc/capturer.go" line_range="417" />
<code_context>
-
- return fmt.Sprintf("%016x", hash)
+// ComputeNativeID computes a native CDC row ID from LSN tuple.
+func ComputeNativeID(lsn, seqval []byte, op int) string {
+ return hex.EncodeToString(lsn) + ":" + hex.EncodeToString(seqval) + ":" + strconv.Itoa(op)
}
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Consider validating that LSN and SeqVal are non-nil/non-empty when building the native ID to avoid weakened uniqueness guarantees.
`ComputeNativeID` will generate IDs even when `lsn` or `seqval` are nil/empty, producing values like `::<op>` or `<lsn>::<op>`. This no longer matches the documented uniqueness (`__$start_lsn + __$seqval + __$operation`) and can cause collisions if `seqval` is missing for multiple rows at the same LSN/operation. Consider returning an error/empty string or failing earlier when `lsn`/`seqval` are nil/zero-length, rather than silently weakening uniqueness.
Suggested implementation:
```golang
import (
"encoding/hex"
"strconv"
)
// ComputeNativeID computes a native CDC row ID from LSN tuple.
// If either lsn or seqval are nil or empty, it returns an empty string
// instead of generating a weakened/non-unique native ID.
func ComputeNativeID(lsn, seqval []byte, op int) string {
if len(lsn) == 0 || len(seqval) == 0 {
return ""
}
return hex.EncodeToString(lsn) + ":" + hex.EncodeToString(seqval) + ":" + strconv.Itoa(op)
}
```
1. Ensure that `internal/cdc/capturer.go` already has an `import` block; if it does, move the `encoding/hex` and `strconv` imports into that existing block instead of introducing a new one.
2. Audit all call sites of `ComputeNativeID` to handle the possibility of an empty string return (e.g., treat it as an error condition or avoid using it as a primary key/native ID when empty).
3. Optionally add unit tests to cover the cases where `lsn` or `seqval` are `nil` or empty, and verify that `ComputeNativeID` returns an empty string in those cases.
</issue_to_address>
### Comment 2
<location path="internal/cdc/query.go" line_range="214-216" />
<code_context>
}
}
+ // Extract seqval (sequence value within transaction - native CDC unique key part 2)
+ var seqval []byte
+ if idx, ok := colIndex["__$seqval"]; ok {
+ if s, ok := dest[idx].(scannerpkg.DBType); ok {
+ if val, err := s.Value(); err == nil && val != nil {
</code_context>
<issue_to_address>
**issue (bug_risk):** The seqval extraction path ignores Value() errors, which may hide issues and produce empty SeqVal silently.
Here `s.Value()` errors are ignored (`if val, err := s.Value(); err == nil && val != nil { ... }`), which means type/driver issues will silently leave `seqval` as nil while downstream logic assumes `LSN+SeqVal+Op` is a reliable native key. Please surface or at least log these errors (even at debug level) so CDC uniqueness issues and schema/driver mismatches are detectable instead of failing silently.
</issue_to_address>
### Comment 3
<location path="internal/cdc/capturer_test.go" line_range="294-303" />
<code_context>
+// TestComputeNativeID_Deterministic verifies same inputs produce same ID
</code_context>
<issue_to_address>
**suggestion (testing):** Add tests for the exact native ID format and edge cases (empty/nil LSN/SeqVal).
The new tests cover determinism and component sensitivity, but they don’t exercise the full ID contract of `ComputeNativeID`.
Specifically, there are no assertions that:
- The output uses the exact `hex(lsn) + ":" + hex(seqval) + ":" + op` format (including colons, lowercase hex, and decimal `op`).
- Behavior with empty or nil `lsn`/`seqval` is defined and intentional (e.g., what we get for `nil, nil, 2`).
- Leading zeros in the hex encoding are preserved (e.g., `[]byte{0x00, 0x10}` vs `[]byte{0x10}`).
Consider adding tests like:
- `TestComputeNativeID_Format` asserting `ComputeNativeID([]byte{0x00, 0x10}, []byte{0x00, 0x0b}, 2) == "00000010:00000b:2"`.
- `TestComputeNativeID_EmptyLSNOrSeqVal` documenting and verifying behavior for empty/nil inputs.
That will lock in the ID format and catch subtle regressions early.
Suggested implementation:
```golang
}
// TestComputeNativeID_Deterministic verifies same inputs produce same ID.
func TestComputeNativeID_Deterministic(t *testing.T) {
lsn := []byte{0x00, 0x00, 0x2B, 0x00, 0x01, 0xD8}
seqval := []byte{0x01, 0x02, 0x03, 0x04}
id1 := ComputeNativeID(lsn, seqval, 2)
id2 := ComputeNativeID(lsn, seqval, 2)
if id1 != id2 {
t.Errorf("Same inputs must produce same ID: %s != %s", id1, id2)
}
}
// TestComputeNativeID_Format verifies the exact ID format, including
// hex encoding, colons, lowercase, and decimal op.
func TestComputeNativeID_Format(t *testing.T) {
// Leading zeros in the byte slices must be preserved in the hex representation.
lsn := []byte{0x00, 0x00, 0x00, 0x10}
seqval := []byte{0x00, 0x00, 0x0b}
op := 2
got := ComputeNativeID(lsn, seqval, op)
want := "00000010:00000b:2"
if got != want {
t.Fatalf("unexpected native ID format: got %q, want %q", got, want)
}
}
// TestComputeNativeID_EmptyLSNOrSeqVal documents and verifies behavior
// when LSN and/or SeqVal are empty or nil.
func TestComputeNativeID_EmptyLSNOrSeqVal(t *testing.T) {
tests := []struct {
name string
lsn []byte
seqval []byte
op int
want string
}{
{
name: "nil LSN and nil SeqVal",
lsn: nil,
seqval: nil,
op: 2,
// Empty hex components produce empty strings around the colons.
want: "::2",
},
{
name: "empty LSN and non-empty SeqVal",
lsn: []byte{},
seqval: []byte{0x01},
op: 3,
want: ":01:3",
},
{
name: "non-empty LSN and empty SeqVal",
lsn: []byte{0x0a},
seqval: []byte{},
op: 4,
want: "0a::4",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := ComputeNativeID(tt.lsn, tt.seqval, tt.op)
if got != tt.want {
t.Fatalf("ComputeNativeID(%v, %v, %d) = %q, want %q",
tt.lsn, tt.seqval, tt.op, got, tt.want)
}
})
}
}
```
These tests assume the current `ComputeNativeID` implementation:
- Produces lowercase hex without a `0x` prefix.
- Preserves leading zeros via padding to fixed widths (e.g., `"00000010"` and `"00000b"`).
- Renders empty/nil slices as empty hex components, resulting in strings like `"::2"`, `":01:3"`, and `"0a::4"`.
If the actual behavior differs (for example, different padding width or non-empty representation for nil/empty), update the `want` strings in `TestComputeNativeID_Format` and `TestComputeNativeID_EmptyLSNOrSeqVal` to match the real, intended contract of `ComputeNativeID`.
</issue_to_address>
### Comment 4
<location path="internal/store/sqlite/store_test.go" line_range="207" />
<code_context>
+func TestStore_Write_SameLSNDifferentContent(t *testing.T) {
</code_context>
<issue_to_address>
**suggestion (testing):** Add a complementary test that verifies duplicate native IDs are de-duplicated and logged.
Given the new native ID semantics and `INSERT OR IGNORE`, there’s also a key negative case to cover:
- Two changes with the *same* `ID` (same LSN+SeqVal+Op) should produce only one row and emit a `slog.Warn` for the ignored duplicate.
Please add a test (e.g. `TestStore_Write_DuplicateID_IsIgnoredAndLogged`) that:
- Writes a batch with two `core.Change` entries sharing the same `ID` but different payloads.
- Asserts that only one row is present in `changes`.
- Verifies via a test logger or hook that a warning is logged when the duplicate is ignored.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
|
||
| return fmt.Sprintf("%016x", hash) | ||
| // ComputeNativeID computes a native CDC row ID from LSN tuple. | ||
| func ComputeNativeID(lsn, seqval []byte, op int) string { |
There was a problem hiding this comment.
suggestion (bug_risk): Consider validating that LSN and SeqVal are non-nil/non-empty when building the native ID to avoid weakened uniqueness guarantees.
ComputeNativeID will generate IDs even when lsn or seqval are nil/empty, producing values like ::<op> or <lsn>::<op>. This no longer matches the documented uniqueness (__$start_lsn + __$seqval + __$operation) and can cause collisions if seqval is missing for multiple rows at the same LSN/operation. Consider returning an error/empty string or failing earlier when lsn/seqval are nil/zero-length, rather than silently weakening uniqueness.
Suggested implementation:
import (
"encoding/hex"
"strconv"
)
// ComputeNativeID computes a native CDC row ID from LSN tuple.
// If either lsn or seqval are nil or empty, it returns an empty string
// instead of generating a weakened/non-unique native ID.
func ComputeNativeID(lsn, seqval []byte, op int) string {
if len(lsn) == 0 || len(seqval) == 0 {
return ""
}
return hex.EncodeToString(lsn) + ":" + hex.EncodeToString(seqval) + ":" + strconv.Itoa(op)
}
- Ensure that
internal/cdc/capturer.goalready has animportblock; if it does, move theencoding/hexandstrconvimports into that existing block instead of introducing a new one. - Audit all call sites of
ComputeNativeIDto handle the possibility of an empty string return (e.g., treat it as an error condition or avoid using it as a primary key/native ID when empty). - Optionally add unit tests to cover the cases where
lsnorseqvalarenilor empty, and verify thatComputeNativeIDreturns an empty string in those cases.
| // Extract seqval (sequence value within transaction - native CDC unique key part 2) | ||
| var seqval []byte | ||
| if idx, ok := colIndex["__$seqval"]; ok { |
There was a problem hiding this comment.
issue (bug_risk): The seqval extraction path ignores Value() errors, which may hide issues and produce empty SeqVal silently.
Here s.Value() errors are ignored (if val, err := s.Value(); err == nil && val != nil { ... }), which means type/driver issues will silently leave seqval as nil while downstream logic assumes LSN+SeqVal+Op is a reliable native key. Please surface or at least log these errors (even at debug level) so CDC uniqueness issues and schema/driver mismatches are detectable instead of failing silently.
| // TestComputeNativeID_Deterministic verifies same inputs produce same ID | ||
| func TestComputeNativeID_Deterministic(t *testing.T) { | ||
| lsn := []byte{0x00, 0x00, 0x2B, 0x00, 0x01, 0xD8} | ||
| seqval := []byte{0x01, 0x02, 0x03, 0x04} | ||
| id1 := ComputeNativeID(lsn, seqval, 2) | ||
| id2 := ComputeNativeID(lsn, seqval, 2) | ||
|
|
||
| if id1 != id2 { | ||
| t.Errorf("Same inputs must produce same ID: %s != %s", id1, id2) | ||
| } |
There was a problem hiding this comment.
suggestion (testing): Add tests for the exact native ID format and edge cases (empty/nil LSN/SeqVal).
The new tests cover determinism and component sensitivity, but they don’t exercise the full ID contract of ComputeNativeID.
Specifically, there are no assertions that:
- The output uses the exact
hex(lsn) + ":" + hex(seqval) + ":" + opformat (including colons, lowercase hex, and decimalop). - Behavior with empty or nil
lsn/seqvalis defined and intentional (e.g., what we get fornil, nil, 2). - Leading zeros in the hex encoding are preserved (e.g.,
[]byte{0x00, 0x10}vs[]byte{0x10}).
Consider adding tests like:
TestComputeNativeID_FormatassertingComputeNativeID([]byte{0x00, 0x10}, []byte{0x00, 0x0b}, 2) == "00000010:00000b:2".TestComputeNativeID_EmptyLSNOrSeqValdocumenting and verifying behavior for empty/nil inputs.
That will lock in the ID format and catch subtle regressions early.
Suggested implementation:
}
// TestComputeNativeID_Deterministic verifies same inputs produce same ID.
func TestComputeNativeID_Deterministic(t *testing.T) {
lsn := []byte{0x00, 0x00, 0x2B, 0x00, 0x01, 0xD8}
seqval := []byte{0x01, 0x02, 0x03, 0x04}
id1 := ComputeNativeID(lsn, seqval, 2)
id2 := ComputeNativeID(lsn, seqval, 2)
if id1 != id2 {
t.Errorf("Same inputs must produce same ID: %s != %s", id1, id2)
}
}
// TestComputeNativeID_Format verifies the exact ID format, including
// hex encoding, colons, lowercase, and decimal op.
func TestComputeNativeID_Format(t *testing.T) {
// Leading zeros in the byte slices must be preserved in the hex representation.
lsn := []byte{0x00, 0x00, 0x00, 0x10}
seqval := []byte{0x00, 0x00, 0x0b}
op := 2
got := ComputeNativeID(lsn, seqval, op)
want := "00000010:00000b:2"
if got != want {
t.Fatalf("unexpected native ID format: got %q, want %q", got, want)
}
}
// TestComputeNativeID_EmptyLSNOrSeqVal documents and verifies behavior
// when LSN and/or SeqVal are empty or nil.
func TestComputeNativeID_EmptyLSNOrSeqVal(t *testing.T) {
tests := []struct {
name string
lsn []byte
seqval []byte
op int
want string
}{
{
name: "nil LSN and nil SeqVal",
lsn: nil,
seqval: nil,
op: 2,
// Empty hex components produce empty strings around the colons.
want: "::2",
},
{
name: "empty LSN and non-empty SeqVal",
lsn: []byte{},
seqval: []byte{0x01},
op: 3,
want: ":01:3",
},
{
name: "non-empty LSN and empty SeqVal",
lsn: []byte{0x0a},
seqval: []byte{},
op: 4,
want: "0a::4",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := ComputeNativeID(tt.lsn, tt.seqval, tt.op)
if got != tt.want {
t.Fatalf("ComputeNativeID(%v, %v, %d) = %q, want %q",
tt.lsn, tt.seqval, tt.op, got, tt.want)
}
})
}
}These tests assume the current ComputeNativeID implementation:
- Produces lowercase hex without a
0xprefix. - Preserves leading zeros via padding to fixed widths (e.g.,
"00000010"and"00000b"). - Renders empty/nil slices as empty hex components, resulting in strings like
"::2",":01:3", and"0a::4".
If the actual behavior differs (for example, different padding width or non-empty representation for nil/empty), update the want strings in TestComputeNativeID_Format and TestComputeNativeID_EmptyLSNOrSeqVal to match the real, intended contract of ComputeNativeID.
| @@ -203,13 +205,13 @@ func TestStore_GetChangesWithFilter(t *testing.T) { | |||
| tx1 := &core.Transaction{ | |||
| ID: "tx-001", | |||
| Changes: []core.Change{ | |||
There was a problem hiding this comment.
suggestion (testing): Add a complementary test that verifies duplicate native IDs are de-duplicated and logged.
Given the new native ID semantics and INSERT OR IGNORE, there’s also a key negative case to cover:
- Two changes with the same
ID(same LSN+SeqVal+Op) should produce only one row and emit aslog.Warnfor the ignored duplicate.
Please add a test (e.g. TestStore_Write_DuplicateID_IsIgnoredAndLogged) that:
- Writes a batch with two
core.Changeentries sharing the sameIDbut different payloads. - Asserts that only one row is present in
changes. - Verifies via a test logger or hook that a warning is logged when the duplicate is ignored.
![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)
internal/store/sqlite/store.go (1)
107-153:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winMissing
slog.Warnlogging for duplicate skips as stated in PR objectives.The PR summary states: "Added
slog.Warnlogging whenINSERT OR IGNOREskips duplicate rows." However, at lines 150-152, the code checksRowsAffected()and silently incrementsrowsInsertedonly whenn > 0, but does not log anything whenn == 0(indicating a duplicate was skipped byINSERT OR IGNORE).This logging is important for observability and was explicitly listed as a change in the PR objectives.
Add logging for skipped duplicates
if n, _ := res.RowsAffected(); n > 0 { rowsInserted++ + } else { + slog.Warn("INSERT OR IGNORE skipped duplicate CDC change", + "change_id", id, + "table", change.Table, + "lsn", lsnStr) }As per coding guidelines, use
slog.Warnfor warning information such as recoverable errors and abnormal states. A silently skipped duplicate row (especially if unexpected) represents an abnormal state worth logging.🤖 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 `@internal/store/sqlite/store.go` around lines 107 - 153, The loop inserting CDC changes currently ignores when sqlTx.Exec with insertSQL results in 0 rows affected (duplicate skipped); update the branch after checking res.RowsAffected() in the insertion loop (around the use of insertSQL, sqlTx.Exec and rowsInserted) to call slog.Warn with a clear message like "skipped duplicate change on INSERT OR IGNORE" and include context fields (id, table, operation via change.Operation.String(), lsnStr, and maybe change.TransactionID) so skipped duplicates are observable; ensure slog is imported/available and keep the existing rowsInserted logic (increment only when n>0).
🧹 Nitpick comments (4)
internal/cdc/capturer.go (1)
350-392: ⚡ Quick winPopulate
SeqValfield when converting tocore.Change.The function extracts most fields from the
CaptureChangemap (table, txID, lsn, operation, data, commitTime, id, tableKeys), but does not extract theseqvalfield that was added to the map at line 156. This meanscore.Change.SeqValremains nil/empty when returned from this conversion.While this doesn't break functionality (since the ID is already computed), it creates inconsistency: the
SeqValfield exists incore.Changebut is never populated, making the data structure incomplete.Populate SeqVal field for consistency
for _, cc := range captureChanges { table, _ := cc["table"].(string) txID, _ := cc["transaction_id"].(string) lsn, _ := cc["lsn"].([]byte) + seqval, _ := cc["seqval"].([]byte) opVal, _ := cc["operation"].(int) data, _ := cc["data"].(map[string]interface{}) commitTime, _ := cc["commit_time"].(time.Time) id, _ := cc["id"].(string) // Get primary key values from schema cache... changes = append(changes, core.Change{ Table: table, TransactionID: txID, LSN: lsn, + SeqVal: seqval, Operation: core.Operation(opVal), Data: data, CommitTime: commitTime, ID: id, TableKeys: tableKeysStr, }) }🤖 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 `@internal/cdc/capturer.go` around lines 350 - 392, The convertToCoreChanges function currently omits extracting the "seqval" entry from each CaptureChange map, so core.Change.SeqVal stays empty; update convertToCoreChanges to read seqValRaw := cc["seqval"] (or cc["seq_val"] if that key is used), assert/convert it to the expected type (e.g., []byte or string depending on core.Change.SeqVal type) with a safe type assertion and fallback, and set SeqVal: seqVal in the core.Change literal alongside Table, TransactionID, LSN, Operation, Data, CommitTime, ID, and TableKeys so SeqVal is populated consistently for each change.internal/cdc/query.go (1)
214-224: 💤 Low valueConsider logging when
__$seqvalis unexpectedly empty.The extraction correctly follows the pattern used for other CDC metadata. However, since
__$seqvalis part of SQL Server CDC's native unique key, an unexpectedly empty value could lead to non-unique IDs downstream. Consider adding defensive logging to detect this scenario during operation.Optional: Add defensive logging
// Extract seqval (sequence value within transaction - native CDC unique key part 2) var seqval []byte if idx, ok := colIndex["__$seqval"]; ok { if s, ok := dest[idx].(scannerpkg.DBType); ok { if val, err := s.Value(); err == nil && val != nil { if b, ok := val.([]byte); ok { seqval = b } } } } + if len(seqval) == 0 { + slog.Debug("CDC row has empty __$seqval", "lsn", hex.EncodeToString(lsn), "table", tableName) + }🤖 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 `@internal/cdc/query.go` around lines 214 - 224, After extracting seqval in the block that checks colIndex["__$seqval"] and reads dest[idx] via scannerpkg.DBType.Value(), add a defensive check after the extraction: if seqval is nil or len(seqval) == 0 then log a warning including the column presence (colIndex entry), the destination index (idx), and any available row identifiers (e.g., primary key columns from dest or other CDC metadata) so operators can trace which row produced an empty __$seqval; use the package's existing logger (e.g., processLogger or the file's logger) and include context strings like "__$seqval empty" and the relevant identifiers to aid debugging.tests/integration/integration_test.go (1)
207-223: 💤 Low valueClarify the purpose of the synthetic
seqvalfallback in test.The test generates a synthetic
seqval(lines 209-213) when the CDC change has an emptySeqVal. While this is reasonable for testing with mock data, the comment "fallback to a deterministic default if empty" could be more explicit about why this is needed.In production CDC data, an empty
SeqValwould be problematic (see comment onComputeNativeIDin capturer.go). The test's fallback masks this scenario, which is fine for integration testing but worth documenting clearly.Improve comment clarity
// Generate ID from LSN+SeqVal+Op (native CDC unique key) var seqval []byte if len(c.SeqVal) > 0 { seqval = c.SeqVal } else { - seqval = []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, byte(i + 1)} + // Mock CDC data may lack SeqVal; use index-based synthetic value for testing + seqval = []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, byte(i + 1)} }🤖 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/integration_test.go` around lines 207 - 223, The synthetic seqval fallback used when c.SeqVal is empty is masking a production-only issue; update the inline comment above the seqval generation to clearly state that this is a deterministic test-only fallback (used to produce stable change IDs for mock CDC records), mention that in real CDC streams an empty SeqVal would be invalid/problematic (reference ComputeNativeID in capturer.go), and note that this behavior is limited to the test path that builds changeID and populates coreChanges (see seqval, changeID, and the coreChanges[i] assignment) so reviewers understand it is not changing production logic.internal/core/transaction.go (1)
86-120: ⚡ Quick winRemove unused
ComputeChangeIDfunction.This function is no longer used since the PR switched to
ComputeNativeIDininternal/cdc/capturer.go. The content-based hash approach has been completely replaced by the native CDC tuple ID. A search of the codebase confirms no calls to this function exist.🤖 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 `@internal/core/transaction.go` around lines 86 - 120, Remove the unused ComputeChangeID function from internal/core/transaction.go: delete the entire ComputeChangeID implementation and any top-level comments referencing it, then prune any now-unused imports (hex, sha256, json, sort, etc.) from that file; confirm no remaining references to ComputeChangeID exist (the code now uses ComputeNativeID in internal/cdc/capturer.go) and run tests/build to ensure no missing symbols remain.
🤖 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 `@internal/cdc/capturer.go`:
- Around line 416-419: ComputeNativeID currently builds "LSN:SeqVal:Op" without
validating seqval, allowing empty seqval and duplicate IDs; change
ComputeNativeID(lsn, seqval []byte, op int) to return (string, error), validate
that seqval (and lsn) are non-nil/non-empty and return a descriptive error if
missing, and update all call sites that use ComputeNativeID to handle the error
(log/propagate) instead of assuming a valid ID; ensure the error message clearly
names ComputeNativeID and the missing component so callers can react
appropriately.
---
Outside diff comments:
In `@internal/store/sqlite/store.go`:
- Around line 107-153: The loop inserting CDC changes currently ignores when
sqlTx.Exec with insertSQL results in 0 rows affected (duplicate skipped); update
the branch after checking res.RowsAffected() in the insertion loop (around the
use of insertSQL, sqlTx.Exec and rowsInserted) to call slog.Warn with a clear
message like "skipped duplicate change on INSERT OR IGNORE" and include context
fields (id, table, operation via change.Operation.String(), lsnStr, and maybe
change.TransactionID) so skipped duplicates are observable; ensure slog is
imported/available and keep the existing rowsInserted logic (increment only when
n>0).
---
Nitpick comments:
In `@internal/cdc/capturer.go`:
- Around line 350-392: The convertToCoreChanges function currently omits
extracting the "seqval" entry from each CaptureChange map, so core.Change.SeqVal
stays empty; update convertToCoreChanges to read seqValRaw := cc["seqval"] (or
cc["seq_val"] if that key is used), assert/convert it to the expected type
(e.g., []byte or string depending on core.Change.SeqVal type) with a safe type
assertion and fallback, and set SeqVal: seqVal in the core.Change literal
alongside Table, TransactionID, LSN, Operation, Data, CommitTime, ID, and
TableKeys so SeqVal is populated consistently for each change.
In `@internal/cdc/query.go`:
- Around line 214-224: After extracting seqval in the block that checks
colIndex["__$seqval"] and reads dest[idx] via scannerpkg.DBType.Value(), add a
defensive check after the extraction: if seqval is nil or len(seqval) == 0 then
log a warning including the column presence (colIndex entry), the destination
index (idx), and any available row identifiers (e.g., primary key columns from
dest or other CDC metadata) so operators can trace which row produced an empty
__$seqval; use the package's existing logger (e.g., processLogger or the file's
logger) and include context strings like "__$seqval empty" and the relevant
identifiers to aid debugging.
In `@internal/core/transaction.go`:
- Around line 86-120: Remove the unused ComputeChangeID function from
internal/core/transaction.go: delete the entire ComputeChangeID implementation
and any top-level comments referencing it, then prune any now-unused imports
(hex, sha256, json, sort, etc.) from that file; confirm no remaining references
to ComputeChangeID exist (the code now uses ComputeNativeID in
internal/cdc/capturer.go) and run tests/build to ensure no missing symbols
remain.
In `@tests/integration/integration_test.go`:
- Around line 207-223: The synthetic seqval fallback used when c.SeqVal is empty
is masking a production-only issue; update the inline comment above the seqval
generation to clearly state that this is a deterministic test-only fallback
(used to produce stable change IDs for mock CDC records), mention that in real
CDC streams an empty SeqVal would be invalid/problematic (reference
ComputeNativeID in capturer.go), and note that this behavior is limited to the
test path that builds changeID and populates coreChanges (see seqval, changeID,
and the coreChanges[i] assignment) so reviewers understand it is not changing
production logic.
🪄 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: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 3cb6fa82-dd57-4570-9c5c-7e09e6fc79fb
📒 Files selected for processing (7)
internal/cdc/capturer.gointernal/cdc/capturer_test.gointernal/cdc/query.gointernal/core/transaction.gointernal/store/sqlite/store.gointernal/store/sqlite/store_test.gotests/integration/integration_test.go
| // ComputeNativeID computes a native CDC row ID from LSN tuple. | ||
| func ComputeNativeID(lsn, seqval []byte, op int) string { | ||
| return hex.EncodeToString(lsn) + ":" + hex.EncodeToString(seqval) + ":" + strconv.Itoa(op) | ||
| } |
There was a problem hiding this comment.
Validate that seqval is non-empty to prevent ambiguous IDs.
ComputeNativeID does not validate its inputs. If seqval is nil or empty (which could occur if __$seqval is missing from CDC results), the function produces IDs like "0000000100000001::2" with an empty middle component. Multiple changes with empty seqval but the same LSN and operation would then have identical IDs, breaking uniqueness.
According to the PR objectives, the native ID format LSN:SeqVal:Op is chosen specifically because it's "database-guaranteed" unique. An empty seqval defeats this guarantee.
Add validation for required components
// ComputeNativeID computes a native CDC row ID from LSN tuple.
func ComputeNativeID(lsn, seqval []byte, op int) string {
+ if len(lsn) == 0 || len(seqval) == 0 {
+ slog.Warn("ComputeNativeID called with empty LSN or SeqVal",
+ "lsn_len", len(lsn), "seqval_len", len(seqval), "op", op)
+ }
return hex.EncodeToString(lsn) + ":" + hex.EncodeToString(seqval) + ":" + strconv.Itoa(op)
}As per coding guidelines, errors should be both logged and handled appropriately. Consider whether an empty seqval should return an error instead of just logging a warning, depending on whether this scenario is expected in your CDC setup.
📝 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.
| // ComputeNativeID computes a native CDC row ID from LSN tuple. | |
| func ComputeNativeID(lsn, seqval []byte, op int) string { | |
| return hex.EncodeToString(lsn) + ":" + hex.EncodeToString(seqval) + ":" + strconv.Itoa(op) | |
| } | |
| // ComputeNativeID computes a native CDC row ID from LSN tuple. | |
| func ComputeNativeID(lsn, seqval []byte, op int) string { | |
| if len(lsn) == 0 || len(seqval) == 0 { | |
| slog.Warn("ComputeNativeID called with empty LSN or SeqVal", | |
| "lsn_len", len(lsn), "seqval_len", len(seqval), "op", op) | |
| } | |
| return hex.EncodeToString(lsn) + ":" + hex.EncodeToString(seqval) + ":" + strconv.Itoa(op) | |
| } |
🤖 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 `@internal/cdc/capturer.go` around lines 416 - 419, ComputeNativeID currently
builds "LSN:SeqVal:Op" without validating seqval, allowing empty seqval and
duplicate IDs; change ComputeNativeID(lsn, seqval []byte, op int) to return
(string, error), validate that seqval (and lsn) are non-nil/non-empty and return
a descriptive error if missing, and update all call sites that use
ComputeNativeID to handle the error (log/propagate) instead of assuming a valid
ID; ensure the error message clearly names ComputeNativeID and the missing
component so callers can react appropriately.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
1 issue found across 7 files
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="internal/cdc/capturer.go">
<violation number="1" location="internal/cdc/capturer.go:418">
P1: `ComputeNativeID` does not validate that `lsn` and `seqval` are non-empty. If `seqval` is nil (e.g., `__$seqval` missing from CDC results or a driver error in extraction), the generated ID will have an empty middle component like `"<lsn>::<op>"`. Multiple changes with the same LSN and operation but missing seqval will produce identical IDs, silently breaking the uniqueness guarantee this function exists to provide. Return an error or empty string when either component is zero-length so the store layer can reject it.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
| return fmt.Sprintf("%016x", hash) | ||
| // ComputeNativeID computes a native CDC row ID from LSN tuple. | ||
| func ComputeNativeID(lsn, seqval []byte, op int) string { | ||
| return hex.EncodeToString(lsn) + ":" + hex.EncodeToString(seqval) + ":" + strconv.Itoa(op) |
There was a problem hiding this comment.
P1: ComputeNativeID does not validate that lsn and seqval are non-empty. If seqval is nil (e.g., __$seqval missing from CDC results or a driver error in extraction), the generated ID will have an empty middle component like "<lsn>::<op>". Multiple changes with the same LSN and operation but missing seqval will produce identical IDs, silently breaking the uniqueness guarantee this function exists to provide. Return an error or empty string when either component is zero-length so the store layer can reject it.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At internal/cdc/capturer.go, line 418:
<comment>`ComputeNativeID` does not validate that `lsn` and `seqval` are non-empty. If `seqval` is nil (e.g., `__$seqval` missing from CDC results or a driver error in extraction), the generated ID will have an empty middle component like `"<lsn>::<op>"`. Multiple changes with the same LSN and operation but missing seqval will produce identical IDs, silently breaking the uniqueness guarantee this function exists to provide. Return an error or empty string when either component is zero-length so the store layer can reject it.</comment>
<file context>
@@ -410,62 +413,7 @@ func (l LSN) Compare(other []byte) int {
- return fmt.Sprintf("%016x", hash)
+// ComputeNativeID computes a native CDC row ID from LSN tuple.
+func ComputeNativeID(lsn, seqval []byte, op int) string {
+ return hex.EncodeToString(lsn) + ":" + hex.EncodeToString(seqval) + ":" + strconv.Itoa(op)
}
</file context>
2 participants
Fixes: #248
What Changed
Switched Change.ID computation from FNV-1a hash to raw concatenation of
__$start_lsn + __$seqval + __$operation.SeqVal []bytefield toChangestruct__$seqvalfrom resultsslog.Warnlogging whenINSERT OR IGNOREskips duplicate rowsWhy It Changed
Dashboard showed 4-row discrepancy between inserted (3367) and fetched (3371) changes. Root cause: code only used
__$start_lsnbut not__$seqval, which together with__$operationforms CDC's true unique key. Hash-based ID computation is probabilistic—hash collisions, while low probability (3×10⁻¹³ for 3374 rows), are non-zero and unacceptable at scale.Using SQL Server CDC's native row-level unique identifier eliminates collision risk entirely by design.
How to Test
__$start_lsn + __$seqval + __$operationformat forChange.IDchange.ID == ''returns error (no fallback path exists)slog.Warnis logged for skipped rowschangestable to confirm no duplicateChange.IDvalues exist