Adds optimization for related relations (#4345)

* Adds optimization for related relations

* Makes MongoStorageAdapter only able to sort on Join tables

src/Adapters/Storage/Mongo/MongoStorageAdapter.js

@@ -86,7 +86,7 @@ export class MongoStorageAdapter {
   // Public
   connectionPromise;
   database;
-
+  canSortOnJoinTables;
   constructor({
     uri = defaults.DefaultMongoURI,
     collectionPrefix = '',
@@ -98,6 +98,7 @@ export class MongoStorageAdapter {
 
     // MaxTimeMS is not a global MongoDB client option, it is applied per operation.
     this._maxTimeMS = mongoOptions.maxTimeMS;
+    this.canSortOnJoinTables = true;
     delete mongoOptions.maxTimeMS;
   }
 

src/Controllers/DatabaseController.js

@@ -599,8 +599,16 @@ DatabaseController.prototype.deleteEverything = function() {
 
 // Returns a promise for a list of related ids given an owning id.
 // className here is the owning className.
-DatabaseController.prototype.relatedIds = function(className, key, owningId) {
-  return this.adapter.find(joinTableName(className, key), relationSchema, { owningId }, {})
+DatabaseController.prototype.relatedIds = function(className, key, owningId, queryOptions) {
+  const { skip, limit, sort } = queryOptions;
+  const findOptions = {};
+  if (sort && sort.createdAt && this.adapter.canSortOnJoinTables) {
+    findOptions.sort = { '_id' : sort.createdAt };
+    findOptions.limit = limit;
+    findOptions.skip = skip;
+    queryOptions.skip = 0;
+  }
+  return this.adapter.find(joinTableName(className, key), relationSchema, { owningId }, findOptions)
     .then(results => results.map(result => result.relatedId));
 };
 
@@ -693,11 +701,11 @@ DatabaseController.prototype.reduceInRelation = function(className, query, schem
 
 // Modifies query so that it no longer has $relatedTo
 // Returns a promise that resolves when query is mutated
-DatabaseController.prototype.reduceRelationKeys = function(className, query) {
+DatabaseController.prototype.reduceRelationKeys = function(className, query, queryOptions) {
 
   if (query['$or']) {
     return Promise.all(query['$or'].map((aQuery) => {
-      return this.reduceRelationKeys(className, aQuery);
+      return this.reduceRelationKeys(className, aQuery, queryOptions);
     }));
   }
 
@@ -706,11 +714,12 @@ DatabaseController.prototype.reduceRelationKeys = function(className, query) {
     return this.relatedIds(
       relatedTo.object.className,
       relatedTo.key,
-      relatedTo.object.objectId)
+      relatedTo.object.objectId,
+      queryOptions)
       .then((ids) => {
         delete query['$relatedTo'];
         this.addInObjectIdsIds(ids, query);
-        return this.reduceRelationKeys(className, query);
+        return this.reduceRelationKeys(className, query, queryOptions);
       });
   }
 };
@@ -831,8 +840,9 @@ DatabaseController.prototype.find = function(className, query, {
               throw new Parse.Error(Parse.Error.INVALID_KEY_NAME, `Invalid field name: ${fieldName}.`);
             }
           });
+          const queryOptions = { skip, limit, sort, keys, readPreference };
           return (isMaster ? Promise.resolve() : schemaController.validatePermission(className, aclGroup, op))
-            .then(() => this.reduceRelationKeys(className, query))
+            .then(() => this.reduceRelationKeys(className, query, queryOptions))
             .then(() => this.reduceInRelation(className, query, schemaController))
             .then(() => {
               if (!isMaster) {
@@ -871,7 +881,7 @@ DatabaseController.prototype.find = function(className, query, {
                 if (!classExists) {
                   return [];
                 } else {
-                  return this.adapter.find(className, schema, query, { skip, limit, sort, keys, readPreference })
+                  return this.adapter.find(className, schema, query, queryOptions)
                     .then(objects => objects.map(object => {
                       object = untransformObjectACL(object);
                       return filterSensitiveData(isMaster, aclGroup, className, object)