mongo: partition snapshots for numeric and string _id collections (#4487)

## Summary

Extends the MongoDB connector so initial snapshots **parallelize for
numeric and string `_id` collections**, not just ObjectID. Previously
any non-ObjectID `_id` fell back to a single serial full-table
partition, making large naturally-keyed collections (app-store numeric
IDs, package-name strings) too slow to snapshot inside the oplog
retention window.

Based on `v0.36.29` (our deployed release).
It improves initial sync (MongoDb -> Clickhouse from 10 hours to 3 hours
on 8mln collection).


## Approach

`GetQRepPartitions` dispatches on the collection's min/max `_id` BSON
type:

- **ObjectID** — unchanged (uniform division of the 12-byte keyspace).
- **Int32/Int64** — uniform min/max division, reusing
`utils.PartitionHelper` to emit non-overlapping `IntPartitionRange`
partitions (`$gte`/`$lte`).
- **String** — sampling-based quantile boundaries via `$sample` (honours
read preference), anchored by the real min/max, emitted as a new
`StringPartitionRange` proto type with contiguous half-open ranges
(`$gte`/`$lt`).
- **Mixed-type / Double / empty** — safe full-table fallback.

## Changes

- `protos/flow.proto` — add `StringPartitionRange` + `string_range`
oneof field.
- `connectors/mongo/qrep_partition.go` — type dispatcher; numeric +
string builders; pure, unit-tested `computeStringBoundaries`.
- `connectors/mongo/qrep.go` — `IntRange`/`StringRange` cases in
`toRangeFilter`.
- `connectors/utils/monitoring/monitoring.go` — handle `StringRange` in
`addPartitionToQRepRun` (avoids the `unknown range type` error path).

## Testing

- `go build ./...`, `go vet` clean.
- New unit tests for `computeStringBoundaries`
(even/clustered/sparse/duplicate/out-of-range/empty + contiguity),
`toRangeFilter` (Int/String/ObjectID/unsupported), and `_id` type
detection — all passing.

## Notes / limitations

- Numeric uses uniform value division; monitor partition balance on
first run for sparse ID spaces.
- Double `_id` and mixed-type collections fall back to full-table by
design.
- Requires regenerating protobuf bindings (`./generate-protos.sh`)
before building images, as `flow/generated` is gitignored.

---------

Co-authored-by: Ilia Demianenko <ilia.demianenko@clickhouse.com>

Author: andreyzhelnin-st

flow/connectors/mongo/qrep.go

@@ -26,17 +26,9 @@ func (c *MongoConnector) GetQRepPartitions(
 	config *protos.QRepConfig,
 	last *protos.QRepPartition,
 ) ([]*protos.QRepPartition, error) {
-	fullTablePartition := []*protos.QRepPartition{
-		{
-			PartitionId:        utils.FullTablePartitionID,
-			Range:              nil,
-			FullTablePartition: true,
-		},
-	}
-
 	if config.WatermarkColumn != DefaultDocumentKeyColumnName {
 		c.logger.Warn("unexpected watermark column, falling back to full table partition")
-		return fullTablePartition, nil
+		return utils.FullTablePartition(), nil
 	}
 
 	if config.NumRowsPerPartition <= 0 {
@@ -70,10 +62,10 @@ func (c *MongoConnector) GetQRepPartitions(
 
 	if adjustedPartitions.AdjustedNumPartitions <= 1 {
 		c.logger.Info("[mongo] insufficient partitions for parallel snapshot, falling back to full table partition")
-		return fullTablePartition, nil
+		return utils.FullTablePartition(), nil
 	}
 
-	return c.minMaxPartitions(ctx, collection, adjustedPartitions.AdjustedNumPartitions)
+	return c.buildPartitions(ctx, collection, adjustedPartitions.AdjustedNumPartitions)
 }
 
 func (c *MongoConnector) GetDefaultPartitionKeyForTables(
@@ -234,6 +226,29 @@ func toRangeFilter(watermarkColumn string, partitionRange *protos.PartitionRange
 				bson.E{Key: "$lte", Value: endObjectID},
 			}},
 		}, nil
+	case *protos.PartitionRange_IntRange:
+		// Numeric _id: inclusive range. PartitionHelper builds non-overlapping
+		// [start, end] integer ranges. Mongo compares int32/int64 numerically, so
+		// int64 bounds match int32-typed _ids.
+		return bson.D{
+			bson.E{Key: watermarkColumn, Value: bson.D{
+				bson.E{Key: "$gte", Value: r.IntRange.Start},
+				bson.E{Key: "$lte", Value: r.IntRange.End},
+			}},
+		}, nil
+	case *protos.PartitionRange_StringRange:
+		// String _id: half-open [start, end) for all but the last partition;
+		// the last partition is closed [start, end] (EndInclusive == true).
+		endOp := "$lt"
+		if r.StringRange.EndInclusive {
+			endOp = "$lte"
+		}
+		return bson.D{
+			bson.E{Key: watermarkColumn, Value: bson.D{
+				bson.E{Key: "$gte", Value: r.StringRange.Start},
+				bson.E{Key: endOp, Value: r.StringRange.End},
+			}},
+		}, nil
 	default:
 		return nil, fmt.Errorf("unsupported partition range type")
 	}