Some optimizations for mongo connector

[adiom-mark]

Summary by CodeRabbit

• New Features

  • Added skip-batch-overwrite CLI flag for the MongoDB connector, enabling users to optionally skip bulk overwrite operations during data synchronization.

• Performance

  • Optimized partition processing for non-initial-sync operations by reducing unnecessary concurrency overhead.

[coderabbitai]

📝 Walkthrough

Walkthrough

A new SkipBatchOverwrite configuration option is added to the MongoDB connector settings, allowing users to opt-out of bulk overwrite behavior. A corresponding CLI flag enables this feature to be controlled, and the connector logic is updated to conditionally gate bulk write operations based on this setting.

Changes

Cohort / File(s)	Summary
MongoDB Connector Implementation connectors/mongo/conn.go	Added SkipBatchOverwrite field to ConnectorSettings struct; expanded insertBatchOverwrite function signature to accept skipBatchOverwrite parameter; updated bulk overwrite logic to conditionally execute based on flag; modified GeneratePlan to short-circuit partition processing for non-initial-sync runs; threaded parameter through all recursive and call sites.
CLI Configuration internal/app/options/connectorflags.go	Added new boolean CLI flag skip-batch-overwrite bound to Settings.SkipBatchOverwrite field.

Estimated code review effort

🎯 2 (Simple) | ⏱️ ~12 minutes

Poem

🐰 A toggle so neat, a flag so fine,
Skip the bulk overwrites in a single line!
Through function calls we gently thread,
New settings flow where once we spread! ✨

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 inconclusive)

Check name	Status	Explanation	Resolution
Title check	❓ Inconclusive	The title 'Some optimizations for mongo connector' is vague and generic, using non-descriptive language that doesn't convey the specific nature of the changes made.	Use a more specific title that describes the primary change, such as 'Add SkipBatchOverwrite option to mongo connector' or 'Add flag to skip batch overwrite in mongo connector'.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch mongoconnectorfix

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 1

🤖 Fix all issues with AI agents

In `@connectors/mongo/conn.go`:
- Around line 237-243: The goroutine closure in the loop over partitions closes
over the loop variable partition causing races; fix by capturing the current
partition before spawning the goroutine (e.g., assign cur := partition or pass
partition as a parameter) and then use that captured variable inside the eg.Go
closure where ToNS(partition.Namespace) and
c.client.Database(...).Collection(...) are invoked so each goroutine gets the
correct partition value.

🧹 Nitpick comments (1)

internal/app/options/connectorflags.go (1)

757-762: Add a help description for --skip-batch-overwrite.

The flag is missing a Usage field. Without it, the CLI help output shows an empty description, making the option harder to discover. All other flags in this file include Usage descriptions.

💡 Suggested update

 altsrc.NewBoolFlag(&cli.BoolFlag{
 	Name:        "skip-batch-overwrite",
+	Usage:       "Skip bulk overwrite when duplicate keys are encountered",
 	Destination: &settings.SkipBatchOverwrite,
 }),

[coderabbitai]

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

find . -name "conn.go" | grep mongo

Repository: adiom-data/dsync

Length of output: 86

---

🏁 Script executed:

cat -n connectors/mongo/conn.go | sed -n '230,250p'

Repository: adiom-data/dsync

Length of output: 723

---

🏁 Script executed:

cat -n connectors/mongo/conn.go | sed -n '235,260p'

Repository: adiom-data/dsync

Length of output: 1010

---

🏁 Script executed:

rg -n "eg\.Go\(func" connectors/mongo/conn.go

Repository: adiom-data/dsync

Length of output: 114

---

🏁 Script executed:

cat -n connectors/mongo/conn.go | sed -n '1,20p'

Repository: adiom-data/dsync

Length of output: 498

---

🏁 Script executed:

cat -n connectors/mongo/conn.go | sed -n '1060,1085p'

Repository: adiom-data/dsync

Length of output: 1103

---

🏁 Script executed:

cat -n connectors/mongo/conn.go | sed -n '237,270p'

Repository: adiom-data/dsync

Length of output: 1617

---

Capture partition before spawning goroutines.

Line 242 and 254 close over the range variable partition; in Go this variable is reused, so goroutines can observe the last value and race. All concurrent tasks spawned via eg.Go() will reference the same final partition value.

🔧 Safe capture pattern

 for _, partition := range partitions {
 	if !r.Msg.GetInitialSync() {
 		break
 	}
+	partition := partition
 	eg.Go(func() error {
 		ns, _ := ToNS(partition.Namespace)

🤖 Prompt for AI Agents

In `@connectors/mongo/conn.go` around lines 237 - 243, The goroutine closure in
the loop over partitions closes over the loop variable partition causing races;
fix by capturing the current partition before spawning the goroutine (e.g.,
assign cur := partition or pass partition as a parameter) and then use that
captured variable inside the eg.Go closure where ToNS(partition.Namespace) and
c.client.Database(...).Collection(...) are invoked so each goroutine gets the
correct partition value.