[
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,
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.
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. 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.
> 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, 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.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
