[Bug]: MongoDbIO read splitting fails for collections with non-ObjectId _id (splitKeysToFilters hardcodes ObjectId) #39900

What happened?

Reading a MongoDB collection whose _id values are not ObjectIds (e.g. application-defined string IDs, which MongoDB fully supports) fails as soon as read splitting is enabled via withBucketAuto(true) (or withNumSplits(n)).

Observed with Beam 2.75.0 on Dataflow, reading a MongoDB Atlas collection with uniform string _id values:

org.apache.beam.sdk.util.UserCodeException: java.lang.IllegalArgumentException: state should be: hexString has 24 characters
    at org.bson.types.ObjectId.parseHexString(ObjectId.java:384)
    at org.bson.types.ObjectId.<init>(ObjectId.java:193)
    at org.bson.json.JsonReader.visitObjectIdConstructor(JsonReader.java:733)
    ...
    at org.bson.Document.parse(Document.java:129)
    at org.apache.beam.sdk.io.mongodb.MongoDbIO$BoundedMongoDbSource.split(MongoDbIO.java:546)
    at org.apache.beam.sdk.io.Read$BoundedSourceAsSDFWrapperFn.splitRestriction(Read.java:304)

Root cause

BoundedMongoDbSource.splitKeysToFilters unconditionally formats split boundaries as ObjectId("%s"), regardless of the boundary value's actual BSON type:

https://github.com/apache/beam/blob/v2.75.0/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java#L603-L643

The $bucketAuto boundary computation itself works fine for string _id values; only the conversion of boundaries into range filters is broken. When the generated filter string is parsed back with Document.parse inside split(), the ObjectId("<non-hex-string>") constructor throws.

The AggregationQuery path has the same assumption (splitKeysToMatch calls splitKeys.get(i).getObjectId("_id")):

https://github.com/apache/beam/blob/v2.75.0/sdks/java/io/mongodb/src/main/java/org/apache/beam/sdk/io/mongodb/MongoDbIO.java#L675

So there is no way to read a non-ObjectId-keyed collection with parallel splitting.

The existing unit test asserts unparseable output

MongoDbIOTest#testSplitIntoFilters feeds an Integer _id (56) and asserts the output contains ObjectId("56") — a filter that can never be parsed back (ObjectId requires a 24-character hex string). The test passes only because it compares strings without parsing them, which indicates the current behavior is accidental rather than intended:

https://github.com/apache/beam/blob/v2.75.0/sdks/java/io/mongodb/src/test/java/org/apache/beam/sdk/io/mongodb/MongoDbIOTest.java#L100-L107

Steps to reproduce

1. Create a collection where _id values are uniform strings (e.g. "id-abc123").
2. Read it with:

MongoDbIO.read()
    .withUri(uri)
    .withDatabase("db")
    .withCollection("coll")
    .withBucketAuto(true)

3. The job fails in split() with the exception above. (withNumSplits(n) without bucketAuto fails the same way once splitVector returns string boundaries.)

Relationship to existing issues

• Mongo document read with non hex objectid #18600 (imported from Jira BEAM-3165, originally reported in 2017) describes the same root cause. Back then splitting ran unconditionally, so the ObjectId(...)-formatted filter blew up later, when parsed in BoundedMongoDbReader.start(). BEAM-3165 Bypass split if numSplit is zero #17084 — attached to the same Jira — made unsplit reads the default (numSplits <= 0 returns a single source), which resolved the default-configuration symptom but left splitKeysToFilters unchanged, so enabling bucketAuto/numSplits still hits it today (as also noted in the 2023 discussion on Mongo document read with non hex objectid #18600). Filing this issue with a current-code reproduction; happy to consolidate into Mongo document read with non hex objectid #18600 if preferred.

• The Python SDK already supports splitting on uniform integer/string _id values ([BEAM-12119] [BEAM-12122] Add integer and string _id keys support to Python IO MongoDB #14460, BEAM-12119/BEAM-12122), so fixing this brings the Java SDK to parity. mongodbio.py explicitly dispatches on the _id type when creating range trackers:

  if isinstance(start_position, ObjectId):
    return _ObjectIdRangeTracker(start_position, stop_position)
  if isinstance(start_position, int):
    return OffsetRangeTracker(start_position, stop_position)
  if isinstance(start_position, str):
    return LexicographicKeyRangeTracker(start_position, stop_position)

• Out of scope: collections with mixed _id types (see [Bug]: Mixed data types in _id field causing pipeline failures (SDK 2.53.0) #30472, closed as not planned). This issue is only about collections with a uniform non-ObjectId _id type.

Proposed fix (backward compatible)

Serialize split boundaries with their actual BSON types using extended JSON (Document#toJson) instead of unconditional ObjectId("...") string formatting. For ObjectId-keyed collections this produces semantically identical range filters, so existing users are unaffected; the public API does not change. testSplitIntoFilters expectations need updating (they currently assert the unparseable output described above).

I'd like to contribute a fix (we already run this change in production) with unit tests for string, ObjectId, and single-split-key cases.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Prism Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner