SERVER-64994 Extend the planner to allow soft hints about index traversal directionr6.0.0-rc10

(cherry picked from commit 00d2a56763b2b0da941a41684d20e7080da5058e)

17 files changed, 640 insertions, 259 deletions

diff --git a/jstests/core/timeseries/bucket_unpacking_with_sort.js b/jstests/core/timeseries/bucket_unpacking_with_sort.js
index b5f145b6e41..3e92a9e2461 100644
--- a/jstests/core/timeseries/bucket_unpacking_with_sort.js
+++ b/jstests/core/timeseries/bucket_unpacking_with_sort.js
@@ -25,6 +25,7 @@
 load("jstests/libs/fixture_helpers.js");             // For FixtureHelpers.
 load("jstests/aggregation/extras/utils.js");         // For getExplainedPipelineFromAggregation.
 load("jstests/core/timeseries/libs/timeseries.js");  // For TimeseriesTest
+load("jstests/libs/analyze_plan.js");                // For getAggPlanStage
 
 if (!TimeseriesTest.bucketUnpackWithSortEnabled(db.getMongo())) {
     jsTestLog("Skipping test because 'BucketUnpackWithSort' is disabled.");
@@ -133,8 +134,30 @@ const hasInternalBoundedSort = (pipeline) =>
 
 const findFirstMatch = (pipeline) => pipeline.find(stage => stage.hasOwnProperty("$match"));
 
+const getWinningPlan = (explain) => {
+    if (explain.hasOwnProperty("shards")) {
+        for (const shardName in explain.shards) {
+            return explain.shards[shardName].stages[0]["$cursor"].queryPlanner.winningPlan;
+        }
+    }
+    return explain.stages[0]["$cursor"].queryPlanner.winningPlan;
+};
+
+const getAccessPathFromWinningPlan = (winningPlan) => {
+    if (winningPlan.stage == "SHARDING_FILTER" || winningPlan.stage === "FETCH") {
+        return getAccessPathFromWinningPlan(winningPlan.inputStage);
+    } else if (winningPlan.stage === "COLLSCAN" || winningPlan.stage === "IXSCAN") {
+        return winningPlan;
+    }
+};
+
+const getAccessPath = (explain) => {
+    return getAccessPathFromWinningPlan(getWinningPlan(explain));
+};
+
 const setup = (coll, createIndex = null) => {
     if (createIndex) {
+        assert.commandWorked(coll.dropIndexes());
         assert.commandWorked(coll.createIndex(createIndex));
     }
 };
@@ -153,6 +176,7 @@ const setup = (coll, createIndex = null) => {
 const runRewritesTest = (sortSpec,
                          createIndex,
                          hint,
+                         expectedAccessPath,
                          testColl,
                          precise,
                          intermediaryStages = [],
@@ -161,6 +185,7 @@ const runRewritesTest = (sortSpec,
         sortSpec,
         createIndex,
         hint,
+        expectedAccessPath,
         testColl,
         precise,
         intermediaryStages,
@@ -224,6 +249,15 @@ const runRewritesTest = (sortSpec,
                    sortDoc(bucketSpanMatch),
                    'Did not expect an extra $match to check the bucket span');
     }
+
+    if (expectedAccessPath) {
+        const paths = getAggPlanStages(optExplainFull, expectedAccessPath.stage);
+        for (const path of paths) {
+            for (const key in expectedAccessPath) {
+                assert.eq(path[key], expectedAccessPath[key]);
+            }
+        }
+    }
 };
 
 const runDoesntRewriteTest = (sortSpec, createIndex, hint, testColl, intermediaryStages = []) => {
@@ -247,46 +281,75 @@ const runDoesntRewriteTest = (sortSpec, createIndex, hint, testColl, intermediar
     assert(!containsOptimization, optExplainFull);
 };
 
+const forwardCollscan = {
+    stage: "COLLSCAN",
+    direction: "forward"
+};
+const backwardCollscan = {
+    stage: "COLLSCAN",
+    direction: "backward"
+};
+// We drop all other indexes during runRewritesTest, so asserting that an IXSCAN is used is enough.
+const forwardIxscan = {
+    stage: "IXSCAN",
+    direction: "forward"
+};
+const backwardIxscan = {
+    stage: "IXSCAN",
+    direction: "backward"
+};
+
 // Collscan cases
-runRewritesTest({t: 1}, null, null, coll, true);
-runRewritesTest({t: -1}, null, {$natural: -1}, coll, false);
+runRewritesTest({t: 1}, null, null, forwardCollscan, coll, true);
+runRewritesTest({t: -1}, null, null, backwardCollscan, coll, false);
 
 // Indexed cases
-runRewritesTest({t: 1}, {t: 1}, {t: 1}, coll, true);
-runRewritesTest({t: -1}, {t: -1}, {t: -1}, coll, true);
-runRewritesTest({t: 1}, {t: 1}, {t: 1}, coll, true);
-runRewritesTest({m: 1, t: -1}, {m: 1, t: -1}, {m: 1, t: -1}, metaColl, true);
-runRewritesTest({m: -1, t: 1}, {m: -1, t: 1}, {m: -1, t: 1}, metaColl, true);
-runRewritesTest({m: -1, t: -1}, {m: -1, t: -1}, {m: -1, t: -1}, metaColl, true);
-runRewritesTest({m: 1, t: 1}, {m: 1, t: 1}, {m: 1, t: 1}, metaColl, true);
+runRewritesTest({t: 1}, {t: 1}, null, null, coll, true);
+runRewritesTest({t: -1}, {t: -1}, {t: -1}, forwardIxscan, coll, true);
+runRewritesTest({t: 1}, {t: 1}, {t: 1}, forwardIxscan, coll, true);
+runRewritesTest({t: 1}, {t: -1}, {t: -1}, backwardIxscan, coll, false);
+runRewritesTest({t: -1}, {t: 1}, {t: 1}, backwardIxscan, coll, false);
+runRewritesTest({m: 1, t: -1}, {m: 1, t: -1}, {m: 1, t: -1}, forwardIxscan, metaColl, true);
+runRewritesTest({m: -1, t: 1}, {m: -1, t: 1}, {m: -1, t: 1}, forwardIxscan, metaColl, true);
+runRewritesTest({m: -1, t: -1}, {m: -1, t: -1}, {m: -1, t: -1}, forwardIxscan, metaColl, true);
+runRewritesTest({m: 1, t: 1}, {m: 1, t: 1}, {m: 1, t: 1}, forwardIxscan, metaColl, true);
 
 // Intermediary projects that don't modify sorted fields are allowed.
-runRewritesTest({m: 1, t: 1}, {m: 1, t: 1}, {m: 1, t: 1}, metaColl, true, [{$project: {a: 0}}]);
 runRewritesTest(
-    {m: 1, t: 1}, {m: 1, t: 1}, {m: 1, t: 1}, metaColl, true, [{$project: {m: 1, t: 1}}]);
-runRewritesTest({t: 1}, {t: 1}, {t: 1}, metaColl, true, [{$project: {m: 0, _id: 0}}]);
-runRewritesTest({'m.b': 1, t: 1}, {'m.b': 1, t: 1}, {'m.b': 1, t: 1}, metaCollSubFields, true, [
-    {$project: {'m.a': 0}}
+    {m: 1, t: 1}, {m: 1, t: 1}, {m: 1, t: 1}, forwardIxscan, metaColl, true, [{$project: {a: 0}}]);
+runRewritesTest({m: 1, t: 1}, {m: 1, t: 1}, {m: 1, t: 1}, forwardIxscan, metaColl, true, [
+    {$project: {m: 1, t: 1}}
 ]);
+runRewritesTest(
+    {t: 1}, {t: 1}, {t: 1}, forwardIxscan, metaColl, true, [{$project: {m: 0, _id: 0}}]);
+runRewritesTest(
+    {'m.b': 1, t: 1}, {'m.b': 1, t: 1}, {'m.b': 1, t: 1}, forwardIxscan, metaCollSubFields, true, [
+        {$project: {'m.a': 0}}
+    ]);
 
 // Test multiple meta fields
 let metaIndexObj = Object.assign({}, ...subFields.map(field => ({[`m.${field}`]: 1})));
 Object.assign(metaIndexObj, {t: 1});
-runRewritesTest(metaIndexObj, metaIndexObj, metaIndexObj, metaCollSubFields, true);
-runRewritesTest(
-    metaIndexObj, metaIndexObj, metaIndexObj, metaCollSubFields, true, [{$project: {m: 1, t: 1}}]);
+runRewritesTest(metaIndexObj, metaIndexObj, metaIndexObj, forwardIxscan, metaCollSubFields, true);
+runRewritesTest(metaIndexObj, metaIndexObj, metaIndexObj, forwardIxscan, metaCollSubFields, true, [
+    {$project: {m: 1, t: 1}}
+]);
 
 // Check sort-limit optimization.
-runRewritesTest({t: 1}, {t: 1}, {t: 1}, coll, true, [], [{$limit: 10}]);
+runRewritesTest({t: 1}, {t: 1}, {t: 1}, null, coll, true, [], [{$limit: 10}]);
 
 // Check set window fields is optimized as well.
 // Since {k: 1} cannot provide a bounded sort we know if there's a bounded sort it comes form
 // setWindowFields.
-runRewritesTest({k: 1}, {m: 1, t: 1}, {m: 1, t: 1}, metaColl, true, [], [
+runRewritesTest({k: 1}, {m: 1, t: 1}, {m: 1, t: 1}, null, metaColl, true, [], [
     {$setWindowFields: {partitionBy: "$m", sortBy: {t: 1}, output: {arr: {$max: "$t"}}}}
 ]);
+// Test that when a collection scan is hinted, we rewrite to bounded sort even if the hint of
+// the direction is opposite to the sort.
+runRewritesTest({t: -1}, null, {$natural: 1}, backwardCollscan, coll, false, [], []);
+runRewritesTest({t: 1}, null, {$natural: -1}, forwardCollscan, coll, true, [], []);
 
-// Negative tests
+// Negative tests and backwards cases
 for (let m = -1; m < 2; m++) {
     for (let t = -1; t < 2; t++) {
         for (let k = -1; k < 2; k++) {
@@ -321,12 +384,7 @@ for (let m = -1; m < 2; m++) {
                 // For the meta case, negate the time order.
                 // For the non-meta case, use a collscan with a negated order.
                 if (m == 0) {
-                    if (t == 0) {
-                        // Do not execute a test run.
-                    } else {
-                        sort = {t: t};
-                        hint = {$natural: -t};
-                    }
+                    // Do not execute a test run.
                 } else {
                     if (t == 0) {
                         // Do not execute a test run.
@@ -338,8 +396,9 @@ for (let m = -1; m < 2; m++) {
                     }
                 }
 
-                if (sort)
+                if (sort) {
                     runDoesntRewriteTest(sort, createIndex, hint, usesMeta ? metaColl : coll);
+                }
 
                 sort = null;
                 hint = null;
@@ -348,13 +407,7 @@ for (let m = -1; m < 2; m++) {
                 // For the meta case, negate the meta order.
                 // For the non-meta case, use an index instead of a collscan.
                 if (m == 0) {
-                    if (t == 0) {
-                        // Do not execute a test run.
-                    } else {
-                        sort = {t: t};
-                        createIndex = {t: -t};
-                        hint = createIndex;
-                    }
+                    // Do not execute a test run.
                 } else {
                     if (t == 0) {
                         // Do not execute a test run.
@@ -366,8 +419,9 @@ for (let m = -1; m < 2; m++) {
                     }
                 }
 
-                if (sort)
+                if (sort) {
                     runDoesntRewriteTest(sort, createIndex, hint, usesMeta ? metaColl : coll);
+                }
 
                 sort = null;
                 hint = null;
@@ -392,7 +446,8 @@ for (let m = -1; m < 2; m++) {
                 }
 
                 if (sort)
-                    runDoesntRewriteTest(sort, createIndex, hint, usesMeta ? metaColl : coll);
+                    runRewritesTest(
+                        sort, createIndex, hint, backwardIxscan, usesMeta ? metaColl : coll);
             }
         }
     }
@@ -442,11 +497,13 @@ for (const sort of [-1, +1]) {
     for (const m of [-1, +1]) {
         for (const t of [-1, +1]) {
             const index = {m, t};
-            // TODO SERVER-64994 will allow reverse scan.
-            if (t === sort)
-                runRewritesTest({t: sort}, index, index, metaColl, true, [{$match: {m: 7}}]);
-            else
-                runDoesntRewriteTest({t: sort}, index, index, metaColl, [{$match: {m: 7}}]);
+            const expectedAccessPath = t === sort ? forwardIxscan : backwardIxscan;
+            runRewritesTest({t: sort}, index, index, expectedAccessPath, metaColl, t === sort, [
+                {$match: {m: 7}}
+            ]);
+            runRewritesTest({t: sort}, index, null, expectedAccessPath, metaColl, t === sort, [
+                {$match: {m: 7}}
+            ]);
         }
     }
 }
@@ -458,13 +515,16 @@ for (const sort of [-1, +1]) {
             for (const t of [-1, +1]) {
                 for (const trailing of [{}, {x: 1, y: -1}]) {
                     const index = Object.merge({'m.a': a, 'm.b': b, t: t}, trailing);
-                    // TODO SERVER-64994 will allow reverse scan.
-                    if (t === sort)
-                        runRewritesTest({t: sort}, index, index, metaCollSubFields, true, [
-                            {$match: {'m.a': 5, 'm.b': 5}}
-                        ]);
-                    else
-                        runDoesntRewriteTest({t: sort}, index, index, metaCollSubFields, [
+                    const expectedAccessPath = t === sort ? forwardIxscan : backwardIxscan;
+                    runRewritesTest({t: sort},
+                                    index,
+                                    index,
+                                    expectedAccessPath,
+                                    metaCollSubFields,
+                                    t === sort,
+                                    [{$match: {'m.a': 5, 'm.b': 5}}]);
+                    runRewritesTest(
+                        {t: sort}, index, null, expectedAccessPath, metaCollSubFields, t === sort, [
                             {$match: {'m.a': 5, 'm.b': 5}}
                         ]);
                 }
@@ -494,11 +554,15 @@ for (const ixA of [-1, +1]) {
                         // the index key. The index and sort are compatible iff they agree on
                         // whether or not these two fields are in the same direction.
                         if (ixB * ixT === sortB * sortT) {
-                            // TODO SERVER-64994 will allow reverse scan.
-                            if (ixT === sortT)
-                                runRewritesTest(sort, ix, ix, metaCollSubFields, true, predicate);
-                            else
-                                runDoesntRewriteTest(sort, ix, ix, metaCollSubFields, predicate);
+                            runRewritesTest(
+                                sort, ix, ix, null, metaCollSubFields, ixT === sortT, predicate);
+                            runRewritesTest(sort,
+                                            ix,
+                                            null,
+                                            ixT === sortT ? forwardIxscan : backwardIxscan,
+                                            metaCollSubFields,
+                                            ixT === sortT,
+                                            predicate);
                         } else {
                             runDoesntRewriteTest(sort, ix, ix, metaCollSubFields, predicate);
                         }
@@ -530,11 +594,15 @@ for (const ixA of [-1, +1]) {
                         // in the same direction.
                         const predicate = [{$match: {'m.a': {$gte: -999, $lte: 999}, 'm.b': 7}}];
                         if (ixA * ixT === sortA * sortT) {
-                            // TODO SERVER-64994 will allow reverse scan.
-                            if (ixT === sortT)
-                                runRewritesTest(sort, ix, ix, metaCollSubFields, true, predicate);
-                            else
-                                runDoesntRewriteTest(sort, ix, ix, metaCollSubFields, predicate);
+                            runRewritesTest(
+                                sort, ix, ix, null, metaCollSubFields, ixT === sortT, predicate);
+                            runRewritesTest(sort,
+                                            ix,
+                                            null,
+                                            ixT === sortT ? forwardIxscan : backwardIxscan,
+                                            metaCollSubFields,
+                                            ixT === sortT,
+                                            predicate);
                         } else {
                             runDoesntRewriteTest(sort, ix, ix, metaCollSubFields, predicate);
                         }
@@ -578,8 +646,12 @@ runDoesntRewriteTest({t: 1},
 {
     // When the collation of the query matches the index, an equality predicate in the query
     // becomes a 1-point interval in the index bounds.
-    runRewritesTest({t: 1}, {m: 1, t: 1}, {m: 1, t: 1}, csStringColl, true, [{$match: {m: 'a'}}]);
-    runRewritesTest({t: 1}, {m: 1, t: 1}, {m: 1, t: 1}, ciStringColl, true, [{$match: {m: 'a'}}]);
+    runRewritesTest({t: 1}, {m: 1, t: 1}, {m: 1, t: 1}, forwardIxscan, csStringColl, true, [
+        {$match: {m: 'a'}}
+    ]);
+    runRewritesTest({t: 1}, {m: 1, t: 1}, {m: 1, t: 1}, forwardIxscan, ciStringColl, true, [
+        {$match: {m: 'a'}}
+    ]);
     // When the collation doesn't match, then the equality predicate is not a 1-point interval
     // in the index.
     csStringColl.dropIndexes();
diff --git a/jstests/core/timeseries/bucket_unpacking_with_sort_plan_cache.js b/jstests/core/timeseries/bucket_unpacking_with_sort_plan_cache.js
index 4e8783e7375..25e1c99312e 100644
--- a/jstests/core/timeseries/bucket_unpacking_with_sort_plan_cache.js
+++ b/jstests/core/timeseries/bucket_unpacking_with_sort_plan_cache.js
@@ -66,68 +66,83 @@ const bucketsName = "system.buckets." + collName;
 const stageName = "$_internalBoundedSort";
 const bucketsColl = db[bucketsName];
 
-const numDocs = 20;
-// Setup with a few documents.
-setupCollection(coll, collName, numDocs);
-
-// Create indexes so that we have something to multiplan.
-assert.commandWorked(coll.createIndex({"m.a": 1, "m.i": 1, t: 1}));
-assert.commandWorked(coll.createIndex({"m.b": 1, "m.i": 1, t: 1}));
-
-// Check that the rewrite is performed before caching.
-const pipeline = [{$sort: {"m.i": 1, t: 1}}, {$match: {"m.a": 1, "m.b": 1}}];
-let explain = coll.explain().aggregate(pipeline);
-assert.eq(getAggPlanStages(explain, stageName).length, 1, explain);
-
-// Check the cache is empty.
-assert.eq(db[bucketsName].getPlanCache().list().length, 0);
-
-// Run in order to cache the plan.
-let result = coll.aggregate(pipeline).toArray();
-assert.eq(result.length, 20, result);
-
-// Check the answer was cached.
-assert.eq(db[bucketsName].getPlanCache().list().length, 1);
-
-// Check that the solution still uses internal bounded sort.
-explain = coll.explain().aggregate(pipeline);
-assert(getAggPlanStages(explain, stageName).length === 1, explain);
-
-// Get constants needed for replanning.
-const cursorStageName = "$cursor";
-const planCacheKey =
-    getPlanCacheKeyFromExplain(getAggPlanStage(explain, cursorStageName)[cursorStageName], db);
-const planCacheEntry = (() => {
-    const planCache = bucketsColl.getPlanCache().list([{$match: {planCacheKey}}]);
-    assert.eq(planCache.length, 1, planCache);
-    return planCache[0];
-})();
-let ratio = (() => {
-    const getParamRes = assert.commandWorked(
-        db.adminCommand({getParameter: 1, internalQueryCacheEvictionRatio: 1}));
-    return getParamRes["internalQueryCacheEvictionRatio"];
-})();
-
-// Remove existing docs, add docs to trigger replanning.
-assert.commandWorked(coll.deleteMany({"m.a": 1, "m.b": 1}));
-let numNewDocs = ratio * planCacheEntry.works + 1;
-addDocs(coll, numNewDocs, [1, 0]);
-addDocs(coll, numNewDocs, [0, 1]);
-
-// Turn on profiling.
-db.setProfilingLevel(2);
-
-// Rerun command with replanning.
-const comment = jsTestName();
-result = coll.aggregate(pipeline, {comment}).toArray();
-assert.eq(result.length, 0);
-
-// Check that the plan was replanned.
-const replanProfileEntry = getLatestProfilerEntry(db, {'command.comment': comment});
-assert(replanProfileEntry.replanned, replanProfileEntry);
+const testBoundedSorterPlanCache = (sortDirection, indexDirection) => {
+    // Setup with a few documents.
+    const numDocs = 20;
+    setupCollection(coll, collName, numDocs);
+
+    assert.commandWorked(
+        coll.createIndex({"m.a": indexDirection, "m.i": indexDirection, t: indexDirection}));
+    assert.commandWorked(
+        coll.createIndex({"m.b": indexDirection, "m.i": indexDirection, t: indexDirection}));
+
+    // Check that the rewrite is performed before caching.
+    const pipeline =
+        [{$sort: {"m.i": sortDirection, t: sortDirection}}, {$match: {"m.a": 1, "m.b": 1}}];
+    let explain = coll.explain().aggregate(pipeline);
+    assert.eq(getAggPlanStages(explain, stageName).length, 1, explain);
+    const traversalDirection = sortDirection === indexDirection ? "forward" : "backward";
+    assert.eq(getAggPlanStage(explain, "IXSCAN").direction, traversalDirection, explain);
+
+    // Check the cache is empty.
+    assert.eq(db[bucketsName].getPlanCache().list().length, 0);
+
+    // Run in order to cache the plan.
+    let result = coll.aggregate(pipeline).toArray();
+    assert.eq(result.length, 20, result);
+
+    // Check the answer was cached.
+    assert.eq(db[bucketsName].getPlanCache().list().length, 1);
+
+    // Check that the solution still uses internal bounded sort with the correct order.
+    explain = coll.explain().aggregate(pipeline);
+    assert.eq(getAggPlanStages(explain, stageName).length, 1, explain);
+    assert.eq(getAggPlanStage(explain, "IXSCAN").direction, traversalDirection, explain);
+
+    // Get constants needed for replanning.
+    const cursorStageName = "$cursor";
+    const planCacheKey =
+        getPlanCacheKeyFromExplain(getAggPlanStage(explain, cursorStageName)[cursorStageName], db);
+    const planCacheEntry = (() => {
+        const planCache = bucketsColl.getPlanCache().list([{$match: {planCacheKey}}]);
+        assert.eq(planCache.length, 1, planCache);
+        return planCache[0];
+    })();
+    let ratio = (() => {
+        const getParamRes = assert.commandWorked(
+            db.adminCommand({getParameter: 1, internalQueryCacheEvictionRatio: 1}));
+        return getParamRes["internalQueryCacheEvictionRatio"];
+    })();
+
+    // Remove existing docs, add docs to trigger replanning.
+    assert.commandWorked(coll.deleteMany({"m.a": 1, "m.b": 1}));
+    let numNewDocs = ratio * planCacheEntry.works + 1;
+    addDocs(coll, numNewDocs, [1, 0]);
+    addDocs(coll, numNewDocs, [0, 1]);
+
+    // Turn on profiling.
+    db.setProfilingLevel(2);
+
+    // Rerun command with replanning.
+    const comment = jsTestName();
+    result = coll.aggregate(pipeline, {comment}).toArray();
+    assert.eq(result.length, 0);
+
+    // Check that the plan was replanned.
+    const replanProfileEntry = getLatestProfilerEntry(db, {'command.comment': comment});
+    assert(replanProfileEntry.replanned, replanProfileEntry);
+
+    // Check that rewrite happens with replanning.
+    explain = coll.explain().aggregate(pipeline);
+    assert.eq(getAggPlanStages(explain, stageName).length,
+              1,
+              {explain, stages: getAggPlanStages(explain, stageName)});
+    assert.eq(getAggPlanStage(explain, "IXSCAN").direction, traversalDirection, explain);
+};
 
-// Check that rewrite happens with replanning.
-explain = coll.explain().aggregate(pipeline);
-assert(getAggPlanStages(explain, stageName).length === 1,
-       {explain, stages: getAggPlanStages(explain, stageName)});
+for (const sortDirection of [-1, 1]) {
+    for (const indexDirection of [-1, 1]) {
+        testBoundedSorterPlanCache(sortDirection, indexDirection);
+    }
+}
 })();
diff --git a/jstests/core/timeseries/timeseries_internal_bounded_sort.js b/jstests/core/timeseries/timeseries_internal_bounded_sort.js
index c909fe8e397..45cb1af79da 100644
--- a/jstests/core/timeseries/timeseries_internal_bounded_sort.js
+++ b/jstests/core/timeseries/timeseries_internal_bounded_sort.js
@@ -105,21 +105,12 @@ function runTest(ascending) {
 
         // Check plan using control.max.t
         if (ascending) {
-            // TODO (SERVER-64994): We can remove this manual re-write once we support index
-            // direction hints
-            const opt = buckets
-                            .aggregate([
-                                {$sort: {'control.max.t': ascending ? 1 : -1}},
-                                unpackStage,
-                                {
-                                    $_internalBoundedSort: {
-                                        sortKey: {t: 1},
-                                        bound: {base: "max", offsetSeconds: -bucketMaxSpanSeconds}
-                                    }
-                                },
-                            ])
+            const opt = coll.aggregate(
+                                [
+                                    {$sort: {t: 1}},
+                                ],
+                                {hint: {t: -1}})
                             .toArray();
-
             assertSorted(opt, ascending);
             assert.eq(reference, opt);
         } else {
diff --git a/jstests/core/timeseries/timeseries_internal_bounded_sort_compound.js b/jstests/core/timeseries/timeseries_internal_bounded_sort_compound.js
index c579ecc6e3e..6f3a57f39e2 100644
--- a/jstests/core/timeseries/timeseries_internal_bounded_sort_compound.js
+++ b/jstests/core/timeseries/timeseries_internal_bounded_sort_compound.js
@@ -152,38 +152,17 @@ function runTest(sortSpec) {
                                   sortSpec,
                                   sortSpec);
         } else {
-            // TODO (SERVER-64994): We can remove this manual re-write once we support index
-            // direction hints
-            const optFromMinQuery = [
-                {$sort: {meta: sortSpec.m, 'control.min.t': sortSpec.t}},
-                unpackStage,
-                {
-                    $_internalBoundedSort: {
-                        sortKey: sortSpec,
-                        bound: {base: "min", offsetSeconds: bucketMaxSpanSeconds}
-                    }
-                },
-            ];
-            const optFromMin = buckets.aggregate(optFromMinQuery).toArray();
+            const optFromMinQuery = [{$sort: {m: sortSpec.m, t: sortSpec.t}}];
+            const optFromMin = coll.aggregate(optFromMinQuery).toArray();
             assertSorted(optFromMin, sortSpec);
             assert.eq(reference, optFromMin);
         }
 
         // Check plan using control.max.t
         if (sortSpec.t > 0) {
-            // TODO (SERVER-64994): We can remove this manual re-write once we support index
-            // direction hints
-            const optFromMaxQuery = [
-                {$sort: {meta: sortSpec.m, 'control.max.t': sortSpec.t}},
-                unpackStage,
-                {
-                    $_internalBoundedSort: {
-                        sortKey: sortSpec,
-                        bound: {base: "max", offsetSeconds: -bucketMaxSpanSeconds}
-                    }
-                },
-            ];
-            const optFromMax = buckets.aggregate(optFromMaxQuery).toArray();
+            const optFromMaxQuery = [{$sort: {m: sortSpec.m, t: sortSpec.t}}];
+            const optFromMax =
+                coll.aggregate(optFromMaxQuery, {hint: {m: -sortSpec.m, t: -sortSpec.t}}).toArray();
             assertSorted(optFromMax, sortSpec);
             assert.eq(reference, optFromMax);
         } else {
diff --git a/jstests/noPassthrough/timeseries_sort.js b/jstests/noPassthrough/timeseries_sort.js
new file mode 100644
index 00000000000..53842e20c1c
--- /dev/null
+++ b/jstests/noPassthrough/timeseries_sort.js
@@ -0,0 +1,131 @@
+/**
+ * Test that we correctly use the index created when a time series collection is sharded.
+ *
+ * @tags: [
+ *   requires_fcv_51,
+ *   requires_find_command,
+ *   requires_sharding,
+ * ]
+ */
+(function() {
+"use strict";
+
+load("jstests/core/timeseries/libs/timeseries.js");
+load("jstests/libs/analyze_plan.js");  // For getAggPlanStage
+
+Random.setRandomSeed();
+
+const dbName = 'testDB';
+const collName = 'timeseries_sort';
+const timeField = 't';
+const metaField = 'm';
+
+const bucketsCollName = `system.buckets.${collName}`;
+const fullBucketsCollName = `${dbName}.system.buckets.${collName}`;
+
+const st = new ShardingTest({shards: 2});
+const sDB = st.s.getDB(dbName);
+assert.commandWorked(sDB.adminCommand({enableSharding: dbName}));
+
+if (!TimeseriesTest.shardedtimeseriesCollectionsEnabled(st.shard0)) {
+    jsTestLog("Skipping test because the sharded time-series collection feature flag is disabled");
+    st.stop();
+    return;
+}
+
+st.ensurePrimaryShard(dbName, st.shard0.shardName);
+
+// Shard time-series collection.
+const shardKey = {
+    [timeField]: 1
+};
+assert.commandWorked(sDB.adminCommand({
+    shardCollection: `${dbName}.${collName}`,
+    key: shardKey,
+    timeseries: {timeField, metaField, granularity: "hours"}
+}));
+
+// Split the chunks.
+const splitPoint = {
+    [`control.min.${timeField}`]: new Date(50 * 1000)
+};
+assert.commandWorked(sDB.adminCommand({split: fullBucketsCollName, middle: splitPoint}));
+
+// // Move one of the chunks into the second shard.
+const primaryShard = st.getPrimaryShard(dbName);
+const otherShard = st.getOther(primaryShard);
+assert.commandWorked(sDB.adminCommand(
+    {movechunk: fullBucketsCollName, find: splitPoint, to: otherShard.name, _waitForDelete: true}));
+
+const coll = sDB.getCollection(collName);
+const bucketsColl = sDB.getCollection(bucketsCollName);
+
+const hasInternalBoundedSort = (explain) => {
+    for (const shardName in explain.shards) {
+        const pipeline = explain.shards[shardName].stages;
+        if (!pipeline.some((stage) => stage.hasOwnProperty("$_internalBoundedSort"))) {
+            return false;
+        }
+    }
+    return true;
+};
+
+const assertAccessPath = (pipeline, hint, accessPath, direction) => {
+    const options = (hint) ? {hint: hint} : {};
+    const explain = coll.explain().aggregate(pipeline, options);
+    assert(hasInternalBoundedSort(explain));
+
+    const paths = getAggPlanStages(explain, accessPath);
+    for (const path of paths) {
+        assert.eq(path.stage, accessPath);
+        assert.eq(path.direction, direction > 0 ? "forward" : "backward");
+    }
+};
+
+const assertNoRewrite = (pipeline) => {
+    const explain = coll.explain().aggregate(pipeline);
+    assert(!hasInternalBoundedSort(explain));
+};
+
+for (let i = 0; i < 100; i++) {
+    assert.commandWorked(
+        sDB.getCollection(collName).insert({t: new Date(i * 1000), m: i % 4, k: i}));
+}
+
+// Ensure that each shard owns one chunk.
+const counts = st.chunkCounts(bucketsCollName, dbName);
+assert.eq(1, counts[primaryShard.shardName], counts);
+assert.eq(1, counts[otherShard.shardName], counts);
+
+assert.eq(coll.count(), 100);
+assert.eq(bucketsColl.count(), 4);
+
+assert.eq(coll.getIndexes().length, 1);
+assert.eq(coll.getIndexes()[0].name, "control.min.t_1");
+
+const forwardSort = {
+    $sort: {t: 1}
+};
+const backwardSort = {
+    $sort: {t: -1}
+};
+// One match before the split, one after the split.
+for (const matchDate of [new Date(25 * 1000), new Date(75 * 1000)]) {
+    const match = {$match: {t: matchDate}};
+    assertAccessPath([match, forwardSort], null, "IXSCAN", 1);
+    assertAccessPath([match, backwardSort], null, "IXSCAN", -1);
+    assertNoRewrite([match, {$sort: {t: -1, m: 1}}]);
+    assertNoRewrite([match, {$sort: {t: 1, m: 1}}]);
+}
+const kMatch = {
+    $match: {k: 1}
+};
+assertAccessPath([forwardSort], null, "COLLSCAN", 1);
+assertAccessPath([backwardSort], null, "COLLSCAN", -1);
+assertAccessPath([kMatch, forwardSort], null, "COLLSCAN", 1);
+assertAccessPath([kMatch, backwardSort], null, "COLLSCAN", -1);
+assertAccessPath([forwardSort], {$natural: -1}, "COLLSCAN", 1);
+assertAccessPath([backwardSort], {$natural: 1}, "COLLSCAN", -1);
+
+st.stop();
+})();
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index b57dbab5fdc..619bd431584 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -36,6 +36,7 @@
 #include "mongo/db/pipeline/pipeline_d.h"
 
 #include "mongo/base/exact_cast.h"
+#include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/bson/simple_bsonobj_comparator.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/database.h"
@@ -87,9 +88,11 @@
 #include "mongo/db/query/query_feature_flags_gen.h"
 #include "mongo/db/query/query_knobs_gen.h"
 #include "mongo/db/query/query_planner.h"
+#include "mongo/db/query/query_planner_params.h"
 #include "mongo/db/query/sort_pattern.h"
 #include "mongo/db/query/stage_types.h"
 #include "mongo/db/s/collection_sharding_state.h"
+#include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/top.h"
 #include "mongo/db/storage/record_store.h"
@@ -244,7 +247,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> attemptToGetExe
     SkipThenLimit skipThenLimit,
     boost::optional<std::string> groupIdForDistinctScan,
     const AggregateCommandRequest* aggRequest,
-    const size_t plannerOpts,
+    const QueryPlannerParams& plannerOpts,
     const MatchExpressionParser::AllowedFeatureSet& matcherFeatures,
     Pipeline* pipeline) {
     auto findCommand = std::make_unique<FindCommandRequest>(nss);
@@ -313,7 +316,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> attemptToGetExe
         // index would produce one result for '1' and another for '2', which would be incorrect.
         auto distinctExecutor =
             getExecutorDistinct(&collections.getMainCollection(),
-                                plannerOpts | QueryPlannerParams::STRICT_DISTINCT_ONLY,
+                                plannerOpts.options | QueryPlannerParams::STRICT_DISTINCT_ONLY,
                                 &parsedDistinct);
         if (!distinctExecutor.isOK()) {
             return distinctExecutor.getStatus().withContext(
@@ -1111,6 +1114,41 @@ bool PipelineD::sortAndKeyPatternPartAgreeAndOnMeta(const BucketUnpacker& bucket
     return (keyPatternFieldPath.tail() == sortFieldPath.tail());
 }
 
+boost::optional<TraversalPreference> createTimeSeriesTraversalPreference(
+    DocumentSourceInternalUnpackBucket* unpack, DocumentSourceSort* sort) {
+    const auto metaField = unpack->bucketUnpacker().getMetaField();
+    BSONObjBuilder builder;
+    // Reverse the sort pattern so we can look for indexes that match.
+    for (const auto& sortPart : sort->getSortKeyPattern()) {
+        if (!sortPart.fieldPath) {
+            return boost::none;
+        }
+        const int reversedDirection = sortPart.isAscending ? -1 : 1;
+        const auto& path = sortPart.fieldPath->fullPath();
+        if (metaField.has_value() &&
+            (expression::isPathPrefixOf(*metaField, path) || *metaField == path)) {
+            std::string rewrittenField =
+                std::string{timeseries::kBucketMetaFieldName} + path.substr(metaField->size());
+            builder.append(rewrittenField, reversedDirection);
+        } else if (path == unpack->bucketUnpacker().getTimeField()) {
+            if (reversedDirection == 1) {
+                builder.append(unpack->bucketUnpacker().getMinField(path), reversedDirection);
+            } else {
+                builder.append(unpack->bucketUnpacker().getMaxField(path), reversedDirection);
+            }
+        } else {
+            // The field wasn't meta or time, so no direction preference should be made.
+            return boost::none;
+        }
+    }
+
+    TraversalPreference traversalPreference;
+    traversalPreference.sortPattern = builder.obj();
+    traversalPreference.clusterField = unpack->getMinTimeField();
+    traversalPreference.direction = -1;
+    return traversalPreference;
+}
+
 std::pair<PipelineD::AttachExecutorCallback, std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>>
 PipelineD::buildInnerQueryExecutorGeneric(const MultipleCollectionAccessor& collections,
                                           const NamespaceString& nss,
@@ -1166,6 +1204,19 @@ PipelineD::buildInnerQueryExecutorGeneric(const MultipleCollectionAccessor& coll
         ? DepsTracker::kDefaultUnavailableMetadata & ~DepsTracker::kOnlyTextScore
         : DepsTracker::kDefaultUnavailableMetadata;
 
+    // If this is a query on a time-series collection then it may be eligible for a post-planning
+    // sort optimization. We check eligibility and perform the rewrite here.
+    auto [unpack, sort] = findUnpackThenSort(pipeline->_sources);
+    QueryPlannerParams plannerOpts;
+    if (serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+        serverGlobalParams.featureCompatibility.isGreaterThanOrEqualTo(
+            multiversion::FeatureCompatibilityVersion::kVersion_6_0) &&
+        feature_flags::gFeatureFlagBucketUnpackWithSort.isEnabled(
+            serverGlobalParams.featureCompatibility) &&
+        unpack && sort) {
+        plannerOpts.traversalPreference = createTimeSeriesTraversalPreference(unpack, sort);
+    }
+
     // Create the PlanExecutor.
     bool shouldProduceEmptyDocs = false;
     auto exec = uassertStatusOK(prepareExecutor(expCtx,
@@ -1179,11 +1230,11 @@ PipelineD::buildInnerQueryExecutorGeneric(const MultipleCollectionAccessor& coll
                                                 skipThenLimit,
                                                 aggRequest,
                                                 Pipeline::kAllowedMatcherFeatures,
-                                                &shouldProduceEmptyDocs));
+                                                &shouldProduceEmptyDocs,
+                                                std::move(plannerOpts)));
 
     // If this is a query on a time-series collection then it may be eligible for a post-planning
     // sort optimization. We check eligibility and perform the rewrite here.
-    auto [unpack, sort] = findUnpackThenSort(pipeline->_sources);
     if (serverGlobalParams.featureCompatibility.isVersionInitialized() &&
         serverGlobalParams.featureCompatibility.isGreaterThanOrEqualTo(
             multiversion::FeatureCompatibilityVersion::kVersion_6_0) &&
@@ -1513,24 +1564,22 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
     SkipThenLimit skipThenLimit,
     const AggregateCommandRequest* aggRequest,
     const MatchExpressionParser::AllowedFeatureSet& matcherFeatures,
-    bool* hasNoRequirements) {
+    bool* hasNoRequirements,
+    QueryPlannerParams plannerOpts) {
     invariant(hasNoRequirements);
 
-    // Any data returned from the inner executor must be owned.
-    size_t plannerOpts = QueryPlannerParams::DEFAULT;
-
     bool isChangeStream =
         pipeline->peekFront() && pipeline->peekFront()->constraints().isChangeStreamStage();
     if (isChangeStream) {
         invariant(expCtx->tailableMode == TailableModeEnum::kTailableAndAwaitData);
-        plannerOpts |= (QueryPlannerParams::TRACK_LATEST_OPLOG_TS |
-                        QueryPlannerParams::ASSERT_MIN_TS_HAS_NOT_FALLEN_OFF_OPLOG);
+        plannerOpts.options |= (QueryPlannerParams::TRACK_LATEST_OPLOG_TS |
+                                QueryPlannerParams::ASSERT_MIN_TS_HAS_NOT_FALLEN_OFF_OPLOG);
     }
 
     // The $_requestReshardingResumeToken parameter is only valid for an oplog scan.
     if (aggRequest && aggRequest->getRequestReshardingResumeToken()) {
-        plannerOpts |= (QueryPlannerParams::TRACK_LATEST_OPLOG_TS |
-                        QueryPlannerParams::ASSERT_MIN_TS_HAS_NOT_FALLEN_OFF_OPLOG);
+        plannerOpts.options |= (QueryPlannerParams::TRACK_LATEST_OPLOG_TS |
+                                QueryPlannerParams::ASSERT_MIN_TS_HAS_NOT_FALLEN_OFF_OPLOG);
     }
 
     // If there is a sort stage eligible for pushdown, serialize its SortPattern to a BSONObj. The
@@ -1570,7 +1619,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
     if (*hasNoRequirements) {
         // This query might be eligible for count optimizations, since the remaining stages in the
         // pipeline don't actually need to read any data produced by the query execution layer.
-        plannerOpts |= QueryPlannerParams::IS_COUNT;
+        plannerOpts.options |= QueryPlannerParams::IS_COUNT;
     } else {
         // Build a BSONObj representing a projection eligible for pushdown. If there is an inclusion
         // projection at the front of the pipeline, it will be removed and handled by the PlanStage
@@ -1588,7 +1637,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> PipelineD::prep
         // top-k sort, which both sorts and limits.)
         bool allowExpressions = !sortStage && !skipThenLimit.getSkip() && !skipThenLimit.getLimit();
         projObj = buildProjectionForPushdown(deps, pipeline, allowExpressions);
-        plannerOpts |= QueryPlannerParams::RETURN_OWNED_DATA;
+        plannerOpts.options |= QueryPlannerParams::RETURN_OWNED_DATA;
     }
 
     if (rewrittenGroupStage) {
diff --git a/src/mongo/db/pipeline/pipeline_d.h b/src/mongo/db/pipeline/pipeline_d.h
index cd40bc33b8b..c109e75b1b8 100644
--- a/src/mongo/db/pipeline/pipeline_d.h
+++ b/src/mongo/db/pipeline/pipeline_d.h
@@ -30,6 +30,7 @@
 #pragma once
 
 #include "mongo/db/exec/bucket_unpacker.h"
+#include "mongo/db/query/query_planner_params.h"
 #include <boost/intrusive_ptr.hpp>
 #include <memory>
 
@@ -44,6 +45,7 @@
 #include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/query/multiple_collection_accessor.h"
 #include "mongo/db/query/plan_executor.h"
+#include "mongo/db/query/query_planner.h"
 
 namespace mongo {
 class Collection;
@@ -202,7 +204,8 @@ private:
         SkipThenLimit skipThenLimit,
         const AggregateCommandRequest* aggRequest,
         const MatchExpressionParser::AllowedFeatureSet& matcherFeatures,
-        bool* hasNoRequirements);
+        bool* hasNoRequirements,
+        QueryPlannerParams plannerOpts = QueryPlannerParams{});
 
     /**
      * Build a PlanExecutor and prepare a callback to create a special DocumentSourceGeoNearCursor
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index bfb02428c8f..db04d6a276a 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -85,6 +85,7 @@
 #include "mongo/db/query/query_knobs_gen.h"
 #include "mongo/db/query/query_planner.h"
 #include "mongo/db/query/query_planner_common.h"
+#include "mongo/db/query/query_planner_params.h"
 #include "mongo/db/query/query_settings.h"
 #include "mongo/db/query/query_settings_decoration.h"
 #include "mongo/db/query/sbe_cached_solution_planner.h"
@@ -583,10 +584,10 @@ public:
     PrepareExecutionHelper(OperationContext* opCtx,
                            CanonicalQuery* cq,
                            PlanYieldPolicy* yieldPolicy,
-                           size_t plannerOptions)
+                           const QueryPlannerParams& plannerOptions)
         : _opCtx{opCtx}, _cq{cq}, _yieldPolicy{yieldPolicy} {
         invariant(_cq);
-        _plannerParams.options = plannerOptions;
+        _plannerParams = plannerOptions;
     }
 
     /**
@@ -777,7 +778,7 @@ public:
                                   WorkingSet* ws,
                                   CanonicalQuery* cq,
                                   PlanYieldPolicy* yieldPolicy,
-                                  size_t plannerOptions)
+                                  const QueryPlannerParams& plannerOptions)
         : PrepareExecutionHelper{opCtx, std::move(cq), yieldPolicy, plannerOptions},
           _collection(collection),
           _ws{ws} {}
@@ -972,7 +973,10 @@ public:
                                     CanonicalQuery* cq,
                                     PlanYieldPolicy* yieldPolicy,
                                     size_t plannerOptions)
-        : PrepareExecutionHelper{opCtx, std::move(cq), yieldPolicy, plannerOptions},
+        : PrepareExecutionHelper{opCtx,
+                                 std::move(cq),
+                                 yieldPolicy,
+                                 QueryPlannerParams{plannerOptions}},
           _collections(collections) {}
 
     const CollectionPtr& getMainCollection() const override {
@@ -1176,7 +1180,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getClassicExecu
     const CollectionPtr& collection,
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
-    size_t plannerOptions) {
+    const QueryPlannerParams& plannerParams) {
     // Mark that this query uses the classic engine, unless this has already been set.
     OpDebug& opDebug = CurOp::get(opCtx)->debug();
     if (!opDebug.classicEngineUsed) {
@@ -1184,7 +1188,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getClassicExecu
     }
     auto ws = std::make_unique<WorkingSet>();
     ClassicPrepareExecutionHelper helper{
-        opCtx, collection, ws.get(), canonicalQuery.get(), nullptr, plannerOptions};
+        opCtx, collection, ws.get(), canonicalQuery.get(), nullptr, plannerParams};
     auto executionResult = helper.prepare();
     if (!executionResult.isOK()) {
         return executionResult.getStatus();
@@ -1199,7 +1203,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getClassicExecu
                                        std::move(root),
                                        &collection,
                                        yieldPolicy,
-                                       plannerOptions,
+                                       plannerParams.options,
                                        {},
                                        std::move(solution));
 }
@@ -1284,7 +1288,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExe
     const MultipleCollectionAccessor& collections,
     std::unique_ptr<CanonicalQuery> cq,
     PlanYieldPolicy::YieldPolicy requestedYieldPolicy,
-    size_t plannerOptions) {
+    const QueryPlannerParams& plannerParams) {
     // Mark that this query uses the SBE engine, unless this has already been set.
     OpDebug& opDebug = CurOp::get(opCtx)->debug();
     if (!opDebug.classicEngineUsed) {
@@ -1297,7 +1301,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExe
     auto nss = cq->nss();
     auto yieldPolicy = makeSbeYieldPolicy(opCtx, requestedYieldPolicy, mainColl, nss);
     SlotBasedPrepareExecutionHelper helper{
-        opCtx, collections, cq.get(), yieldPolicy.get(), plannerOptions};
+        opCtx, collections, cq.get(), yieldPolicy.get(), plannerParams.options};
     auto planningResultWithStatus = helper.prepare();
     if (!planningResultWithStatus.isOK()) {
         return planningResultWithStatus.getStatus();
@@ -1314,7 +1318,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExe
                                                   planningResult->decisionWorks(),
                                                   planningResult->needsSubplanning(),
                                                   yieldPolicy.get(),
-                                                  plannerOptions)) {
+                                                  plannerParams.options)) {
         // Do the runtime planning and pick the best candidate plan.
         auto candidates = planner->plan(std::move(solutions), std::move(roots));
 
@@ -1322,7 +1326,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExe
                                            std::move(cq),
                                            std::move(candidates),
                                            collections,
-                                           plannerOptions,
+                                           plannerParams.options,
                                            std::move(nss),
                                            std::move(yieldPolicy));
     }
@@ -1353,7 +1357,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getSlotBasedExe
                                        std::move(roots[0]),
                                        {},
                                        collections,
-                                       plannerOptions,
+                                       plannerParams.options,
                                        std::move(nss),
                                        std::move(yieldPolicy));
 }
@@ -1365,11 +1369,11 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     std::function<void(CanonicalQuery*)> extractAndAttachPipelineStages,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
-    size_t plannerOptions) {
+    const QueryPlannerParams& plannerParams) {
     invariant(canonicalQuery);
     const auto& mainColl = collections.getMainCollection();
     canonicalQuery->setSbeCompatible(
-        sbe::isQuerySbeCompatible(&mainColl, canonicalQuery.get(), plannerOptions));
+        sbe::isQuerySbeCompatible(&mainColl, canonicalQuery.get(), plannerParams.options));
 
     // Use SBE if 'canonicalQuery' is SBE compatible.
     if (!canonicalQuery->getForceClassicEngine() && canonicalQuery->isSbeCompatible()) {
@@ -1383,12 +1387,12 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
             canonicalQuery->setSbeCompatible(false);
         } else {
             return getSlotBasedExecutor(
-                opCtx, collections, std::move(canonicalQuery), yieldPolicy, plannerOptions);
+                opCtx, collections, std::move(canonicalQuery), yieldPolicy, plannerParams);
         }
     }
 
     return getClassicExecutor(
-        opCtx, mainColl, std::move(canonicalQuery), yieldPolicy, plannerOptions);
+        opCtx, mainColl, std::move(canonicalQuery), yieldPolicy, plannerParams);
 }
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
@@ -1404,7 +1408,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
                        std::move(canonicalQuery),
                        extractAndAttachPipelineStages,
                        yieldPolicy,
-                       plannerOptions);
+                       QueryPlannerParams{plannerOptions});
 }
 
 //
@@ -1417,13 +1421,13 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     std::function<void(CanonicalQuery*)> extractAndAttachPipelineStages,
     bool permitYield,
-    size_t plannerOptions) {
+    QueryPlannerParams plannerParams) {
     auto yieldPolicy = (permitYield && !opCtx->inMultiDocumentTransaction())
         ? PlanYieldPolicy::YieldPolicy::YIELD_AUTO
         : PlanYieldPolicy::YieldPolicy::INTERRUPT_ONLY;
 
     if (OperationShardingState::isComingFromRouter(opCtx)) {
-        plannerOptions |= QueryPlannerParams::INCLUDE_SHARD_FILTER;
+        plannerParams.options |= QueryPlannerParams::INCLUDE_SHARD_FILTER;
     }
 
     return getExecutor(opCtx,
@@ -1431,7 +1435,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind
                        std::move(canonicalQuery),
                        extractAndAttachPipelineStages,
                        yieldPolicy,
-                       plannerOptions);
+                       plannerParams);
 }
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind(
@@ -1447,7 +1451,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind
                            std::move(canonicalQuery),
                            extractAndAttachPipelineStages,
                            permitYield,
-                           plannerOptions);
+                           QueryPlannerParams{plannerOptions});
 }
 
 namespace {
diff --git a/src/mongo/db/query/get_executor.h b/src/mongo/db/query/get_executor.h
index 20ca265bcbb..e913108679c 100644
--- a/src/mongo/db/query/get_executor.h
+++ b/src/mongo/db/query/get_executor.h
@@ -42,6 +42,7 @@
 #include "mongo/db/query/multiple_collection_accessor.h"
 #include "mongo/db/query/parsed_distinct.h"
 #include "mongo/db/query/plan_executor.h"
+#include "mongo/db/query/query_planner.h"
 #include "mongo/db/query/query_planner_params.h"
 #include "mongo/db/query/query_settings.h"
 #include "mongo/db/query/query_solution.h"
@@ -157,7 +158,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     std::function<void(CanonicalQuery*)> extractAndAttachPipelineStages,
     PlanYieldPolicy::YieldPolicy yieldPolicy,
-    size_t plannerOptions = 0);
+    const QueryPlannerParams& plannerOptions);
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutor(
     OperationContext* opCtx,
@@ -192,7 +193,7 @@ StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind
     std::unique_ptr<CanonicalQuery> canonicalQuery,
     std::function<void(CanonicalQuery*)> extractAndAttachPipelineStages,
     bool permitYield = false,
-    size_t plannerOptions = QueryPlannerParams::DEFAULT);
+    QueryPlannerParams plannerOptions = QueryPlannerParams{});
 
 StatusWith<std::unique_ptr<PlanExecutor, PlanExecutor::Deleter>> getExecutorFind(
     OperationContext* opCtx,
diff --git a/src/mongo/db/query/planner_access.cpp b/src/mongo/db/query/planner_access.cpp
index dbf0751d60e..e05dfd8a5d1 100644
--- a/src/mongo/db/query/planner_access.cpp
+++ b/src/mongo/db/query/planner_access.cpp
@@ -379,7 +379,9 @@ std::unique_ptr<QuerySolutionNode> QueryPlannerAccess::makeCollectionScan(
     const BSONObj& hint = query.getFindCommandRequest().getHint();
     if (!hint.isEmpty()) {
         BSONElement natural = hint[query_request_helper::kNaturalSortField];
-        if (natural) {
+        // If we have a natural hint and a time series traversal preference, let the traversal
+        // preference decide what order to scan, so that we can avoid a blocking sort.
+        if (natural && !params.traversalPreference) {
             // If the hint is {$natural: +-1} this changes the direction of the collection scan.
             csn->direction = natural.safeNumberInt() >= 0 ? 1 : -1;
         }
diff --git a/src/mongo/db/query/planner_access.h b/src/mongo/db/query/planner_access.h
index 3a133aae486..6ea44830415 100644
--- a/src/mongo/db/query/planner_access.h
+++ b/src/mongo/db/query/planner_access.h
@@ -35,6 +35,7 @@
 #include "mongo/db/query/index_bounds_builder.h"
 #include "mongo/db/query/index_tag.h"
 #include "mongo/db/query/interval_evaluation_tree.h"
+#include "mongo/db/query/query_planner.h"
 #include "mongo/db/query/query_planner_params.h"
 #include "mongo/db/query/query_solution.h"
 
diff --git a/src/mongo/db/query/planner_analysis.cpp b/src/mongo/db/query/planner_analysis.cpp
index 4fb673d8bfc..40d8d7b0d0d 100644
--- a/src/mongo/db/query/planner_analysis.cpp
+++ b/src/mongo/db/query/planner_analysis.cpp
@@ -894,6 +894,41 @@ bool QueryPlannerAnalysis::explodeForSort(const CanonicalQuery& query,
     return true;
 }
 
+// This function is used to check if the given index pattern and direction in the traversal
+// preference can be used to satisfy the given sort pattern (specifically for time series
+// collections).
+bool sortMatchesTraversalPreference(const TraversalPreference& traversalPreference,
+                                    const BSONObj& indexPattern) {
+    BSONObjIterator sortIter(traversalPreference.sortPattern);
+    BSONObjIterator indexIter(indexPattern);
+    while (sortIter.more() && indexIter.more()) {
+        BSONElement sortPart = sortIter.next();
+        BSONElement indexPart = indexIter.next();
+
+        if (!sortPart.isNumber() || !indexPart.isNumber()) {
+            return false;
+        }
+
+        // If the field doesn't match or the directions don't match, we return false.
+        if (strcmp(sortPart.fieldName(), indexPart.fieldName()) != 0 ||
+            (sortPart.safeNumberInt() > 0) != (indexPart.safeNumberInt() > 0)) {
+            return false;
+        }
+    }
+
+    if (!indexIter.more() && sortIter.more()) {
+        // The sort still has more, so it cannot be a prefix of the index.
+        return false;
+    }
+    return true;
+}
+
+bool isShardedCollScan(QuerySolutionNode* solnRoot) {
+    return solnRoot->getType() == StageType::STAGE_SHARDING_FILTER &&
+        solnRoot->children.size() == 1 &&
+        solnRoot->children[0]->getType() == StageType::STAGE_COLLSCAN;
+}
+
 // static
 QuerySolutionNode* QueryPlannerAnalysis::analyzeSort(const CanonicalQuery& query,
                                                      const QueryPlannerParams& params,
@@ -906,6 +941,28 @@ QuerySolutionNode* QueryPlannerAnalysis::analyzeSort(const CanonicalQuery& query
             "ntoreturn on the find command should not be set",
             findCommand.getNtoreturn() == boost::none);
 
+    if (params.traversalPreference) {
+        // If we've been passed a traversal preference, we might want to reverse the order we scan
+        // the data to avoid a blocking sort later in the pipeline.
+        auto providedSorts = solnRoot->providedSorts();
+
+        BSONObj solnSortPattern;
+        if (solnRoot->getType() == StageType::STAGE_COLLSCAN || isShardedCollScan(solnRoot)) {
+            BSONObjBuilder builder;
+            builder.append(params.traversalPreference->clusterField, 1);
+            solnSortPattern = builder.obj();
+        } else {
+            solnSortPattern = providedSorts.getBaseSortPattern();
+        }
+
+        if (sortMatchesTraversalPreference(params.traversalPreference.get(), solnSortPattern) &&
+            QueryPlannerCommon::scanDirectionsEqual(solnRoot,
+                                                    -params.traversalPreference->direction)) {
+            QueryPlannerCommon::reverseScans(solnRoot, true);
+            return solnRoot;
+        }
+    }
+
     const BSONObj& sortObj = findCommand.getSort();
 
     if (sortObj.isEmpty()) {
diff --git a/src/mongo/db/query/planner_analysis.h b/src/mongo/db/query/planner_analysis.h
index 106a2f48a31..b48b94c9605 100644
--- a/src/mongo/db/query/planner_analysis.h
+++ b/src/mongo/db/query/planner_analysis.h
@@ -30,6 +30,7 @@
 #pragma once
 
 #include "mongo/db/query/canonical_query.h"
+#include "mongo/db/query/query_planner.h"
 #include "mongo/db/query/query_planner_params.h"
 #include "mongo/db/query/query_solution.h"
 
diff --git a/src/mongo/db/query/query_planner.cpp b/src/mongo/db/query/query_planner.cpp
index 7b4c144c906..c258c9e6867 100644
--- a/src/mongo/db/query/query_planner.cpp
+++ b/src/mongo/db/query/query_planner.cpp
@@ -31,15 +31,15 @@
 
 #include "mongo/platform/basic.h"
 
-#include "mongo/db/query/query_planner.h"
-
 #include <boost/optional.hpp>
 #include <vector>
 
 #include "mongo/base/string_data.h"
+#include "mongo/bson/bsonobj.h"
 #include "mongo/bson/simple_bsonelement_comparator.h"
 #include "mongo/db/bson/dotted_path_support.h"
 #include "mongo/db/catalog/clustered_collection_util.h"
+#include "mongo/db/exec/bucket_unpacker.h"
 #include "mongo/db/index/wildcard_key_generator.h"
 #include "mongo/db/index_names.h"
 #include "mongo/db/matcher/expression_algo.h"
@@ -51,14 +51,19 @@
 #include "mongo/db/query/classic_plan_cache.h"
 #include "mongo/db/query/collation/collation_index_key.h"
 #include "mongo/db/query/collation/collator_interface.h"
+#include "mongo/db/query/internal_plans.h"
 #include "mongo/db/query/plan_cache.h"
 #include "mongo/db/query/plan_enumerator.h"
 #include "mongo/db/query/planner_access.h"
 #include "mongo/db/query/planner_analysis.h"
 #include "mongo/db/query/planner_ixselect.h"
+#include "mongo/db/query/projection_parser.h"
+#include "mongo/db/query/query_knobs_gen.h"
+#include "mongo/db/query/query_planner.h"
 #include "mongo/db/query/query_planner_common.h"
 #include "mongo/db/query/query_solution.h"
 #include "mongo/logv2/log.h"
+#include "mongo/util/assert_util_core.h"
 
 namespace mongo {
 namespace log_detail {
@@ -166,8 +171,8 @@ bool hintMatchesClusterKey(const boost::optional<ClusteredCollectionInfo>& clust
 }
 
 /**
- * Returns the dependencies for the CanoncialQuery, split by those needed to answer the filter, and
- * those needed for "everything else" which is the project and sort.
+ * Returns the dependencies for the CanoncialQuery, split by those needed to answer the filter,
+ * and those needed for "everything else" which is the project and sort.
  */
 std::pair<DepsTracker, DepsTracker> computeDeps(const CanonicalQuery& query) {
     DepsTracker filterDeps;
@@ -181,8 +186,8 @@ std::pair<DepsTracker, DepsTracker> computeDeps(const CanonicalQuery& query) {
     if (auto sortPattern = query.getSortPattern()) {
         sortPattern->addDependencies(&outputDeps);
     }
-    // There's no known way a sort would depend on the whole document, and we already verified that
-    // the projection doesn't depend on the whole document.
+    // There's no known way a sort would depend on the whole document, and we already verified
+    // that the projection doesn't depend on the whole document.
     tassert(6430503, "Unexpectedly required entire object", !outputDeps.needWholeDocument);
     return {std::move(filterDeps), std::move(outputDeps)};
 }
@@ -284,8 +289,8 @@ string optionString(size_t options) {
         ss << "DEFAULT ";
     }
     while (options) {
-        // The expression (x & (x - 1)) yields x with the lowest bit cleared.  Then the exclusive-or
-        // of the result with the original yields the lowest bit by itself.
+        // The expression (x & (x - 1)) yields x with the lowest bit cleared.  Then the
+        // exclusive-or of the result with the original yields the lowest bit by itself.
         size_t new_options = options & (options - 1);
         QueryPlannerParams::Options opt = QueryPlannerParams::Options(new_options ^ options);
         options = new_options;
@@ -473,12 +478,16 @@ std::unique_ptr<QuerySolution> buildCollscanSoln(const CanonicalQuery& query,
     return QueryPlannerAnalysis::analyzeDataAccess(query, params, std::move(solnRoot));
 }
 
-std::unique_ptr<QuerySolution> buildWholeIXSoln(const IndexEntry& index,
-                                                const CanonicalQuery& query,
-                                                const QueryPlannerParams& params,
-                                                int direction = 1) {
+std::unique_ptr<QuerySolution> buildWholeIXSoln(
+    const IndexEntry& index,
+    const CanonicalQuery& query,
+    const QueryPlannerParams& params,
+    const boost::optional<int>& direction = boost::none) {
+    tassert(6499400,
+            "Cannot pass both an explicit direction and a traversal preference",
+            !(direction.has_value() && params.traversalPreference));
     std::unique_ptr<QuerySolutionNode> solnRoot(
-        QueryPlannerAccess::scanWholeIndex(index, query, params, direction));
+        QueryPlannerAccess::scanWholeIndex(index, query, params, direction.value_or(1)));
     return QueryPlannerAnalysis::analyzeDataAccess(query, params, std::move(solnRoot));
 }
 
@@ -698,7 +707,8 @@ StatusWith<std::unique_ptr<QuerySolution>> QueryPlanner::planFromCache(
         return s;
     }
 
-    // The MatchExpression tree is in canonical order. We must order the nodes for access planning.
+    // The MatchExpression tree is in canonical order. We must order the nodes for access
+    // planning.
     prepareForAccessPlanning(clone.get());
 
     LOGV2_DEBUG(20965, 5, "Tagged tree", "tree"_attr = redact(clone->debugString()));
@@ -729,8 +739,8 @@ StatusWith<std::unique_ptr<QuerySolution>> QueryPlanner::planFromCache(
 
 StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
     const CanonicalQuery& query, const QueryPlannerParams& params) {
-    // It's a little silly to ask for a count and for owned data. This could indicate a bug earlier
-    // on.
+    // It's a little silly to ask for a count and for owned data. This could indicate a bug
+    // earlier on.
     tassert(5397500,
             "Count and owned data requested",
             !((params.options & QueryPlannerParams::IS_COUNT) &&
@@ -776,10 +786,10 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
         const BSONObj& hintObj = query.getFindCommandRequest().getHint();
         const auto naturalHint = hintObj[query_request_helper::kNaturalSortField];
         if (naturalHint || hintMatchesClusterKey(params.clusteredInfo, hintObj)) {
-            // The hint can be {$natural: +/-1}. If this happens, output a collscan. We expect any
-            // $natural sort to have been normalized to a $natural hint upstream. Additionally, if
-            // the hint matches the collection's cluster key, we also output a collscan utilizing
-            // the cluster key.
+            // The hint can be {$natural: +/-1}. If this happens, output a collscan. We expect
+            // any $natural sort to have been normalized to a $natural hint upstream.
+            // Additionally, if the hint matches the collection's cluster key, we also output a
+            // collscan utilizing the cluster key.
 
             if (naturalHint) {
                 // Perform validation specific to $natural.
@@ -800,8 +810,8 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
 
                 const auto clusterKey = params.clusteredInfo->getIndexSpec().getKey();
 
-                // Check if the query collator is compatible with the collection collator for the
-                // provided min and max values.
+                // Check if the query collator is compatible with the collection collator for
+                // the provided min and max values.
                 if ((!minObj.isEmpty() &&
                      !indexCompatibleMaxMin(minObj,
                                             query.getCollator(),
@@ -842,17 +852,17 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
         }
     }
 
-    // Hints require us to only consider the hinted index. If index filters in the query settings
-    // were used to override the allowed indices for planning, we should not use the hinted index
-    // requested in the query.
+    // Hints require us to only consider the hinted index. If index filters in the query
+    // settings were used to override the allowed indices for planning, we should not use the
+    // hinted index requested in the query.
     BSONObj hintedIndex;
     if (!params.indexFiltersApplied) {
         hintedIndex = query.getFindCommandRequest().getHint();
     }
 
-    // Either the list of indices passed in by the caller, or the list of indices filtered according
-    // to the hint. This list is later expanded in order to allow the planner to handle wildcard
-    // indexes.
+    // Either the list of indices passed in by the caller, or the list of indices filtered
+    // according to the hint. This list is later expanded in order to allow the planner to
+    // handle wildcard indexes.
     std::vector<IndexEntry> fullIndexList;
 
     // Will hold a copy of the index entry chosen by the hint.
@@ -892,7 +902,8 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
     } else {
         relevantIndices = fullIndexList;
 
-        // Relevant indices should only ever exceed a size of 1 when there is a hint in the case of
+        // Relevant indices should only ever exceed a size of 1 when there is a hint in the case
+        // of
         // $** index.
         if (relevantIndices.size() > 1) {
             for (auto&& entry : relevantIndices) {
@@ -927,13 +938,13 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
         invariant(*hintedIndexEntry == fullIndexList.front());
 
         // In order to be fully compatible, the min has to be less than the max according to the
-        // index key pattern ordering. The first step in verifying this is "finish" the min and max
-        // by replacing empty objects and stripping field names.
+        // index key pattern ordering. The first step in verifying this is "finish" the min and
+        // max by replacing empty objects and stripping field names.
         BSONObj finishedMinObj = finishMinObj(*hintedIndexEntry, minObj, maxObj);
         BSONObj finishedMaxObj = finishMaxObj(*hintedIndexEntry, minObj, maxObj);
 
-        // Now we have the final min and max. This index is only relevant for the min/max query if
-        // min < max.
+        // Now we have the final min and max. This index is only relevant for the min/max query
+        // if min < max.
         if (finishedMinObj.woCompare(finishedMaxObj, hintedIndexEntry->keyPattern, false) >= 0) {
             return Status(ErrorCodes::Error(51175),
                           "The value provided for min() does not come before the value provided "
@@ -1065,9 +1076,9 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
                         "About to build solntree from tagged tree",
                         "tree"_attr = redact(nextTaggedTree->debugString()));
 
-            // Store the plan cache index tree before calling prepareForAccessingPlanning(), so that
-            // the PlanCacheIndexTree has the same sort as the MatchExpression used to generate the
-            // plan cache key.
+            // Store the plan cache index tree before calling prepareForAccessingPlanning(), so
+            // that the PlanCacheIndexTree has the same sort as the MatchExpression used to
+            // generate the plan cache key.
             std::unique_ptr<MatchExpression> clone(nextTaggedTree->shallowClone());
             std::unique_ptr<PlanCacheIndexTree> cacheData;
             auto statusWithCacheData = cacheDataFromTaggedTree(clone.get(), relevantIndices);
@@ -1080,8 +1091,8 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
                 cacheData = std::move(statusWithCacheData.getValue());
             }
 
-            // We have already cached the tree in canonical order, so now we can order the nodes for
-            // access planning.
+            // We have already cached the tree in canonical order, so now we can order the nodes
+            // for access planning.
             prepareForAccessPlanning(nextTaggedTree.get());
 
             // This can fail if enumeration makes a mistake.
@@ -1130,7 +1141,8 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
 
     // An index was hinted. If there are any solutions, they use the hinted index.  If not, we
     // scan the entire index to provide results and output that as our plan.  This is the
-    // desired behavior when an index is hinted that is not relevant to the query. In the case that
+    // desired behavior when an index is hinted that is not relevant to the query. In the case
+    // that
     // $** index is hinted, we do not want this behavior.
     if (!hintedIndex.isEmpty() && relevantIndices.size() == 1) {
         if (out.size() > 0) {
@@ -1141,6 +1153,7 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
                 ErrorCodes::NoQueryExecutionPlans,
                 "$hint: refusing to build whole-index solution, because it's a wildcard index");
         }
+
         // Return hinted index solution if found.
         auto soln = buildWholeIXSoln(relevantIndices.front(), query, params);
         if (!soln) {
@@ -1173,8 +1186,9 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
         if (!usingIndexToSort) {
             for (size_t i = 0; i < fullIndexList.size(); ++i) {
                 const IndexEntry& index = fullIndexList[i];
-                // Only a regular index or the non-hashed prefix of a compound hashed index can be
-                // used to provide a sort. In addition, the index needs to be a non-sparse index.
+                // Only a regular index or the non-hashed prefix of a compound hashed index can
+                // be used to provide a sort. In addition, the index needs to be a non-sparse
+                // index.
                 //
                 // TODO: Sparse indexes can't normally provide a sort, because non-indexed
                 // documents could potentially be missing from the result set.  However, if the
@@ -1194,14 +1208,14 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
                     continue;
                 }
 
-                // If the index collation differs from the query collation, the index should not be
-                // used to provide a sort, because strings will be ordered incorrectly.
+                // If the index collation differs from the query collation, the index should not
+                // be used to provide a sort, because strings will be ordered incorrectly.
                 if (!CollatorInterface::collatorsMatch(index.collator, query.getCollator())) {
                     continue;
                 }
 
-                // Partial indexes can only be used to provide a sort only if the query predicate is
-                // compatible.
+                // Partial indexes can only be used to provide a sort only if the query
+                // predicate is compatible.
                 if (index.filterExpr && !expression::isSubsetOf(query.root(), index.filterExpr)) {
                     continue;
                 }
@@ -1260,10 +1274,10 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
                 if (direction != 0) {
                     auto soln = buildCollscanSoln(query, isTailable, params, direction);
                     if (soln) {
-                        LOGV2_DEBUG(
-                            6082401,
-                            5,
-                            "Planner: outputting soln that uses clustered index to provide sort");
+                        LOGV2_DEBUG(6082401,
+                                    5,
+                                    "Planner: outputting soln that uses clustered index to "
+                                    "provide sort");
                         SolutionCacheData* scd = new SolutionCacheData();
                         scd->solnType = SolutionCacheData::COLLSCAN_SOLN;
                         scd->wholeIXSolnDir = direction;
@@ -1276,8 +1290,8 @@ StatusWith<std::vector<std::unique_ptr<QuerySolution>>> QueryPlanner::plan(
         }
     }
 
-    // If a projection exists, there may be an index that allows for a covered plan, even if none
-    // were considered earlier.
+    // If a projection exists, there may be an index that allows for a covered plan, even if
+    // none were considered earlier.
     const auto projection = query.getProj();
     if (params.options & QueryPlannerParams::GENERATE_COVERED_IXSCANS && out.size() == 0 &&
         query.getQueryObj().isEmpty() && projection && !projection->requiresDocument()) {
diff --git a/src/mongo/db/query/query_planner_common.cpp b/src/mongo/db/query/query_planner_common.cpp
index 85534e6128c..a0527200205 100644
--- a/src/mongo/db/query/query_planner_common.cpp
+++ b/src/mongo/db/query/query_planner_common.cpp
@@ -34,13 +34,47 @@
 #include "mongo/base/exact_cast.h"
 #include "mongo/db/query/projection_ast_path_tracking_visitor.h"
 #include "mongo/db/query/query_planner_common.h"
+#include "mongo/db/query/query_solution.h"
+#include "mongo/db/query/stage_types.h"
 #include "mongo/db/query/tree_walker.h"
+#include "mongo/logv2/log.h"
 #include "mongo/logv2/redaction.h"
 #include "mongo/util/assert_util.h"
 
 namespace mongo {
 
-void QueryPlannerCommon::reverseScans(QuerySolutionNode* node) {
+bool QueryPlannerCommon::scanDirectionsEqual(QuerySolutionNode* node, int direction) {
+    StageType type = node->getType();
+
+    boost::optional<int> scanDir;
+    if (STAGE_IXSCAN == type) {
+        IndexScanNode* isn = static_cast<IndexScanNode*>(node);
+        scanDir = isn->direction;
+    } else if (STAGE_DISTINCT_SCAN == type) {
+        DistinctNode* dn = static_cast<DistinctNode*>(node);
+        scanDir = dn->direction;
+    } else if (STAGE_COLLSCAN == type) {
+        CollectionScanNode* collScan = static_cast<CollectionScanNode*>(node);
+        scanDir = collScan->direction;
+    } else {
+        // We shouldn't encounter a sort stage.
+        invariant(!isSortStageType(type));
+    }
+
+    // If we found something with a direction, and the direction doesn't match, we return false.
+    if (scanDir && scanDir != direction) {
+        return false;
+    }
+
+    for (size_t i = 0; i < node->children.size(); ++i) {
+        if (!scanDirectionsEqual(node->children[i], direction)) {
+            return false;
+        }
+    }
+    return true;
+}
+
+void QueryPlannerCommon::reverseScans(QuerySolutionNode* node, bool reverseCollScans) {
     StageType type = node->getType();
 
     if (STAGE_IXSCAN == type) {
@@ -70,6 +104,9 @@ void QueryPlannerCommon::reverseScans(QuerySolutionNode* node) {
         // reverse direction of comparison for merge
         MergeSortNode* msn = static_cast<MergeSortNode*>(node);
         msn->sort = reverseSortObj(msn->sort);
+    } else if (reverseCollScans && STAGE_COLLSCAN == type) {
+        CollectionScanNode* collScan = static_cast<CollectionScanNode*>(node);
+        collScan->direction *= -1;
     } else {
         // Reversing scans is done in order to determine whether or not we need to add an explicit
         // SORT stage. There shouldn't already be one present in the plan.
@@ -77,7 +114,7 @@ void QueryPlannerCommon::reverseScans(QuerySolutionNode* node) {
     }
 
     for (size_t i = 0; i < node->children.size(); ++i) {
-        reverseScans(node->children[i]);
+        reverseScans(node->children[i], reverseCollScans);
     }
 }
 
diff --git a/src/mongo/db/query/query_planner_common.h b/src/mongo/db/query/query_planner_common.h
index 3c3bb88936c..6d441155b54 100644
--- a/src/mongo/db/query/query_planner_common.h
+++ b/src/mongo/db/query/query_planner_common.h
@@ -79,10 +79,17 @@ public:
     }
 
     /**
+     * Traverses the tree rooted at 'node'. Tests scan directions recursively to see if they are
+     * equal to the given direction argument. Returns true if they are and false otherwise.
+     */
+    static bool scanDirectionsEqual(QuerySolutionNode* node, int direction);
+
+    /**
      * Traverses the tree rooted at 'node'.  For every STAGE_IXSCAN encountered, reverse
-     * the scan direction and index bounds.
+     * the scan direction and index bounds, unless reverseCollScans equals true, in which case
+     * STAGE_COLLSCAN is reversed as well.
      */
-    static void reverseScans(QuerySolutionNode* node);
+    static void reverseScans(QuerySolutionNode* node, bool reverseCollScans = false);
 
     /**
      * Extracts all field names for the sortKey meta-projection and stores them in the returned
diff --git a/src/mongo/db/query/query_planner_params.h b/src/mongo/db/query/query_planner_params.h
index c7ddb2c928a..c8542cda90e 100644
--- a/src/mongo/db/query/query_planner_params.h
+++ b/src/mongo/db/query/query_planner_params.h
@@ -57,9 +57,24 @@ struct SecondaryCollectionInfo {
     long long storageSizeBytes{0};
 };
 
+
+// This holds information about the internal traversal preference used for time series. If we choose
+// an index that involves fields we're interested in, we prefer a specific direction to avoid a
+// blocking sort.
+struct TraversalPreference {
+    // If we end up with an index that provides {sortPattern}, we prefer to scan it in direction
+    // {direction}.
+    BSONObj sortPattern;
+    int direction;
+    // Cluster key for the collection this query accesses (for time-series it's control.min.time).
+    // If a collection scan is chosen, this will be compared against the sortPattern to see if we
+    // can satisfy the traversal preference.
+    std::string clusterField;
+};
+
 struct QueryPlannerParams {
-    QueryPlannerParams()
-        : options(DEFAULT),
+    QueryPlannerParams(size_t options = DEFAULT)
+        : options(options),
           indexFiltersApplied(false),
           maxIndexedSolutions(internalQueryPlannerMaxIndexedSolutions.load()),
           clusteredCollectionCollator(nullptr) {}
@@ -172,6 +187,8 @@ struct QueryPlannerParams {
 
     // List of information about any secondary collections that can be executed against.
     std::map<NamespaceString, SecondaryCollectionInfo> secondaryCollectionsInfo;
+
+    boost::optional<TraversalPreference> traversalPreference = boost::none;
 };
 
 }  // namespace mongo