feat(dsl): add MongoDB connector

[johntomcat7408-cmyk]

What changes were proposed in this pull request?

• Add a MongoDB source and sink connector with SPI registration.
• Support bounded reads, explicit integer range partitions with [lowerbound, upperbound)
  boundaries, and count offsets.
• Support ordered batch inserts and BSON type conversion.
• Add connector configuration, runtime wiring, unit and integration tests, a GQL example,
  and CN/EN docs.

The source targets static collections. CDC, upsert, automatic retry, and exactly-once delivery
are not included.

Closes #797.

How was this PR tested?

• Tests have been added for the changes

• Production environment verified

• MongoTableConnectorTest and MongoRowConverterTest: 15 tests passed on JDK 8 and JDK 11.

• MongoTableConnectorIT: passed with Docker in
  GitHub Actions;
  0 failures, errors, or skips.

• JDK 8 reactor build through the MongoDB connector passed: 55 modules.

• JDK 11 runtime reactor package passed: 75 modules, including Doris, MongoDB, and runtime.

• Checkstyle and Apache RAT passed.

[johntomcat7408-cmyk]

I verified PR merge commit b42bc097a2b144e0ae9713bb44af174d1cfb5983 (head 148d4791bf2f8199b0703077b26ecba7940d2713) locally on Windows 11 with Oracle JDK 8u451, Maven 3.8.9, and protoc 3.21.7.

• All 107 modules compiled successfully, including test sources (mvn ... -DskipTests).
• MongoDB connector unit tests passed: 12 tests, 0 failures.

The failed JDK 8 job stopped at setup-protoc due to GitHub API rate limiting before Maven started. Could you please rerun it?

[kitalkuyo-gita]

Using skip(offset) for pagination poses potential performance risks.

Issue

The fetch() method utilizes query.skip((int) offset), relying on a count-based offset (where the offset represents the number of documents to skip). Combining skip with sort is highly inefficient when dealing with large collections or high offset values; this leads to full collection scans and memory pressure, with response times degrading linearly as the offset increases.

Risks/Impact

With large datasets, this can result in extremely slow read operations, Out-of-Memory (OOM) errors, timeouts, or task failures. For dynamic collections, it may lead to duplicate records or missed data.

Proposed Solution

Switch to range-based pagination (cursor-based) using the "last seen value." Instead of skip, use the sorting fields (e.g., partitionField + _id) to perform queries such as > lastSeen.
Extend the offset semantics to support two strategies: COUNT_OFFSET and BOOKMARK_OFFSET. While COUNT_OFFSET should be retained as the default for backward compatibility, it is strongly recommended to implement BOOKMARK_OFFSET (storing the _id or partition field value of the last returned record) to enable efficient pagination. ### Example:
New MongoOffset example (saving lastId):

public class MongoOffset implements Offset {
private final Long countOffset;          // Existing semantics (nullable)
private final String lastIdHex;          // New semantics (nullable)

public MongoOffset(long countOffset) { this.countOffset = countOffset; this.lastIdHex = null; }
public MongoOffset(String lastIdHex) { this.lastIdHex = lastIdHex; this.countOffset = null; }

public boolean isBookmark() { return lastIdHex != null; }
public String getLastIdHex() { return lastIdHex; }
@Override public long getOffset() { return countOffset == null ? 0L : countOffset; }
@Override public boolean isTimestamp() { return false; }
@Override public String humanReadable() { return lastIdHex != null ? lastIdHex : String.valueOf(countOffset); }
}

Prioritize bookmark-based pagination in fetch() (example snippet):

// When the offset is a bookmark (lastIdHex), use Filters.gt("_id", new ObjectId(lastIdHex))
Bson filter = mongoPartition.toFilter();
if (startOffset.isPresent() && startOffset.get() instanceof MongoOffset
&& ((MongoOffset) startOffset.get()).isBookmark()) {
String lastIdHex = ((MongoOffset) startOffset.get()).getLastIdHex();
filter = Filters.and(filter, Filters.gt("_id", new ObjectId(lastIdHex)));
} else {
// Fall back to the original skip(offset) behavior (for compatibility)
query = query.skip((int) offset);
}
// Apply limit + sort directly when no skip is used (efficient)
query = query.sort(Sorts.ascending(mongoPartition.hasRange() ? mongoPartition.getField() : "_id"))
.limit(limit);

When returning nextOffset, generate a MongoOffset(lastIdHex) using the _id of the last record; this allows the next fetch operation to use > lastId for pagination (O(1) overhead).

[kitalkuyo-gita]

