Add Top-N pushdown support for the MongoDB connector #26504
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add Top-N (ORDER BY + LIMIT) pushdown support for the MongoDB connector. MongoDB considers null values to be less than any other value. We can handle sort items with SortOrder.ASC_NULLS_LAST or SortOrder.DESC_NULLS_FIRST with a MongoDB aggregation pipeline, where we add computed fields to sort correctly. We add sort fields to the projection, so that they are considered by MongoDB. We add Top-N pushdown support only for simple nested queries (that work on the same collection).
codluca
force-pushed
the
26366
branch
2 times, most recently
from
8973aee to
0331d72
Compare
|
The pull request is ready for review. Thanks! |
| OptionalInt.of(toIntExact(limit))), | ||
| true, | ||
| false)); | ||
| } | ||
|
|
||
| @Override | ||
| public Optional<TopNApplicationResult<ConnectorTableHandle>> applyTopN(ConnectorSession session, ConnectorTableHandle table, long topNCount, List<SortItem> sortItems, Map<String, ColumnHandle> assignments) |
There was a problem hiding this comment.
Please put each variable on a new line
There was a problem hiding this comment.
Thanks for the review!
Fixed.
| sortDocument.append(columnName, direction); | ||
| } | ||
| List<MongoTableSort> tableSortList = handle.sort().orElse(new ArrayList<>()); | ||
| MongoTableSort tableSort = new MongoTableSort(sortDocument, sortNullFieldsDocument == null ? Optional.empty() : Optional.of(sortNullFieldsDocument), toIntExact(topNCount)); |
There was a problem hiding this comment.
We could use Optional.ofNullable
| if (sortItem.getSortOrder() == SortOrder.ASC_NULLS_LAST || sortItem.getSortOrder() == SortOrder.DESC_NULLS_FIRST) { | ||
| String sortColumnName = "_sortNulls_" + columnName; | ||
| Document condition = new Document(); | ||
| condition.append("$cond", List.of(new Document("$eq", new ArrayList<>(Arrays.asList("$" + columnName, null))), 1, 0)); |
There was a problem hiding this comment.
Suggested change
| condition.append("$cond", List.of(new Document("$eq", new ArrayList<>(Arrays.asList("$" + columnName, null))), 1, 0)); | |
| condition.append("$cond", ImmutableList.of(new Document("$eq", Arrays.asList("$" + columnName, null)), 1, 0)); |
| Document sortNullFieldsDocument = null; | ||
| for (SortItem sortItem : sortItems) { | ||
| String columnName = sortItem.getName(); | ||
| int direction = (sortItem.getSortOrder() == SortOrder.ASC_NULLS_FIRST || sortItem.getSortOrder() == SortOrder.ASC_NULLS_LAST) ? 1 : -1; |
There was a problem hiding this comment.
can we give 1 and -1 a meaningful name?
| .toList(); | ||
| builder.append(", orderBy = ").append(sortEntries); | ||
| if (sortNullFields.isPresent()) { | ||
| List<String> nullFields = sortNullFields.get().keySet().stream() |
There was a problem hiding this comment.
Use ImmutableList.copyOf
| for (String sortField : tableSort.sort().keySet()) { | ||
| // Sorting on the field does not work unless we add it to the projection | ||
| if (!projection.containsKey(sortField)) { | ||
| projection.append(sortField, 1); |
There was a problem hiding this comment.
Same here, it's not obvious for the meaning of 1
Praveen2112
reviewed
Sep 6, 2025
| public Optional<TopNApplicationResult<ConnectorTableHandle>> applyTopN( | ||
| ConnectorSession session, | ||
| ConnectorTableHandle table, | ||
| long topNCount, |
There was a problem hiding this comment.
What if the topNCount is more than range of integer ?
There was a problem hiding this comment.
Added check similar to applyLimit method
// MongoDB doesn't support topN number greater than integer max
if (topNCount > Integer.MAX_VALUE) {
return Optional.empty();
}```
| MongoColumnHandle columnHandle = (MongoColumnHandle) columnHandleEntry.getValue(); | ||
| if (!columnHandleEntry.getKey().equals(columnHandle.baseName())) { | ||
| // We don't support complex nested queries | ||
| return Optional.empty(); |
There was a problem hiding this comment.
How about for dbRef column type ?
There was a problem hiding this comment.
As I understand it, applyTopN is called only for one table, no JOIN. The dbRef from MongoDB would be a reference to another collection, what we achieve with a JOIN. If it's something else, please clarify.
I wrote a simple test, and the execution doesn't enter the applyTopN method.
* Test topN dbRef.
*/
@Test
public void testJoinTopNDbRef() {
Session session = Session.builder(getSession())
.setSystemProperty(IGNORE_STATS_CALCULATOR_FAILURES, "false")
.build();
assertQuery(
session,
"SELECT n.name, c.name " +
"FROM nation n " +
"JOIN customer c ON c.nationkey = n.nationkey " +
"ORDER BY n.nationkey, c.name LIMIT 2");
}```
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Description
Add Top-N (ORDER BY + LIMIT) pushdown support for the MongoDB connector.
Additional context and related issues
MongoDB considers null values to be less than any other value. We can handle sort items with SortOrder.ASC_NULLS_LAST or SortOrder.DESC_NULLS_FIRST with a MongoDB aggregation pipeline, where we add computed fields to sort correctly.
Release notes
( ) This is not user-visible or is docs only, and no release notes are required.
(X ) Release notes are required. Please propose a release note for me.
( ) Release notes are required, with the following suggested text: