[
https://issues.apache.org/jira/browse/FLINK-40242?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Michael updated FLINK-40242:
----------------------------
Description:
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, example:
{noformat}
# filter
{a: X, b: Y}
# requires an index
{a: 1, b: 1, _id: 1}
{noformat}
Note: 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.
was:
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.
{noformat}
# filter
{a: X, b: Y}
# requires an index
{a: 1, b: 1, _id: 1}{noformat}
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.
> 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
> Priority: Minor
>
> 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, example:
> {noformat}
> # filter
> {a: X, b: Y}
> # requires an index
> {a: 1, b: 1, _id: 1}
> {noformat}
> Note: 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)
