Michael created FLINK-40242:
-------------------------------
Summary: Support filtered pagination splits using a supporting
index for efficient queries
Key: FLINK-40242
URL: https://issues.apache.org/jira/browse/FLINK-40242
Project: Flink
Issue Type: Improvement
Components: Connectors / MongoDB
Reporter: Michael
h2. Background
The {{PAGINATION}} partition strategy splits a MongoDB collection into
evenly-sized chunks by walking the {{{}{}}}{{{}"{}}}{{{}_id"{}}} index and
skipping a fixed number of records per split. Today this boundary computation
is always keyed on {{"}}{{{}_id{}}}{{{}"{}}}{{{}{}}} and ignores any
user-supplied filter.
As a result, when a source is configured with a filter, pagination splitting
does not take the filter into account when computing boundaries. The per-split
boundary queries are not aligned with the filtered result set, which can
produce uneven splits and force MongoDB into a collection scan rather than an
index scan — expensive on large collections.
h2. Proposed Change
Make the {{PAGINATION}} strategy filter-aware so that split boundaries are
computed via an index-backed query when a filter is present:
* When a filter is supplied, select the best _supporting index_ whose leading
equality fields match the filter and are followed by a trailing {{_id}} field
(e.g. filter {{{a: X, b: Y}}} requires an index {{{}{a: 1, b: 1, _id: 1}{}}}).
When no filter is supplied, behavior is unchanged and the {{_id}} index is used.
* Compute split boundaries by walking the chosen index with
{{{}hint{}}}/{{{}min{}}}/{{{}max{}}} bounds, so MongoDB serves the boundary
queries as an index scan instead of a collection scan. Only equality predicates
(including those nested within {{{}$and{}}}) qualify for the index bounds.
* Splits remain evenly sized by record count while restricting to the filtered
range.
h2. API Additions
* {{MongoReadOptions.MongoReadOptionsBuilder#setFilter(Bson)}} — sets a filter
used by the {{PAGINATION}} strategy. Requires a supporting index as described
above.
* {{MongoSourceBuilder#setFilter(Bson)}} now propagates the filter into read
options.
* {{MongoSourceBuilder#setPartitionRecordSize(int)}} is exposed on the builder
for configuring records-per-partition directly.
h2. Documentation
* Updates the DataStream connector docs (English and Chinese) to document
{{{}setFilter{}}}/{{{}setPartitionRecordSize{}}}, the supporting-index
requirement, and the ascending-{{{}_id{}}} limitation.
* Adds a tip on including the shard key as a trailing index field for
efficient shard targeting on sharded collections.
h2. Notes / Limitations
* The {{PAGINATION}} strategy requires the {{_id}} index to be in ascending
({{{}_id: 1{}}}) order; descending {{_id}} order is not supported.
* When a filter is supplied, a supporting index (equality fields followed by
{{{}_id{}}}) must exist; otherwise the splitter falls back to the {{_id}} index
and the query will not be index-optimized for the filter.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
