Improve support for docdb - #374
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (3)
✅ Files skipped from review due to trivial changes (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdds DocumentDB-aware Mongo flavor handling, centralizes change-stream resume/start option selection, and introduces a DocumentDB-capable sampling implementation for partition planning that concurrently collects, deduplicates, validates, and type-sorts sampled Changes
Sequence Diagram(s)mermaid Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 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 |
![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
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@connectors/mongo/conn.go`:
- Around line 1008-1016: The changeStreamOpts function sets resume/start options
unconditionally which can pass an empty cursor; update changeStreamOpts (method
on type conn) to guard the resume logic by checking len(cursor) > 0 before
calling opts.SetResumeAfter(bson.Raw(cursor)) or
opts.SetStartAfter(bson.Raw(cursor)), mirroring the existing empty-cursor
pattern used in DecodeCursor; only call the Set* method when the cursor bytes
are non-empty so optional cursors are respected.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: f592fe42-3aba-470e-8092-606d772e32ad
📒 Files selected for processing (2)
connectors/mongo/conn.gointernal/app/options/connectorflags.go
| func (c *conn) changeStreamOpts(cursor []byte) *moptions.ChangeStreamOptions { | ||
| opts := moptions.ChangeStream() | ||
| if c.flavor == FlavorDocumentDB { | ||
| opts.SetResumeAfter(bson.Raw(cursor)) | ||
| } else { | ||
| opts.SetStartAfter(bson.Raw(cursor)) | ||
| } | ||
| return opts | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Check helper implementation and whether cursor guards exist:"
rg -n -C3 'func \(c \*conn\) changeStreamOpts|SetResumeAfter|SetStartAfter|len\(cursor\)' connectors/mongo/conn.go
echo
echo "Check stream call sites using request cursors:"
rg -n -C2 'StreamLSN|StreamUpdates|GetCursor\(\)|changeStreamOpts\(' connectors/mongo/conn.go
echo
echo "Check API definitions/comments for cursor optionality:"
rg -n -C2 'message StreamLSNRequest|message StreamUpdatesRequest|cursor' --iglob '*.proto' || trueRepository: adiom-data/dsync
Length of output: 5660
Add empty cursor guard to changeStreamOpts for consistency.
The function currently passes cursor to SetResumeAfter/SetStartAfter without checking if it's empty. The codebase already uses this pattern in DecodeCursor (line 379), and proto definitions document cursor as optional ("if present"). Add the guard before setting resume options:
Proposed fix
func (c *conn) changeStreamOpts(cursor []byte) *moptions.ChangeStreamOptions {
opts := moptions.ChangeStream()
+ if len(cursor) == 0 {
+ return opts
+ }
if c.flavor == FlavorDocumentDB {
opts.SetResumeAfter(bson.Raw(cursor))
} else {
opts.SetStartAfter(bson.Raw(cursor))
}
return opts
}📝 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.
| func (c *conn) changeStreamOpts(cursor []byte) *moptions.ChangeStreamOptions { | |
| opts := moptions.ChangeStream() | |
| if c.flavor == FlavorDocumentDB { | |
| opts.SetResumeAfter(bson.Raw(cursor)) | |
| } else { | |
| opts.SetStartAfter(bson.Raw(cursor)) | |
| } | |
| return opts | |
| } | |
| func (c *conn) changeStreamOpts(cursor []byte) *moptions.ChangeStreamOptions { | |
| opts := moptions.ChangeStream() | |
| if len(cursor) == 0 { | |
| return opts | |
| } | |
| if c.flavor == FlavorDocumentDB { | |
| opts.SetResumeAfter(bson.Raw(cursor)) | |
| } else { | |
| opts.SetStartAfter(bson.Raw(cursor)) | |
| } | |
| return opts | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@connectors/mongo/conn.go` around lines 1008 - 1016, The changeStreamOpts
function sets resume/start options unconditionally which can pass an empty
cursor; update changeStreamOpts (method on type conn) to guard the resume logic
by checking len(cursor) > 0 before calling opts.SetResumeAfter(bson.Raw(cursor))
or opts.SetStartAfter(bson.Raw(cursor)), mirroring the existing empty-cursor
pattern used in DecodeCursor; only call the Set* method when the cursor bytes
are non-empty so optional cursors are respected.
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
connectors/mongo/conn.go (1)
1008-1016:⚠️ Potential issue | 🟡 MinorAdd empty cursor guard to
changeStreamOpts.When
cursoris empty (initial stream with no resume point), callingSetResumeAfterorSetStartAfterwith an emptybson.Rawmay cause unexpected behavior. The codebase already guards empty cursors inDecodeCursor(line 379). Add an early return for consistency with the optional cursor semantics documented in proto definitions.Proposed fix
func (c *conn) changeStreamOpts(cursor []byte) *moptions.ChangeStreamOptions { opts := moptions.ChangeStream() + if len(cursor) == 0 { + return opts + } if c.flavor == FlavorDocumentDB { opts.SetResumeAfter(bson.Raw(cursor)) } else { opts.SetStartAfter(bson.Raw(cursor)) } return opts }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connectors/mongo/conn.go` around lines 1008 - 1016, The changeStreamOpts function should guard against an empty cursor byte slice and return default moptions.ChangeStream() immediately; modify conn.changeStreamOpts(cursor []byte) to check if len(cursor)==0 and return opts before calling opts.SetResumeAfter or opts.SetStartAfter so empty cursors are treated as no-resume/start point (consistent with DecodeCursor behavior), leaving the existing FlavorDocumentDB branch and SetResumeAfter/SetStartAfter calls unchanged for non-empty cursors.
🧹 Nitpick comments (2)
connectors/mongo/docdb.go (2)
70-91: Consider limiting concurrency for large sample counts.For large
numSamplesvalues (the code warns at >1,000,000 inconn.go), spawning one goroutine per sample could exhaust resources or overwhelm the database with concurrent requests. Consider using a semaphore or bounded worker pool.Example bounded concurrency
+ const maxConcurrency = 100 results := make([]sampleResult, numSamples) eg, ctx := errgroup.WithContext(ctx) + eg.SetLimit(maxConcurrency) for i := int64(0); i < numSamples; i++ {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connectors/mongo/docdb.go` around lines 70 - 91, The current loop spawns one goroutine per index which can exhaust resources when numSamples is large; modify the logic that builds results (results, sampleResult, numSamples, errgroup, col.Aggregate, eg.Go, eg.Wait) to use bounded concurrency by adding a semaphore or worker-pool: create a configurable maxConcurrency (or derive one), acquire a slot before launching each eg.Go and release it in the goroutine (or instead push indices to a jobs channel and start maxConcurrency worker goroutines that call col.Aggregate and write into results), keep using errgroup to collect errors and call eg.Wait; this limits simultaneous Aggregate calls while preserving the current result-writing and error-handling behavior.
27-45: Type mismatch incompareBSONRawValueswill panic.The function only switches on
a.Typeand assumesbhas the same type. If called with mismatched types (e.g., due to a bug upstream), it will produce incorrect comparisons for the non-ObjectID/Binary cases or panic. The caller validates same-type at lines 108-111, so this is safe in current usage, but a defensive check would prevent future misuse.Optional defensive guard
func compareBSONRawValues(a, b bson.RawValue) int { + if a.Type != b.Type { + panic("compareBSONRawValues called with mismatched types") + } switch a.Type {🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@connectors/mongo/docdb.go` around lines 27 - 45, compareBSONRawValues currently assumes a.Type == b.Type and will misbehave or panic if types differ; add a defensive guard at the start of compareBSONRawValues that checks if a.Type != b.Type and, when they differ, returns a deterministic ordering (for example by returning cmp.Compare(int(a.Type), int(b.Type)) or similar) so callers get a stable result instead of hitting incorrect logic or a panic; keep the existing per-type switch for when types match.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@connectors/mongo/docdb.go`:
- Around line 47-62: In sampleIDs, after the non-DocumentDB aggregation loop
that uses res.Next(ctx) and appends res.Current.Lookup("_id") to ids, check the
cursor error via res.Err() and return that error (wrapped with context) instead
of silently returning partial results; ensure this check happens before
returning ids and after res.Close(ctx) or immediately after the loop so any
iteration failures on the cursor are propagated from sampleIDs.
---
Duplicate comments:
In `@connectors/mongo/conn.go`:
- Around line 1008-1016: The changeStreamOpts function should guard against an
empty cursor byte slice and return default moptions.ChangeStream() immediately;
modify conn.changeStreamOpts(cursor []byte) to check if len(cursor)==0 and
return opts before calling opts.SetResumeAfter or opts.SetStartAfter so empty
cursors are treated as no-resume/start point (consistent with DecodeCursor
behavior), leaving the existing FlavorDocumentDB branch and
SetResumeAfter/SetStartAfter calls unchanged for non-empty cursors.
---
Nitpick comments:
In `@connectors/mongo/docdb.go`:
- Around line 70-91: The current loop spawns one goroutine per index which can
exhaust resources when numSamples is large; modify the logic that builds results
(results, sampleResult, numSamples, errgroup, col.Aggregate, eg.Go, eg.Wait) to
use bounded concurrency by adding a semaphore or worker-pool: create a
configurable maxConcurrency (or derive one), acquire a slot before launching
each eg.Go and release it in the goroutine (or instead push indices to a jobs
channel and start maxConcurrency worker goroutines that call col.Aggregate and
write into results), keep using errgroup to collect errors and call eg.Wait;
this limits simultaneous Aggregate calls while preserving the current
result-writing and error-handling behavior.
- Around line 27-45: compareBSONRawValues currently assumes a.Type == b.Type and
will misbehave or panic if types differ; add a defensive guard at the start of
compareBSONRawValues that checks if a.Type != b.Type and, when they differ,
returns a deterministic ordering (for example by returning
cmp.Compare(int(a.Type), int(b.Type)) or similar) so callers get a stable result
instead of hitting incorrect logic or a panic; keep the existing per-type switch
for when types match.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 885513d1-c094-4899-914c-0c4d7f28cbfd
📒 Files selected for processing (3)
connectors/mongo/conn.goconnectors/mongo/docdb.gointernal/app/options/connectorflags.go
🚧 Files skipped from review as they are similar to previous changes (1)
- internal/app/options/connectorflags.go
1 participant
Summary by CodeRabbit