Using ordered insertMany in flush() results in partial writes without retries (impacting data consistency/availability).

Issue

When using ordered insertMany(true), an error causes the first portion of the batch to remain written while the method immediately throws an IOException. There is no mechanism for retries, rollbacks, or reporting granular details regarding which records were successfully written.

Risks/Impact

Network instability or transient MongoDB backend issues can lead to partial writes and task failures, resulting in upstream duplication or downstream data loss (prone to user misinterpretation).

Proposed Solution

Introduce configurable retries (max attempts, backoff strategy) and implement granular handling of MongoBulkWriteException upon insertion failure (optional: switch to unordered batch writes and retry failed sub-items based on error codes).
Provide configuration options: geaflow.dsl.mongodb.sink.retry.maxAttempts, .retry.backoff.ms, and .sink.ordered (true/false). Example implementation:

private void flushWithRetry() throws IOException {
if (batch.isEmpty()) return;
int attempts = 0;
int maxAttempts = 3; // Read from configuration
long backoffMs = 200L;
List<Document> toWrite = new ArrayList<>(batch);
while (true) {
try {
collection.insertMany(toWrite, new InsertManyOptions().ordered(true));
batch.clear();
return;
} catch (MongoBulkWriteException bwe) {
// If partial writes occurred, bwe.getWriteResult() is available, and bwe.getWriteErrors() returns the failed items.
// Log error info and retry failed items (but be mindful of idempotency and the risk of duplicate writes).
// Simple approach: Log and re-throw an exception with context (safer), or switch to unordered mode and retry failed items.
throw new IOException("Bulk write partially failed: " + bwe.getWriteErrors(), bwe);
} catch (MongoException e) {
attempts++;
if (attempts >= maxAttempts) {
throw new IOException("Failed to write MongoDB after retries", e);
}
try { Thread.sleep(backoffMs); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); }
backoffMs *= 2;
}
}
}

Note: To implement reliable retries, you need to evaluate write idempotency (e.g., potential _id conflicts) and the consequences of unordered bulk writes (which may result in out-of-order execution).

[kitalkuyo-gita]

listPartitions logic differs from expectations (partitionNum is effectively capped by the range size)

Issue

listPartitions uses count = range.min(BigInteger.valueOf(partitionNum)).intValueExact();. When the difference between the upper and lower bounds is less than partitionNum, the number of partitions actually returned is lower than the partitionNum specified by the user (this scenario is covered by tests in the PR). While this behavior is logical, it requires clear documentation and appropriate error or warning messages.

Risk/Impact

Users may specify a certain number of partitions to meet specific requirements; receiving fewer actual partitions for small ranges could lead to confusion regarding performance tuning.

Solution (Proposed)

The documentation should clarify this behavior, or a WARN log entry should be generated during initialization indicating that "requested partition count was reduced to X due to range size." Explicit checks or logging can be added to the code:

int requested = tableConf.getInteger(MongoConfigKeys.GEAFLOW_DSL_MONGODB_PARTITION_NUM);
if (count < requested) {
LOG.warn("Requested partitionNum={} reduced to {} because range size is {}.", requested, count, range);
}

[kitalkuyo-gita]

Partition/Pagination Sorting Requires Indexes (Performance)

Issue

fetch() utilizes sort(mongoPartition.getField(), "_id") or sort("_id"). Without a suitable compound index on the partitionField (or _id), the query results in costly sorting or scanning operations.

Risk/Impact

Queries on large collections may trigger disk-based sorting, out-of-memory errors, slow query performance, or even block the production cluster.

Solution (Recommendation)

Enforce and validate (during initialization or opening) that users create the necessary index on the collection (e.g., {partitionField: 1, _id: 1})—or clearly document this requirement. Implement detection logic (using .listIndexes()) to log a warning or throw an exception, prompting the user to optimize the setup.

Example

// Check for the index after open(); issue a warning or optionally fail
boolean hasIndex = false;
for (Document idx : collection.listIndexes()) {
Document key = (Document) idx.get("key");
if (key.containsKey(partitionField) && key.containsKey("_id")) {
hasIndex = true;
break;
}
}
if (!hasIndex) {
LOG.warn("MongoDB collection {} missing index on ({}, _id). Recommended to create index for efficient reads.", collectionName, partitionField);
}

[kitalkuyo-gita]

@johntomcat7408-cmyk Hello! I’ve been extremely busy lately and have just seen your PR. Thank you very much for your valuable contribution to the community. We take a rigorous and serious approach with every new contributor; I have carefully reviewed your code and identified some performance issues, so I hope you can take a moment to look at them. Best regards.