SERVER-50370 Implement dotted path sort semantics in SBE

17 files changed, 1201 insertions, 248 deletions

diff --git a/jstests/core/sort5.js b/jstests/core/sort5.js
index 88c48b3f66c..01a267bf6b6 100644
--- a/jstests/core/sort5.js
+++ b/jstests/core/sort5.js
@@ -1,8 +1,4 @@
 // test compound sorting
-// TODO SERVER-50370: remove sbe_incompatible tag
-// @tags: [
-//   sbe_incompatible,
-// ]
 var t = db.sort5;
 t.drop();
 
diff --git a/jstests/core/sort9.js b/jstests/core/sort9.js
index 10065564eb4..57496b40da1 100644
--- a/jstests/core/sort9.js
+++ b/jstests/core/sort9.js
@@ -1,8 +1,4 @@
 // Unindexed array sorting SERVER-2884
-// TODO SERVER-50370: remove sbe_incompatible tag
-// @tags: [
-//   sbe_incompatible,
-// ]
 
 t = db.jstests_sort9;
 t.drop();
diff --git a/jstests/core/sort_dotted_paths.js b/jstests/core/sort_dotted_paths.js
new file mode 100644
index 00000000000..8cd3bd497a3
--- /dev/null
+++ b/jstests/core/sort_dotted_paths.js
@@ -0,0 +1,289 @@
+/**
+ * Test sorting with dotted field paths.
+ */
+(function() {
+"use strict";
+
+const coll = db.sort_dotted_paths;
+coll.drop();
+
+// Basic tests to verify that sorting deals with undefined, null, missing fields, and nested arrays
+// as expected.
+assert.commandWorked(coll.insert([
+    {_id: 1, a: 1},
+    {_id: 2, a: undefined},
+    {_id: 3, a: null},
+    {_id: 4, a: {}},
+    {_id: 5, a: []},
+    {_id: 6, a: [1]},
+    {_id: 7, a: [[1]]},
+    {_id: 8},
+    {_id: 9, a: [undefined]}
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({a: 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 5}, {_id: 9}, {_id: 3}, {_id: 8}, {_id: 1}, {_id: 6}, {_id: 4}, {_id: 7}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({a: -1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 4}, {_id: 1}, {_id: 6}, {_id: 3}, {_id: 8}, {_id: 2}, {_id: 5}, {_id: 9}]);
+
+// Test out sort({"a.0":1}) on a collection of documents where field 'a' is a mix of different
+// types (arrays of varying size for some documents, non-arrays for other documents).
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.0": 1, _id: 1}).toArray(),
+    [{_id: 9}, {_id: 1}, {_id: 2}, {_id: 3}, {_id: 4}, {_id: 5}, {_id: 8}, {_id: 6}, {_id: 7}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.0": -1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 6}, {_id: 1}, {_id: 2}, {_id: 3}, {_id: 4}, {_id: 5}, {_id: 8}, {_id: 9}]);
+
+// Test out sort({a:1,b:1}) on a collection where a is an array for some documents and b is an array
+// for other documents.
+assert(coll.drop());
+assert.commandWorked(coll.insert([
+    {_id: 1, a: 1, b: [2]},
+    {_id: 2, a: 2, b: []},
+    {_id: 3, a: [], b: 4},
+    {_id: 4, a: [1], b: 4},
+    {_id: 5, a: [2, 3], b: 2},
+    {_id: 6, a: 2, b: [3, 5]},
+    {_id: 7, a: 1, b: [1, 3]},
+    {_id: 8, a: [1, 2], b: 3},
+    {_id: 9, a: 2, b: 3}
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({a: 1, b: 1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 7}, {_id: 1}, {_id: 8}, {_id: 4}, {_id: 2}, {_id: 5}, {_id: 6}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({a: 1, b: -1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 4}, {_id: 7}, {_id: 8}, {_id: 1}, {_id: 6}, {_id: 9}, {_id: 5}, {_id: 2}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({a: -1, b: 1, _id: 1}).toArray(),
+    [{_id: 5}, {_id: 2}, {_id: 6}, {_id: 8}, {_id: 9}, {_id: 7}, {_id: 1}, {_id: 4}, {_id: 3}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({a: -1, b: -1, _id: 1}).toArray(),
+    [{_id: 5}, {_id: 6}, {_id: 8}, {_id: 9}, {_id: 2}, {_id: 4}, {_id: 7}, {_id: 1}, {_id: 3}]);
+
+// Verify that sort({a:1,b:1}) fails with a "parallel arrays" error when there is at least one
+// document where both a and b are arrays.
+assert(coll.drop());
+assert.commandWorked(
+    coll.insert([{_id: 1, a: [], b: 1}, {_id: 2, a: 1, b: []}, {_id: 3, a: [], b: []}]));
+
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {a: 1, b: 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {_id: 1, a: 1, b: 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {a: 1, _id: 1, b: 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {a: 1, b: 1, _id: 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+
+// Verify that sort({a:1,b:1}) does not fail with a "parallel arrays" error when documents where
+// both a and b are arrays are filtered out.
+const filter1 = {
+    $or: [{a: {$not: {$type: "array"}}}, {b: {$not: {$type: "array"}}}]
+};
+const output1 = [{_id: 1, a: [], b: 1}, {_id: 2, a: 1, b: []}];
+assert.eq(coll.find(filter1).sort({a: 1, b: 1}).toArray(), output1);
+assert.eq(coll.find(filter1).sort({_id: 1, a: 1, b: 1}).toArray(), output1);
+assert.eq(coll.find(filter1).sort({a: 1, _id: 1, b: 1}).toArray(), output1);
+assert.eq(coll.find(filter1).sort({a: 1, b: 1, _id: 1}).toArray(), output1);
+
+// Basic tests for a sort pattern that contains a path of length 2.
+assert(coll.drop());
+assert.commandWorked(coll.insert([
+    {_id: 1, a: {b: 2}},
+    {_id: 2, a: {b: []}},
+    {_id: 3, a: {b: [2, 3]}},
+    {_id: 4, a: {c: 1}},
+    {_id: 5, a: []},
+    {_id: 6, a: [{b: [5]}]},
+    {_id: 7, a: [{b: [4, 5]}, {b: 7}]},
+    {_id: 8, a: [{b: [1, 3]}, {b: [2, 5]}]},
+    {_id: 9, a: [{b: []}, {b: [1, 3]}]}
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 9}, {_id: 4}, {_id: 5}, {_id: 8}, {_id: 1}, {_id: 3}, {_id: 7}, {_id: 6}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, _id: -1}).toArray(),
+    [{_id: 9}, {_id: 2}, {_id: 5}, {_id: 4}, {_id: 8}, {_id: 3}, {_id: 1}, {_id: 7}, {_id: 6}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 6}, {_id: 8}, {_id: 3}, {_id: 9}, {_id: 1}, {_id: 4}, {_id: 5}, {_id: 2}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, _id: -1}).toArray(),
+    [{_id: 7}, {_id: 8}, {_id: 6}, {_id: 9}, {_id: 3}, {_id: 1}, {_id: 5}, {_id: 4}, {_id: 2}]);
+
+// Basic tests for a sort pattern that contains two paths of length 2 with a common prefix.
+assert(coll.drop());
+assert.commandWorked(coll.insert([
+    {_id: 1, a: [{b: 4, c: 1}, {b: 1, c: 2}]},
+    {_id: 2, a: [{b: 2, c: 1}, {b: 4, c: 2}]},
+    {_id: 3, a: [{b: 6, c: 1}, {b: 9, c: 2}]},
+    {_id: 4, a: [{b: 4, c: 1}, {b: 5, c: 3}]},
+    {_id: 5, a: [{b: 2, c: 1}, {b: 7, c: 3}]},
+    {_id: 6, a: [{b: 5, c: 1}, {b: 3, c: 3}]},
+    {_id: 7, a: [{b: 7, c: 2}, {b: 2, c: 3}]},
+    {_id: 8, a: [{b: 3, c: 2}, {b: 8, c: 3}]},
+    {_id: 9, a: [{b: 8, c: 2}, {b: 6, c: 3}]},
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, "a.b": 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 5}, {_id: 1}, {_id: 4}, {_id: 6}, {_id: 3}, {_id: 8}, {_id: 7}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, "a.b": -1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 6}, {_id: 1}, {_id: 4}, {_id: 2}, {_id: 5}, {_id: 9}, {_id: 7}, {_id: 8}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, "a.b": 1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 6}, {_id: 4}, {_id: 9}, {_id: 5}, {_id: 8}, {_id: 1}, {_id: 2}, {_id: 3}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, "a.b": -1, _id: 1}).toArray(),
+    [{_id: 8}, {_id: 5}, {_id: 9}, {_id: 4}, {_id: 6}, {_id: 7}, {_id: 3}, {_id: 2}, {_id: 1}]);
+
+// Basic tests for a sort pattern that contains a path of length 2 and a path of length 1, where
+// one path is a prefix of the other path.
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, a: 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 5}, {_id: 1}, {_id: 4}, {_id: 6}, {_id: 3}, {_id: 8}, {_id: 7}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, a: -1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 6}, {_id: 1}, {_id: 4}, {_id: 2}, {_id: 5}, {_id: 9}, {_id: 7}, {_id: 8}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, a: 1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 6}, {_id: 4}, {_id: 9}, {_id: 5}, {_id: 8}, {_id: 1}, {_id: 2}, {_id: 3}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, a: -1, _id: 1}).toArray(),
+    [{_id: 8}, {_id: 5}, {_id: 9}, {_id: 4}, {_id: 6}, {_id: 7}, {_id: 3}, {_id: 2}, {_id: 1}]);
+
+// Verify that sort({"a.b":1,"a.c":1}) fails with a "parallel arrays" error when there is at least
+// one document where both a.b and a.c are arrays.
+assert(coll.drop());
+assert.commandWorked(coll.insert([{_id: 1, a: {b: [1, 2], c: [3, 4]}}]));
+
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {"a.b": 1, "a.c": 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+
+// Verify that sort({"a.b":1,"c.d":1}) fails with a "parallel arrays" error when there is at least
+// onw document where both a.b and c.d are arrays.
+assert(coll.drop());
+assert.commandWorked(coll.insert({a: {b: [1, 2]}, c: {d: [3, 4]}}));
+
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {"a.b": 1, "c.d": 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+
+// More tests for a sort pattern that contains two paths of length 2 with a common prefix.
+assert(coll.drop());
+assert.commandWorked(coll.insert([
+    {_id: 1, a: [{b: [4, 1], c: 1}, {b: [1, 5], c: 2}]},
+    {_id: 2, a: [{b: 2, c: [1, 3]}, {b: 4, c: [2, 4]}]},
+    {_id: 3, a: [{b: [6, 4], c: 1}, {b: [9, 7], c: 2}]},
+    {_id: 4, a: [{b: 4, c: [1, 2]}, {b: 5, c: [3, 2]}]},
+    {_id: 5, a: [{b: [2, 3], c: 1}, {b: [7, 6], c: 3}]},
+    {_id: 6, a: [{b: 5, c: []}, {b: 3, c: 3}]},
+    {_id: 7, a: [{b: [], c: 2}, {b: 2, c: 3}]},
+    {_id: 8, a: [{b: 3, c: [2]}, {b: 8, c: [3]}]},
+    {_id: 9, a: [{b: [8], c: 2}, {b: [6], c: 3}]},
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, "a.b": 1, _id: 1}).toArray(),
+    [{_id: 6}, {_id: 1}, {_id: 2}, {_id: 5}, {_id: 3}, {_id: 4}, {_id: 7}, {_id: 8}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, "a.b": -1, _id: 1}).toArray(),
+    [{_id: 6}, {_id: 3}, {_id: 1}, {_id: 4}, {_id: 5}, {_id: 2}, {_id: 9}, {_id: 8}, {_id: 7}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, "a.b": 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 7}, {_id: 6}, {_id: 4}, {_id: 5}, {_id: 9}, {_id: 8}, {_id: 1}, {_id: 3}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, "a.b": -1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 8}, {_id: 5}, {_id: 9}, {_id: 4}, {_id: 6}, {_id: 7}, {_id: 3}, {_id: 1}]);
+
+// More tests for a sort pattern that contains a path of length 2 and a path of length 1, where
+// one path is a prefix of the other path.
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, a: 1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 1}, {_id: 2}, {_id: 5}, {_id: 6}, {_id: 8}, {_id: 4}, {_id: 3}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, a: -1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 1}, {_id: 5}, {_id: 2}, {_id: 8}, {_id: 6}, {_id: 3}, {_id: 4}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, a: 1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 8}, {_id: 9}, {_id: 5}, {_id: 6}, {_id: 4}, {_id: 1}, {_id: 2}, {_id: 7}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, a: -1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 9}, {_id: 8}, {_id: 5}, {_id: 1}, {_id: 4}, {_id: 6}, {_id: 2}, {_id: 7}]);
+
+// Test out sort({"a.0.b":1}) on a collection of documents where field "a" and sub-field "b" are
+// a mix of different types.
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.0.b": 1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 1}, {_id: 2}, {_id: 5}, {_id: 8}, {_id: 3}, {_id: 4}, {_id: 6}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.0.b": 1, _id: -1}).toArray(),
+    [{_id: 7}, {_id: 1}, {_id: 5}, {_id: 2}, {_id: 8}, {_id: 4}, {_id: 3}, {_id: 6}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.0.b": -1, _id: 1}).toArray(),
+    [{_id: 9}, {_id: 3}, {_id: 6}, {_id: 1}, {_id: 4}, {_id: 5}, {_id: 8}, {_id: 2}, {_id: 7}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.0.b": -1, _id: -1}).toArray(),
+    [{_id: 9}, {_id: 3}, {_id: 6}, {_id: 4}, {_id: 1}, {_id: 8}, {_id: 5}, {_id: 2}, {_id: 7}]);
+
+// Tests for a sort pattern that contains two paths of length 2 that do not have a common prefix.
+assert(coll.drop());
+assert.commandWorked(coll.insert([
+    {_id: 1, a: {b: 1}, c: [{d: [4, 1]}, {d: [1, 5]}]},
+    {_id: 2, a: [{b: [1, 3]}, {b: [2, 4]}], c: {d: 2}},
+    {_id: 3, a: {b: 1}, c: [{d: [6, 4]}, {d: [9, 7]}]},
+    {_id: 4, a: [{b: [2]}, {b: [3, 2]}], c: {d: 4}},
+    {_id: 5, a: {b: 1}, c: [{d: [2, 3]}, {d: [7, 6]}]},
+    {_id: 6, a: [{b: []}, {b: 3}], c: {d: 4}},
+    {_id: 7, a: {b: 2}, c: [{d: []}, {d: 2}]},
+    {_id: 8, a: [{b: [2]}, {b: [3]}], c: {d: 3}},
+    {_id: 9, a: {b: 3}, c: [{d: [8]}, {d: [6]}]},
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, "c.d": 1, _id: 1}).toArray(),
+    [{_id: 6}, {_id: 1}, {_id: 2}, {_id: 5}, {_id: 3}, {_id: 7}, {_id: 8}, {_id: 4}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, "c.d": -1, _id: 1}).toArray(),
+    [{_id: 6}, {_id: 3}, {_id: 5}, {_id: 1}, {_id: 2}, {_id: 4}, {_id: 8}, {_id: 7}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, "c.d": 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 8}, {_id: 4}, {_id: 6}, {_id: 9}, {_id: 7}, {_id: 1}, {_id: 5}, {_id: 3}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, "c.d": -1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 9}, {_id: 4}, {_id: 6}, {_id: 8}, {_id: 7}, {_id: 3}, {_id: 5}, {_id: 1}]);
+
+// Basic tests for a sort pattern that contains a path of length 3.
+assert(coll.drop());
+assert.commandWorked(coll.insert([
+    {_id: 1, a: {b: {c: 2}}},
+    {_id: 2, a: {b: {c: []}}},
+    {_id: 3, a: [{b: []}, {b: {c: []}}]},
+    {_id: 4, a: [{b: {c: 1}}]},
+    {_id: 5, a: {b: [{c: 3}, {d: 1}]}},
+    {_id: 6, a: {b: [{c: [5]}]}},
+    {_id: 7, a: []},
+    {_id: 8, a: {b: [{c: [1, 3]}, {c: [2, 5]}]}},
+    {_id: 9, a: [{b: [{c: []}, {c: [1, 3]}]}]}
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b.c": 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 3}, {_id: 9}, {_id: 5}, {_id: 7}, {_id: 4}, {_id: 8}, {_id: 1}, {_id: 6}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b.c": 1, _id: -1}).toArray(),
+    [{_id: 9}, {_id: 3}, {_id: 2}, {_id: 7}, {_id: 5}, {_id: 8}, {_id: 4}, {_id: 1}, {_id: 6}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b.c": -1, _id: 1}).toArray(),
+    [{_id: 6}, {_id: 8}, {_id: 5}, {_id: 9}, {_id: 1}, {_id: 4}, {_id: 3}, {_id: 7}, {_id: 2}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b.c": -1, _id: -1}).toArray(),
+    [{_id: 8}, {_id: 6}, {_id: 9}, {_id: 5}, {_id: 1}, {_id: 4}, {_id: 7}, {_id: 3}, {_id: 2}]);
+})();
diff --git a/jstests/core/sort_dotted_paths_collation.js b/jstests/core/sort_dotted_paths_collation.js
new file mode 100644
index 00000000000..092e16b10d4
--- /dev/null
+++ b/jstests/core/sort_dotted_paths_collation.js
@@ -0,0 +1,276 @@
+/**
+ * Test sorting with dotted field paths using a case-insensitive collation.
+ *
+ * @tags: [
+ *   assumes_no_implicit_collection_creation_after_drop,
+ * ]
+ */
+(function() {
+"use strict";
+
+const coll = db.sort_dotted_paths_collation;
+coll.drop();
+
+// Create a collection with a collation that is case-insensitive
+const caseInsensitive = {
+    collation: {locale: "en_US", strength: 2}
+};
+assert.commandWorked(db.createCollection(coll.getName(), caseInsensitive));
+
+// Test out sort({a:1,b:1}) on a collection where a is an array for some documents and b is an array
+// for other documents.
+assert.commandWorked(coll.insert([
+    {_id: 1, a: "a", b: ["b"]},
+    {_id: 2, a: "b", b: []},
+    {_id: 3, a: [], b: "d"},
+    {_id: 4, a: ["a"], b: "d"},
+    {_id: 5, a: ["b", "C"], b: "b"},
+    {_id: 6, a: "b", b: ["C", "e"]},
+    {_id: 7, a: "a", b: ["a", "C"]},
+    {_id: 8, a: ["a", "b"], b: "C"},
+    {_id: 9, a: "b", b: "C"}
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({a: 1, b: 1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 7}, {_id: 1}, {_id: 8}, {_id: 4}, {_id: 2}, {_id: 5}, {_id: 6}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({a: 1, b: -1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 4}, {_id: 7}, {_id: 8}, {_id: 1}, {_id: 6}, {_id: 9}, {_id: 5}, {_id: 2}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({a: -1, b: 1, _id: 1}).toArray(),
+    [{_id: 5}, {_id: 2}, {_id: 6}, {_id: 8}, {_id: 9}, {_id: 7}, {_id: 1}, {_id: 4}, {_id: 3}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({a: -1, b: -1, _id: 1}).toArray(),
+    [{_id: 5}, {_id: 6}, {_id: 8}, {_id: 9}, {_id: 2}, {_id: 4}, {_id: 7}, {_id: 1}, {_id: 3}]);
+
+// Verify that sort({a:1,b:1}) fails with a "parallel arrays" error when there is at least one
+// document where both a and b are arrays.
+assert(coll.drop());
+assert.commandWorked(db.createCollection(coll.getName(), caseInsensitive));
+assert.commandWorked(
+    coll.insert([{_id: 1, a: [], b: "a"}, {_id: 2, a: "a", b: []}, {_id: 3, a: [], b: []}]));
+
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {a: 1, b: 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {_id: 1, a: 1, b: 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {a: 1, _id: 1, b: 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {a: 1, b: 1, _id: 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+
+// Verify that sort({a:1,b:1}) does not fail with a "parallel arrays" error when documents where
+// both a and b are arrays are filtered out.
+const filter1 = {
+    $or: [{a: {$not: {$type: "array"}}}, {b: {$not: {$type: "array"}}}]
+};
+const output1 = [{_id: 1, a: [], b: "a"}, {_id: 2, a: "a", b: []}];
+assert.eq(coll.find(filter1).sort({a: 1, b: 1}).toArray(), output1);
+assert.eq(coll.find(filter1).sort({_id: 1, a: 1, b: 1}).toArray(), output1);
+assert.eq(coll.find(filter1).sort({a: 1, _id: 1, b: 1}).toArray(), output1);
+assert.eq(coll.find(filter1).sort({a: 1, b: 1, _id: 1}).toArray(), output1);
+
+// Basic tests for a sort pattern that contains a path of length 2.
+assert(coll.drop());
+assert.commandWorked(db.createCollection(coll.getName(), caseInsensitive));
+assert.commandWorked(coll.insert([
+    {_id: 1, a: {b: "b"}},
+    {_id: 2, a: {b: []}},
+    {_id: 3, a: {b: ["b", "C"]}},
+    {_id: 4, a: {c: "a"}},
+    {_id: 5, a: []},
+    {_id: 6, a: [{b: ["e"]}]},
+    {_id: 7, a: [{b: ["d", "e"]}, {b: "G"}]},
+    {_id: 8, a: [{b: ["a", "C"]}, {b: ["b", "e"]}]},
+    {_id: 9, a: [{b: []}, {b: ["a", "C"]}]}
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 9}, {_id: 4}, {_id: 5}, {_id: 8}, {_id: 1}, {_id: 3}, {_id: 7}, {_id: 6}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, _id: -1}).toArray(),
+    [{_id: 9}, {_id: 2}, {_id: 5}, {_id: 4}, {_id: 8}, {_id: 3}, {_id: 1}, {_id: 7}, {_id: 6}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 6}, {_id: 8}, {_id: 3}, {_id: 9}, {_id: 1}, {_id: 4}, {_id: 5}, {_id: 2}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, _id: -1}).toArray(),
+    [{_id: 7}, {_id: 8}, {_id: 6}, {_id: 9}, {_id: 3}, {_id: 1}, {_id: 5}, {_id: 4}, {_id: 2}]);
+
+// Basic tests for a sort pattern that contains two paths of length 2 with a common prefix.
+assert(coll.drop());
+assert.commandWorked(db.createCollection(coll.getName(), caseInsensitive));
+assert.commandWorked(coll.insert([
+    {_id: 1, a: [{b: "d", c: "a"}, {b: "a", c: "b"}]},
+    {_id: 2, a: [{b: "b", c: "a"}, {b: "d", c: "b"}]},
+    {_id: 3, a: [{b: "F", c: "a"}, {b: "I", c: "b"}]},
+    {_id: 4, a: [{b: "d", c: "a"}, {b: "e", c: "C"}]},
+    {_id: 5, a: [{b: "b", c: "a"}, {b: "G", c: "C"}]},
+    {_id: 6, a: [{b: "e", c: "a"}, {b: "C", c: "C"}]},
+    {_id: 7, a: [{b: "G", c: "b"}, {b: "b", c: "C"}]},
+    {_id: 8, a: [{b: "C", c: "b"}, {b: "h", c: "C"}]},
+    {_id: 9, a: [{b: "h", c: "b"}, {b: "F", c: "C"}]},
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, "a.b": 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 5}, {_id: 1}, {_id: 4}, {_id: 6}, {_id: 3}, {_id: 8}, {_id: 7}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, "a.b": -1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 6}, {_id: 1}, {_id: 4}, {_id: 2}, {_id: 5}, {_id: 9}, {_id: 7}, {_id: 8}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, "a.b": 1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 6}, {_id: 4}, {_id: 9}, {_id: 5}, {_id: 8}, {_id: 1}, {_id: 2}, {_id: 3}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, "a.b": -1, _id: 1}).toArray(),
+    [{_id: 8}, {_id: 5}, {_id: 9}, {_id: 4}, {_id: 6}, {_id: 7}, {_id: 3}, {_id: 2}, {_id: 1}]);
+
+// Basic tests for a sort pattern that contains a path of length 2 and a path of length 1, where
+// one path is a prefix of the other path.
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, a: 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 5}, {_id: 1}, {_id: 4}, {_id: 6}, {_id: 3}, {_id: 8}, {_id: 7}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, a: -1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 6}, {_id: 1}, {_id: 4}, {_id: 2}, {_id: 5}, {_id: 9}, {_id: 7}, {_id: 8}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, a: 1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 6}, {_id: 4}, {_id: 9}, {_id: 5}, {_id: 8}, {_id: 1}, {_id: 2}, {_id: 3}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, a: -1, _id: 1}).toArray(),
+    [{_id: 8}, {_id: 5}, {_id: 9}, {_id: 4}, {_id: 6}, {_id: 7}, {_id: 3}, {_id: 2}, {_id: 1}]);
+
+// Verify that sort({"a.b":1,"a.c":1}) fails with a "parallel arrays" error when there is at least
+// one document where both a.b and a.c are arrays.
+assert(coll.drop());
+assert.commandWorked(db.createCollection(coll.getName(), caseInsensitive));
+assert.commandWorked(coll.insert([{_id: 1, a: {b: ["a", "b"], c: ["C", "d"]}}]));
+
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {"a.b": 1, "a.c": 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+
+// Verify that sort({"a.b":1,"c.d":1}) fails with a "parallel arrays" error when there is at least
+// onw document where both a.b and c.d are arrays.
+assert(coll.drop());
+assert.commandWorked(db.createCollection(coll.getName(), caseInsensitive));
+assert.commandWorked(coll.insert({a: {b: ["a", "b"]}, c: {d: ["C", "d"]}}));
+
+assert.commandFailedWithCode(db.runCommand({find: coll.getName(), sort: {"a.b": 1, "c.d": 1}}),
+                             [ErrorCodes.BadValue, ErrorCodes.CannotIndexParallelArrays]);
+
+// More tests for a sort pattern that contains two paths of length 2 with a common prefix.
+assert(coll.drop());
+assert.commandWorked(db.createCollection(coll.getName(), caseInsensitive));
+assert.commandWorked(coll.insert([
+    {_id: 1, a: [{b: ["d", "a"], c: "a"}, {b: ["a", "e"], c: "b"}]},
+    {_id: 2, a: [{b: "b", c: ["a", "C"]}, {b: "d", c: ["b", "d"]}]},
+    {_id: 3, a: [{b: ["F", "d"], c: "a"}, {b: ["I", "G"], c: "b"}]},
+    {_id: 4, a: [{b: "d", c: ["a", "b"]}, {b: "e", c: ["C", "b"]}]},
+    {_id: 5, a: [{b: ["b", "C"], c: "a"}, {b: ["G", "F"], c: "C"}]},
+    {_id: 6, a: [{b: "e", c: []}, {b: "C", c: "C"}]},
+    {_id: 7, a: [{b: [], c: "b"}, {b: "b", c: "C"}]},
+    {_id: 8, a: [{b: "C", c: ["b"]}, {b: "h", c: ["C"]}]},
+    {_id: 9, a: [{b: ["h"], c: "b"}, {b: ["F"], c: "C"}]},
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, "a.b": 1, _id: 1}).toArray(),
+    [{_id: 6}, {_id: 1}, {_id: 2}, {_id: 5}, {_id: 3}, {_id: 4}, {_id: 7}, {_id: 8}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": 1, "a.b": -1, _id: 1}).toArray(),
+    [{_id: 6}, {_id: 3}, {_id: 1}, {_id: 4}, {_id: 5}, {_id: 2}, {_id: 9}, {_id: 8}, {_id: 7}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, "a.b": 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 7}, {_id: 6}, {_id: 4}, {_id: 5}, {_id: 9}, {_id: 8}, {_id: 1}, {_id: 3}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.c": -1, "a.b": -1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 8}, {_id: 5}, {_id: 9}, {_id: 4}, {_id: 6}, {_id: 7}, {_id: 3}, {_id: 1}]);
+
+// More tests for a sort pattern that contains a path of length 2 and a path of length 1, where
+// one path is a prefix of the other path.
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, a: 1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 1}, {_id: 2}, {_id: 5}, {_id: 6}, {_id: 8}, {_id: 4}, {_id: 3}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, a: -1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 1}, {_id: 5}, {_id: 2}, {_id: 8}, {_id: 6}, {_id: 3}, {_id: 4}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, a: 1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 8}, {_id: 9}, {_id: 5}, {_id: 6}, {_id: 4}, {_id: 1}, {_id: 2}, {_id: 7}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, a: -1, _id: 1}).toArray(),
+    [{_id: 3}, {_id: 9}, {_id: 8}, {_id: 5}, {_id: 1}, {_id: 4}, {_id: 6}, {_id: 2}, {_id: 7}]);
+
+// Test out sort({"a.0.b":1}) on a collection of documents where field "a" and sub-field "b" are
+// a mix of different types.
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.0.b": 1, _id: 1}).toArray(),
+    [{_id: 7}, {_id: 1}, {_id: 2}, {_id: 5}, {_id: 8}, {_id: 3}, {_id: 4}, {_id: 6}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.0.b": 1, _id: -1}).toArray(),
+    [{_id: 7}, {_id: 1}, {_id: 5}, {_id: 2}, {_id: 8}, {_id: 4}, {_id: 3}, {_id: 6}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.0.b": -1, _id: 1}).toArray(),
+    [{_id: 9}, {_id: 3}, {_id: 6}, {_id: 1}, {_id: 4}, {_id: 5}, {_id: 8}, {_id: 2}, {_id: 7}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.0.b": -1, _id: -1}).toArray(),
+    [{_id: 9}, {_id: 3}, {_id: 6}, {_id: 4}, {_id: 1}, {_id: 8}, {_id: 5}, {_id: 2}, {_id: 7}]);
+
+// Tests for a sort pattern that contains two paths of length 2 that do not have a common prefix.
+assert(coll.drop());
+assert.commandWorked(db.createCollection(coll.getName(), caseInsensitive));
+assert.commandWorked(coll.insert([
+    {_id: 1, a: {b: "a"}, c: [{d: ["d", "a"]}, {d: ["a", "e"]}]},
+    {_id: 2, a: [{b: ["a", "C"]}, {b: ["b", "d"]}], c: {d: "b"}},
+    {_id: 3, a: {b: "a"}, c: [{d: ["F", "d"]}, {d: ["I", "G"]}]},
+    {_id: 4, a: [{b: ["b"]}, {b: ["C", "b"]}], c: {d: "d"}},
+    {_id: 5, a: {b: "a"}, c: [{d: ["b", "C"]}, {d: ["G", "F"]}]},
+    {_id: 6, a: [{b: []}, {b: "C"}], c: {d: "d"}},
+    {_id: 7, a: {b: "b"}, c: [{d: []}, {d: "b"}]},
+    {_id: 8, a: [{b: ["b"]}, {b: ["C"]}], c: {d: "C"}},
+    {_id: 9, a: {b: "C"}, c: [{d: ["h"]}, {d: ["F"]}]},
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, "c.d": 1, _id: 1}).toArray(),
+    [{_id: 6}, {_id: 1}, {_id: 2}, {_id: 5}, {_id: 3}, {_id: 7}, {_id: 8}, {_id: 4}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": 1, "c.d": -1, _id: 1}).toArray(),
+    [{_id: 6}, {_id: 3}, {_id: 5}, {_id: 1}, {_id: 2}, {_id: 4}, {_id: 8}, {_id: 7}, {_id: 9}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, "c.d": 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 8}, {_id: 4}, {_id: 6}, {_id: 9}, {_id: 7}, {_id: 1}, {_id: 5}, {_id: 3}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b": -1, "c.d": -1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 9}, {_id: 4}, {_id: 6}, {_id: 8}, {_id: 7}, {_id: 3}, {_id: 5}, {_id: 1}]);
+
+// Basic tests for a sort pattern that contains a path of length 3.
+assert(coll.drop());
+assert.commandWorked(db.createCollection(coll.getName(), caseInsensitive));
+assert.commandWorked(coll.insert([
+    {_id: 1, a: {b: {c: "b"}}},
+    {_id: 2, a: {b: {c: []}}},
+    {_id: 3, a: [{b: []}, {b: {c: []}}]},
+    {_id: 4, a: [{b: {c: "a"}}]},
+    {_id: 5, a: {b: [{c: "C"}, {d: "a"}]}},
+    {_id: 6, a: {b: [{c: ["e"]}]}},
+    {_id: 7, a: []},
+    {_id: 8, a: {b: [{c: ["a", "C"]}, {c: ["b", "e"]}]}},
+    {_id: 9, a: [{b: [{c: []}, {c: ["a", "C"]}]}]}
+]));
+
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b.c": 1, _id: 1}).toArray(),
+    [{_id: 2}, {_id: 3}, {_id: 9}, {_id: 5}, {_id: 7}, {_id: 4}, {_id: 8}, {_id: 1}, {_id: 6}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b.c": 1, _id: -1}).toArray(),
+    [{_id: 9}, {_id: 3}, {_id: 2}, {_id: 7}, {_id: 5}, {_id: 8}, {_id: 4}, {_id: 1}, {_id: 6}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b.c": -1, _id: 1}).toArray(),
+    [{_id: 6}, {_id: 8}, {_id: 5}, {_id: 9}, {_id: 1}, {_id: 4}, {_id: 3}, {_id: 7}, {_id: 2}]);
+assert.eq(
+    coll.find({}, {_id: 1}).sort({"a.b.c": -1, _id: -1}).toArray(),
+    [{_id: 8}, {_id: 6}, {_id: 9}, {_id: 5}, {_id: 1}, {_id: 4}, {_id: 7}, {_id: 3}, {_id: 2}]);
+})();
diff --git a/src/mongo/db/exec/sbe/SConscript b/src/mongo/db/exec/sbe/SConscript
index 0bc3157699e..29de1ab4348 100644
--- a/src/mongo/db/exec/sbe/SConscript
+++ b/src/mongo/db/exec/sbe/SConscript
@@ -21,6 +21,7 @@ env.Library(
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/fts/base_fts',
+        '$BUILD_DIR/mongo/db/index/key_generator',
         '$BUILD_DIR/mongo/db/query/collation/collator_interface',
         '$BUILD_DIR/mongo/db/query/datetime/date_time_support',
         '$BUILD_DIR/mongo/db/storage/key_string',
diff --git a/src/mongo/db/exec/sbe/expressions/expression.cpp b/src/mongo/db/exec/sbe/expressions/expression.cpp
index 939e83e0943..bf50e71ce79 100644
--- a/src/mongo/db/exec/sbe/expressions/expression.cpp
+++ b/src/mongo/db/exec/sbe/expressions/expression.cpp
@@ -451,6 +451,8 @@ static stdx::unordered_map<std::string, BuiltinFn> kBuiltinFunctions = {
     {"dateAdd", BuiltinFn{[](size_t n) { return n == 5; }, vm::Builtin::dateAdd, false}},
     {"hasNullBytes", BuiltinFn{[](size_t n) { return n == 1; }, vm::Builtin::hasNullBytes, false}},
     {"ftsMatch", BuiltinFn{[](size_t n) { return n == 2; }, vm::Builtin::ftsMatch, false}},
+    {"generateSortKey",
+     BuiltinFn{[](size_t n) { return n == 2; }, vm::Builtin::generateSortKey, false}},
 };
 
 /**
diff --git a/src/mongo/db/exec/sbe/values/sort_spec.h b/src/mongo/db/exec/sbe/values/sort_spec.h
new file mode 100644
index 00000000000..0082b6b5bf1
--- /dev/null
+++ b/src/mongo/db/exec/sbe/values/sort_spec.h
@@ -0,0 +1,67 @@
+/**
+ *    Copyright (C) 2021-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/exec/sbe/values/value.h"
+#include "mongo/db/index/btree_key_generator.h"
+
+namespace mongo::sbe::value {
+/**
+ * SortSpec is a wrapper around a BSONObj giving a sort pattern (encoded as a BSONObj), a collator,
+ * and a BtreeKeyGenerator object.
+ */
+class SortSpec {
+public:
+    SortSpec(const BSONObj& sortPattern, const CollatorInterface* collator)
+        : _sortPattern(sortPattern.getOwned()), _collator(collator), _keyGen(initKeyGen()) {}
+
+    SortSpec(const SortSpec& other)
+        : _sortPattern(other._sortPattern), _collator(other._collator), _keyGen(initKeyGen()) {}
+
+    SortSpec& operator=(const SortSpec&) = delete;
+
+    KeyString::Value generateSortKey(const BSONObj& obj) const;
+
+    const BSONObj& getPattern() const {
+        return _sortPattern;
+    }
+
+    const CollatorInterface* getCollator() const {
+        return _collator;
+    }
+
+private:
+    BtreeKeyGenerator initKeyGen() const;
+
+    const BSONObj _sortPattern;
+    const CollatorInterface* const _collator;
+    const BtreeKeyGenerator _keyGen;
+};
+}  // namespace mongo::sbe::value
diff --git a/src/mongo/db/exec/sbe/values/value.cpp b/src/mongo/db/exec/sbe/values/value.cpp
index 2221709d460..1efffdf9092 100644
--- a/src/mongo/db/exec/sbe/values/value.cpp
+++ b/src/mongo/db/exec/sbe/values/value.cpp
@@ -34,6 +34,7 @@
 #include "mongo/base/compare_numbers.h"
 #include "mongo/db/exec/js_function.h"
 #include "mongo/db/exec/sbe/values/bson.h"
+#include "mongo/db/exec/sbe/values/sort_spec.h"
 #include "mongo/db/exec/sbe/values/value_builder.h"
 #include "mongo/db/query/collation/collator_interface.h"
 #include "mongo/db/query/datetime/date_time_support.h"
@@ -138,14 +139,54 @@ size_t PcreRegex::getNumberCaptures() const {
     return static_cast<size_t>(numCaptures);
 }
 
+KeyString::Value SortSpec::generateSortKey(const BSONObj& obj) const {
+    KeyStringSet keySet;
+    SharedBufferFragmentBuilder allocator(KeyString::HeapBuilder::kHeapAllocatorDefaultBytes);
+    const bool skipMultikey = false;
+    MultikeyPaths* multikeyPaths = nullptr;
+    _keyGen.getKeys(allocator, obj, skipMultikey, &keySet, multikeyPaths);
+
+    // When 'isSparse' is false, BtreeKeyGenerator::getKeys() is guaranteed to insert at least
+    // one key into 'keySet', so this assertion should always be true.
+    tassert(5037000, "BtreeKeyGenerator failed to generate key", !keySet.empty());
+
+    // Return the first KeyString in the set.
+    return std::move(*keySet.extract_sequence().begin());
+}
+
+BtreeKeyGenerator SortSpec::initKeyGen() const {
+    tassert(
+        5037003, "SortSpec should not be passed an empty sort pattern", !_sortPattern.isEmpty());
+
+    std::vector<const char*> fields;
+    std::vector<BSONElement> fixed;
+    for (auto&& elem : _sortPattern) {
+        fields.push_back(elem.fieldName());
+
+        // BtreeKeyGenerator's constructor's first parameter (the 'fields' vector) and second
+        // parameter (the 'fixed' vector) are parallel vectors. The 'fixed' vector allows the
+        // caller to specify if the any sort keys have already been determined for one or more
+        // of the field paths from the 'fields' vector. In this case, we haven't determined what
+        // the sort keys are for any of the fields paths, so we populate the 'fixed' vector with
+        // EOO values to indicate this.
+        fixed.emplace_back();
+    }
+
+    const bool isSparse = false;
+    auto version = KeyString::Version::kLatestVersion;
+    auto ordering = Ordering::make(_sortPattern);
+
+    return {std::move(fields), std::move(fixed), isSparse, _collator, version, ordering};
+}
+
 std::pair<TypeTags, Value> makeCopyJsFunction(const JsFunction& jsFunction) {
     auto ownedJsFunction = bitcastFrom<JsFunction*>(new JsFunction(jsFunction));
     return {TypeTags::jsFunction, ownedJsFunction};
 }
 
 std::pair<TypeTags, Value> makeCopyShardFilterer(const ShardFilterer& filterer) {
-    auto filter = bitcastFrom<ShardFilterer*>(filterer.clone().release());
-    return {TypeTags::shardFilterer, filter};
+    auto filtererCopy = bitcastFrom<ShardFilterer*>(filterer.clone().release());
+    return {TypeTags::shardFilterer, filtererCopy};
 }
 
 std::pair<TypeTags, Value> makeCopyFtsMatcher(const fts::FTSMatcher& matcher) {
@@ -153,6 +194,11 @@ std::pair<TypeTags, Value> makeCopyFtsMatcher(const fts::FTSMatcher& matcher) {
     return {TypeTags::ftsMatcher, copy};
 }
 
+std::pair<TypeTags, Value> makeCopySortSpec(const SortSpec& ss) {
+    auto ssCopy = bitcastFrom<SortSpec*>(new SortSpec(ss));
+    return {TypeTags::sortSpec, ssCopy};
+}
+
 void releaseValue(TypeTags tag, Value val) noexcept {
     switch (tag) {
         case TypeTags::NumberDecimal:
@@ -200,6 +246,9 @@ void releaseValue(TypeTags tag, Value val) noexcept {
         case TypeTags::ftsMatcher:
             delete getFtsMatcherView(val);
             break;
+        case TypeTags::sortSpec:
+            delete getSortSpecView(val);
+            break;
         default:
             break;
     }
@@ -299,7 +348,7 @@ void writeTagToStream(T& stream, const TypeTags tag) {
             stream << "shardFilterer";
             break;
         case TypeTags::collator:
-            stream << "Collator";
+            stream << "collator";
             break;
         case TypeTags::bsonRegex:
             stream << "bsonRegex";
@@ -313,12 +362,91 @@ void writeTagToStream(T& stream, const TypeTags tag) {
         case TypeTags::ftsMatcher:
             stream << "ftsMatcher";
             break;
+        case TypeTags::sortSpec:
+            stream << "sortSpec";
+            break;
         default:
             stream << "unknown tag";
             break;
     }
 }
 
+namespace {
+template <typename T>
+void writeStringDataToStream(T& stream, StringData sd) {
+    stream << '"';
+    if (sd.size() <= kStringMaxDisplayLength) {
+        stream << sd << '"';
+    } else {
+        stream << sd.substr(0, kStringMaxDisplayLength) << "\"...";
+    }
+}
+
+template <typename T>
+void writeArrayToStream(T& stream, TypeTags tag, Value val) {
+    stream << '[';
+    if (auto ae = ArrayEnumerator{tag, val}; !ae.atEnd()) {
+        for (;;) {
+            auto [tag, val] = ae.getViewOfValue();
+            writeValueToStream(stream, tag, val);
+
+            ae.advance();
+            if (ae.atEnd()) {
+                break;
+            }
+
+            stream << ", ";
+        }
+    }
+    stream << ']';
+}
+
+template <typename T>
+void writeObjectToStream(T& stream, TypeTags tag, Value val) {
+    stream << '{';
+    if (auto oe = ObjectEnumerator{tag, val}; !oe.atEnd()) {
+        for (;;) {
+            stream << "\"" << oe.getFieldName() << "\" : ";
+            auto [tag, val] = oe.getViewOfValue();
+            writeValueToStream(stream, tag, val);
+
+            oe.advance();
+            if (oe.atEnd()) {
+                break;
+            }
+
+            stream << ", ";
+        }
+    }
+    stream << '}';
+}
+
+template <typename T>
+void writeObjectToStream(T& stream, const BSONObj& obj) {
+    writeObjectToStream(stream, TypeTags::bsonObject, bitcastFrom<const char*>(obj.objdata()));
+}
+
+template <typename T>
+void writeObjectIdToStream(T& stream, TypeTags tag, Value val) {
+    auto objId =
+        tag == TypeTags::ObjectId ? getObjectIdView(val)->data() : bitcastTo<uint8_t*>(val);
+
+    stream << (tag == TypeTags::ObjectId ? "ObjectId(\"" : "bsonObjectId(\"")
+           << OID::from(objId).toString() << "\")";
+}
+
+template <typename T>
+void writeCollatorToStream(T& stream, const CollatorInterface* collator) {
+    if (collator) {
+        stream << "Collator(";
+        writeObjectToStream(stream, collator->getSpec().toBSON());
+        stream << ')';
+    } else {
+        stream << "null";
+    }
+}
+}  // namespace
+
 template <typename T>
 void writeValueToStream(T& stream, TypeTags tag, Value val) {
     switch (tag) {
@@ -343,53 +471,29 @@ void writeValueToStream(T& stream, TypeTags tag, Value val) {
         case TypeTags::Null:
             stream << "null";
             break;
-        case TypeTags::Array: {
-            auto arr = getArrayView(val);
-            stream << '[';
-            for (size_t idx = 0; idx < arr->size(); ++idx) {
-                if (idx != 0) {
-                    stream << ", ";
-                }
-                auto [tag, val] = arr->getAt(idx);
-                writeValueToStream(stream, tag, val);
-            }
-            stream << ']';
+        case TypeTags::StringSmall:
+        case TypeTags::StringBig:
+        case TypeTags::bsonString:
+            writeStringDataToStream(stream, getStringOrSymbolView(tag, val));
             break;
-        }
-        case TypeTags::ArraySet: {
-            auto arr = getArraySetView(val);
-            stream << '[';
-            bool first = true;
-            for (const auto& v : arr->values()) {
-                if (!first) {
-                    stream << ", ";
-                }
-                first = false;
-                writeValueToStream(stream, v.first, v.second);
-            }
-            stream << ']';
+        case TypeTags::bsonSymbol:
+            stream << "Symbol(";
+            writeStringDataToStream(stream, getStringOrSymbolView(tag, val));
+            stream << ')';
             break;
-        }
-        case TypeTags::Object: {
-            auto obj = getObjectView(val);
-            stream << '{';
-            for (size_t idx = 0; idx < obj->size(); ++idx) {
-                if (idx != 0) {
-                    stream << ", ";
-                }
-                stream << '"' << obj->field(idx) << '"';
-                stream << " : ";
-                auto [tag, val] = obj->getAt(idx);
-                writeValueToStream(stream, tag, val);
-            }
-            stream << '}';
+        case TypeTags::Array:
+        case TypeTags::ArraySet:
+        case TypeTags::bsonArray:
+            writeArrayToStream(stream, tag, val);
             break;
-        }
-        case TypeTags::ObjectId: {
-            auto objId = getObjectIdView(val);
-            stream << "ObjectId(\"" << OID::from(objId->data()).toString() << "\")";
+        case TypeTags::Object:
+        case TypeTags::bsonObject:
+            writeObjectToStream(stream, tag, val);
+            break;
+        case TypeTags::ObjectId:
+        case TypeTags::bsonObjectId:
+            writeObjectIdToStream(stream, tag, val);
             break;
-        }
         case TypeTags::Nothing:
             stream << "Nothing";
             break;
@@ -399,77 +503,6 @@ void writeValueToStream(T& stream, TypeTags tag, Value val) {
         case TypeTags::MaxKey:
             stream << "maxKey";
             break;
-        case TypeTags::bsonArray: {
-            const char* be = getRawPointerView(val);
-            const char* end = be + ConstDataView(be).read<LittleEndian<uint32_t>>();
-            bool first = true;
-            // Skip document length.
-            be += 4;
-            stream << '[';
-            while (*be != 0) {
-                auto sv = bson::fieldNameView(be);
-
-                if (!first) {
-                    stream << ", ";
-                }
-                first = false;
-
-                auto [tag, val] = bson::convertFrom(true, be, end, sv.size());
-                writeValueToStream(stream, tag, val);
-
-                be = bson::advance(be, sv.size());
-            }
-            stream << ']';
-            break;
-        }
-        case TypeTags::bsonObject: {
-            const char* be = getRawPointerView(val);
-            const char* end = be + ConstDataView(be).read<LittleEndian<uint32_t>>();
-            bool first = true;
-            // Skip document length.
-            be += 4;
-            stream << '{';
-            while (*be != 0) {
-                auto sv = bson::fieldNameView(be);
-
-                if (!first) {
-                    stream << ", ";
-                }
-                first = false;
-
-                stream << '"' << std::string(sv) << '"';
-                stream << " : ";
-                auto [tag, val] = bson::convertFrom(true, be, end, sv.size());
-                writeValueToStream(stream, tag, val);
-
-                be = bson::advance(be, sv.size());
-            }
-            stream << '}';
-            break;
-        }
-        case TypeTags::StringSmall:
-        case TypeTags::StringBig:
-        case TypeTags::bsonString:
-        case TypeTags::bsonSymbol: {
-            auto sv = getStringOrSymbolView(tag, val);
-
-            stream << (tag == TypeTags::bsonSymbol ? "Symbol(\"" : "\"");
-
-            if (sv.size() <= kStringMaxDisplayLength) {
-                stream << sv << "\"";
-            } else {
-                stream << sv.substr(0, kStringMaxDisplayLength) << "\"...";
-            }
-
-            stream << (tag == TypeTags::bsonSymbol ? ")" : "");
-
-            break;
-        }
-        case TypeTags::bsonObjectId: {
-            auto objId = getRawPointerView(val);
-            stream << "bsonObjectId(\"" << OID::from(objId).toString() << "\")";
-            break;
-        }
         case TypeTags::bsonBinData: {
             auto data =
                 reinterpret_cast<const char*>(getBSONBinDataCompat(TypeTags::bsonBinData, val));
@@ -488,12 +521,10 @@ void writeValueToStream(T& stream, TypeTags tag, Value val) {
                                                             hexblob::encodeLower(sd.substr(10, 6)));
                 break;
             }
-            stream << "BinData(" << type << ", ";
-            if (len > kBinDataMaxDisplayLength) {
-                stream << hexblob::encode(data, kBinDataMaxDisplayLength) << "...)";
-            } else {
-                stream << hexblob::encode(data, len) << ")";
-            }
+
+            stream << "BinData(" << type << ", "
+                   << hexblob::encode(data, std::min(len, kBinDataMaxDisplayLength))
+                   << (len > kBinDataMaxDisplayLength ? "...)" : ")");
             break;
         }
         case TypeTags::bsonUndefined:
@@ -517,7 +548,7 @@ void writeValueToStream(T& stream, TypeTags tag, Value val) {
         case TypeTags::timeZoneDB: {
             auto tzdb = getTimeZoneDBView(val);
             auto timeZones = tzdb->getTimeZoneStrings();
-            stream << "TimeZoneDatabase(" + timeZones.front() + "..." + timeZones.back() + ")";
+            stream << "TimeZoneDatabase(" << timeZones.front() << "..." << timeZones.back() + ")";
             break;
         }
         case TypeTags::RecordId:
@@ -531,7 +562,7 @@ void writeValueToStream(T& stream, TypeTags tag, Value val) {
             stream << "ShardFilterer";
             break;
         case TypeTags::collator:
-            stream << "Collator(" << getCollatorView(val)->getSpec().toBSON().toString() << ")";
+            writeCollatorToStream(stream, getCollatorView(val));
             break;
         case TypeTags::bsonRegex: {
             const auto regex = getBsonRegexView(val);
@@ -543,15 +574,28 @@ void writeValueToStream(T& stream, TypeTags tag, Value val) {
             break;
         case TypeTags::bsonDBPointer: {
             const auto dbptr = getBsonDBPointerView(val);
-            stream << "DBPointer(\"" << dbptr.ns << "\", \"" << OID::from(dbptr.id).toString()
-                   << "\")";
+            stream << "DBPointer(";
+            writeStringDataToStream(stream, dbptr.ns);
+            stream << ", ";
+            writeObjectIdToStream(
+                stream, TypeTags::bsonObjectId, bitcastFrom<const uint8_t*>(dbptr.id));
+            stream << ')';
             break;
         }
         case value::TypeTags::ftsMatcher: {
             auto ftsMatcher = getFtsMatcherView(val);
-            stream << "FtsMatcher(" << ftsMatcher->query().toBSON().toString() << ")";
+            stream << "FtsMatcher(";
+            writeObjectToStream(stream, ftsMatcher->query().toBSON());
+            stream << ')';
             break;
         }
+        case TypeTags::sortSpec:
+            stream << "SortSpec(";
+            writeObjectToStream(stream, getSortSpecView(val)->getPattern());
+            stream << ", ";
+            writeCollatorToStream(stream, getSortSpecView(val)->getCollator());
+            stream << ')';
+            break;
         default:
             MONGO_UNREACHABLE;
     }
@@ -919,7 +963,7 @@ std::pair<TypeTags, Value> compareValue(TypeTags lhsTag,
                              lsz + 1);
         return {TypeTags::NumberInt32, bitcastFrom<int32_t>(compareHelper(result, 0))};
     } else if (lhsTag == TypeTags::ksValue && rhsTag == TypeTags::ksValue) {
-        auto result = getKeyStringView(lhsValue)->compare(*getKeyStringView(lhsValue));
+        auto result = getKeyStringView(lhsValue)->compare(*getKeyStringView(rhsValue));
         return {TypeTags::NumberInt32, bitcastFrom<int32_t>(result)};
     } else if (lhsTag == TypeTags::Nothing && rhsTag == TypeTags::Nothing) {
         // Special case for Nothing in a hash table (group) and sort comparison.
diff --git a/src/mongo/db/exec/sbe/values/value.h b/src/mongo/db/exec/sbe/values/value.h
index c7265ba1fe0..f511199faa3 100644
--- a/src/mongo/db/exec/sbe/values/value.h
+++ b/src/mongo/db/exec/sbe/values/value.h
@@ -70,10 +70,11 @@ using SpoolId = int64_t;
 using IndexKeysInclusionSet = std::bitset<Ordering::kMaxCompoundIndexKeys>;
 
 namespace value {
+class SortSpec;
 
-static constexpr std::int32_t kStringMaxDisplayLength = 160;
-static constexpr std::int32_t kBinDataMaxDisplayLength = 80;
-static constexpr std::int32_t kNewUUIDLength = 16;
+static constexpr size_t kStringMaxDisplayLength = 160;
+static constexpr size_t kBinDataMaxDisplayLength = 80;
+static constexpr size_t kNewUUIDLength = 16;
 
 /**
  * Type dispatch tags.
@@ -140,6 +141,9 @@ enum class TypeTags : uint8_t {
 
     // Pointer to fts::FTSMatcher for full text search.
     ftsMatcher,
+
+    // Pointer to a SortSpec object.
+    sortSpec,
 };
 
 inline constexpr bool isNumber(TypeTags tag) noexcept {
@@ -607,8 +611,6 @@ public:
         _compile();
     }
 
-    PcreRegex(StringData pattern) : PcreRegex(pattern, "") {}
-
     PcreRegex(const PcreRegex& other) : PcreRegex(other._pattern, other._options) {}
 
     PcreRegex& operator=(const PcreRegex& other) {
@@ -654,7 +656,7 @@ private:
     std::string _pattern;
     std::string _options;
 
-    pcre* _pcrePtr;
+    pcre* _pcrePtr = nullptr;
 };
 
 constexpr size_t kSmallStringMaxLength = 7;
@@ -929,6 +931,10 @@ inline fts::FTSMatcher* getFtsMatcherView(Value val) noexcept {
     return reinterpret_cast<fts::FTSMatcher*>(val);
 }
 
+inline SortSpec* getSortSpecView(Value val) noexcept {
+    return reinterpret_cast<SortSpec*>(val);
+}
+
 /**
  * Pattern and flags of Regex are stored in BSON as two C strings written one after another.
  *
@@ -1007,6 +1013,8 @@ std::pair<TypeTags, Value> makeCopyShardFilterer(const ShardFilterer&);
 
 std::pair<TypeTags, Value> makeCopyFtsMatcher(const fts::FTSMatcher&);
 
+std::pair<TypeTags, Value> makeCopySortSpec(const SortSpec&);
+
 void releaseValue(TypeTags tag, Value val) noexcept;
 
 inline std::pair<TypeTags, Value> copyValue(TypeTags tag, Value val) {
@@ -1069,6 +1077,8 @@ inline std::pair<TypeTags, Value> copyValue(TypeTags tag, Value val) {
             return makeCopyBsonDBPointer(getBsonDBPointerView(val));
         case TypeTags::ftsMatcher:
             return makeCopyFtsMatcher(*getFtsMatcherView(val));
+        case TypeTags::sortSpec:
+            return makeCopySortSpec(*getSortSpecView(val));
         default:
             break;
     }
@@ -1281,7 +1291,6 @@ private:
 std::pair<TypeTags, Value> arrayToSet(TypeTags tag,
                                       Value val,
                                       CollatorInterface* collator = nullptr);
-
 }  // namespace value
 }  // namespace sbe
 }  // namespace mongo
diff --git a/src/mongo/db/exec/sbe/vm/vm.cpp b/src/mongo/db/exec/sbe/vm/vm.cpp
index e3e8c3b744f..1d2dce15423 100644
--- a/src/mongo/db/exec/sbe/vm/vm.cpp
+++ b/src/mongo/db/exec/sbe/vm/vm.cpp
@@ -41,9 +41,11 @@
 #include "mongo/db/client.h"
 #include "mongo/db/exec/js_function.h"
 #include "mongo/db/exec/sbe/values/bson.h"
+#include "mongo/db/exec/sbe/values/sort_spec.h"
 #include "mongo/db/exec/sbe/values/value.h"
 #include "mongo/db/exec/sbe/vm/datetime.h"
 #include "mongo/db/hasher.h"
+#include "mongo/db/index/btree_key_generator.h"
 #include "mongo/db/query/collation/collation_index_key.h"
 #include "mongo/db/query/datetime/date_time_support.h"
 #include "mongo/db/storage/key_string.h"
@@ -2845,6 +2847,34 @@ std::tuple<bool, value::TypeTags, value::Value> ByteCode::builtinGetRegexFlags(A
     return {true, strType, strValue};
 }
 
+std::tuple<bool, value::TypeTags, value::Value> ByteCode::builtinGenerateSortKey(ArityType arity) {
+    invariant(arity == 2);
+
+    auto [ssOwned, ssTag, ssVal] = getFromStack(0);
+    auto [objOwned, objTag, objVal] = getFromStack(1);
+    if (ssTag != value::TypeTags::sortSpec || !value::isObject(objTag)) {
+        return {false, value::TypeTags::Nothing, 0};
+    }
+
+    auto ss = value::getSortSpecView(ssVal);
+
+    auto obj = [objTag = objTag, objVal = objVal]() {
+        if (objTag == value::TypeTags::bsonObject) {
+            return BSONObj{value::bitcastTo<const char*>(objVal)};
+        } else if (objTag == value::TypeTags::Object) {
+            BSONObjBuilder objBuilder;
+            bson::convertToBsonObj(objBuilder, value::getObjectView(objVal));
+            return objBuilder.obj();
+        } else {
+            MONGO_UNREACHABLE_TASSERT(5037004);
+        }
+    }();
+
+    return {true,
+            value::TypeTags::ksValue,
+            value::bitcastFrom<KeyString::Value*>(new KeyString::Value(ss->generateSortKey(obj)))};
+}
+
 std::tuple<bool, value::TypeTags, value::Value> ByteCode::builtinReverseArray(ArityType arity) {
     invariant(arity == 1);
     auto [inputOwned, inputType, inputVal] = getFromStack(0);
@@ -3130,6 +3160,8 @@ std::tuple<bool, value::TypeTags, value::Value> ByteCode::dispatchBuiltin(Builti
             return builtinGetRegexFlags(arity);
         case Builtin::ftsMatch:
             return builtinFtsMatch(arity);
+        case Builtin::generateSortKey:
+            return builtinGenerateSortKey(arity);
     }
 
     MONGO_UNREACHABLE;
diff --git a/src/mongo/db/exec/sbe/vm/vm.h b/src/mongo/db/exec/sbe/vm/vm.h
index 41741883930..81c20cf2292 100644
--- a/src/mongo/db/exec/sbe/vm/vm.h
+++ b/src/mongo/db/exec/sbe/vm/vm.h
@@ -336,6 +336,7 @@ enum class Builtin : uint8_t {
     getRegexPattern,
     getRegexFlags,
     ftsMatch,
+    generateSortKey,
 };
 
 using SmallArityType = uint8_t;
@@ -752,6 +753,7 @@ private:
     std::tuple<bool, value::TypeTags, value::Value> builtinGetRegexPattern(ArityType arity);
     std::tuple<bool, value::TypeTags, value::Value> builtinGetRegexFlags(ArityType arity);
     std::tuple<bool, value::TypeTags, value::Value> builtinFtsMatch(ArityType arity);
+    std::tuple<bool, value::TypeTags, value::Value> builtinGenerateSortKey(ArityType arity);
 
     std::tuple<bool, value::TypeTags, value::Value> dispatchBuiltin(Builtin f, ArityType arity);
 
diff --git a/src/mongo/db/index/btree_key_generator.cpp b/src/mongo/db/index/btree_key_generator.cpp
index a39c2953484..017089822b5 100644
--- a/src/mongo/db/index/btree_key_generator.cpp
+++ b/src/mongo/db/index/btree_key_generator.cpp
@@ -47,7 +47,6 @@ using IndexVersion = IndexDescriptor::IndexVersion;
 namespace dps = ::mongo::dotted_path_support;
 
 namespace {
-
 const BSONObj nullObj = BSON("" << BSONNULL);
 const BSONElement nullElt = nullObj.firstElement();
 const BSONObj undefinedObj = BSON("" << BSONUndefined);
@@ -94,16 +93,16 @@ BtreeKeyGenerator::BtreeKeyGenerator(std::vector<const char*> fieldNames,
                                      KeyString::Version keyStringVersion,
                                      Ordering ordering)
     : _keyStringVersion(keyStringVersion),
-      _ordering(ordering),
-      _fieldNames(fieldNames),
       _isIdIndex(fieldNames.size() == 1 && std::string("_id") == fieldNames[0]),
       _isSparse(isSparse),
+      _ordering(ordering),
+      _fieldNames(std::move(fieldNames)),
       _nullKeyString(_buildNullKeyString()),
-      _fixed(fixed),
-      _emptyPositionalInfo(fieldNames.size()),
+      _fixed(std::move(fixed)),
+      _emptyPositionalInfo(_fieldNames.size()),
       _collator(collator) {
 
-    for (const char* fieldName : fieldNames) {
+    for (const char* fieldName : _fieldNames) {
         FieldRef fieldRef{fieldName};
         auto pathLength = fieldRef.numParts();
         invariant(pathLength > 0);
diff --git a/src/mongo/db/index/btree_key_generator.h b/src/mongo/db/index/btree_key_generator.h
index 8386c1e8520..da6bff3185c 100644
--- a/src/mongo/db/index/btree_key_generator.h
+++ b/src/mongo/db/index/btree_key_generator.h
@@ -83,18 +83,6 @@ public:
                  boost::optional<RecordId> id = boost::none) const;
 
 private:
-    const KeyString::Version _keyStringVersion;
-    const Ordering _ordering;
-    // These are used by getKeys below.
-    const std::vector<const char*> _fieldNames;
-    const bool _isIdIndex;
-    const bool _isSparse;
-    const KeyString::Value _nullKeyString;  // A full key with all fields null.
-    // True if any of the indexed paths contains a positional path component. This prohibits the key
-    // generator from using the non-multikey fast path.
-    bool _pathsContainPositionalComponent{false};
-
-    std::vector<BSONElement> _fixed;
     /**
      * Stores info regarding traversal of a positional path. A path through a document is
      * considered positional if this path element names an array element. Generally this means
@@ -233,6 +221,22 @@ private:
 
     KeyString::Value _buildNullKeyString() const;
 
+    const KeyString::Version _keyStringVersion;
+
+    const bool _isIdIndex;
+    const bool _isSparse;
+    // True if any of the indexed paths contains a positional path component. This prohibits the key
+    // generator from using the non-multikey fast path.
+    bool _pathsContainPositionalComponent{false};
+
+    const Ordering _ordering;
+
+    // These are used by getKeys below.
+    const std::vector<const char*> _fieldNames;
+    const KeyString::Value _nullKeyString;  // A full key with all fields null.
+
+    std::vector<BSONElement> _fixed;
+
     const std::vector<PositionalPathInfo> _emptyPositionalInfo;
 
     // A vector with size equal to the number of elements in the index key pattern. Each element in
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index 53263dde26d..87f8b82213a 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -1167,20 +1167,22 @@ inline bool isQuerySbeCompatible(OperationContext* opCtx,
                                  size_t plannerOptions) {
     invariant(cq);
     auto expCtx = cq->getExpCtxRaw();
-    auto sortPattern = cq->getSortPattern();
+    const auto& sortPattern = cq->getSortPattern();
     const bool allExpressionsSupported = expCtx && expCtx->sbeCompatible;
     const bool isNotCount = !(plannerOptions & QueryPlannerParams::IS_COUNT);
     const bool doesNotContainMetadataRequirements = cq->metadataDeps().none();
-    const bool doesNotSortOnDottedPath =
+    const bool doesNotSortOnMetaOrPathWithNumericComponents =
         !sortPattern || std::all_of(sortPattern->begin(), sortPattern->end(), [](auto&& part) {
-            return part.fieldPath && part.fieldPath->getPathLength() == 1;
+            return part.fieldPath &&
+                !FieldRef(part.fieldPath->fullPath()).hasNumericPathComponents();
         });
     // OP_QUERY style find commands are not currently supported by SBE.
     const bool isNotLegacy = !CurOp::get(opCtx)->isLegacyQuery();
     // Queries against a time-series collection are not currently supported by SBE.
     const bool isQueryNotAgainstTimeseriesCollection = !(cq->nss().isTimeseriesBucketsCollection());
     return allExpressionsSupported && isNotCount && doesNotContainMetadataRequirements &&
-        doesNotSortOnDottedPath && isNotLegacy && isQueryNotAgainstTimeseriesCollection;
+        isNotLegacy && isQueryNotAgainstTimeseriesCollection &&
+        doesNotSortOnMetaOrPathWithNumericComponents;
 }
 }  // namespace
 
diff --git a/src/mongo/db/query/sbe_stage_builder.cpp b/src/mongo/db/query/sbe_stage_builder.cpp
index 497d0a4e942..032510007da 100644
--- a/src/mongo/db/query/sbe_stage_builder.cpp
+++ b/src/mongo/db/query/sbe_stage_builder.cpp
@@ -47,6 +47,7 @@
 #include "mongo/db/exec/sbe/stages/traverse.h"
 #include "mongo/db/exec/sbe/stages/union.h"
 #include "mongo/db/exec/sbe/stages/unique.h"
+#include "mongo/db/exec/sbe/values/sort_spec.h"
 #include "mongo/db/exec/shard_filterer.h"
 #include "mongo/db/fts/fts_index_format.h"
 #include "mongo/db/fts/fts_query_impl.h"
@@ -681,110 +682,328 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder
     return {std::move(stage), std::move(outputs)};
 }
 
+namespace {
+using MakeSortKeyFn =
+    std::function<std::unique_ptr<sbe::EExpression>(sbe::value::SlotId inputSlot)>;
+
+/**
+ * Given a field path, this function builds a plan stage tree that will produce the corresponding
+ * sort key for that path. The 'makeSortKey' parameter is used to apply any transformations to the
+ * leaf fields' values that are necessary (for example, calling collComparisonKey()).
+ *
+ * Note that when 'level' is 0, this function assumes that 'inputSlot' alrady contains the top-level
+ * field value from the path, and thus it will forgo generating a call to getField(). When 'level'
+ * is 1 or greater, this function will generate a call to getField() to read the field for that
+ * level.
+ */
+std::pair<sbe::value::SlotId, std::unique_ptr<sbe::PlanStage>> generateSortKeyTraversal(
+    std::unique_ptr<sbe::PlanStage> inputStage,
+    sbe::value::SlotId inputSlot,
+    const FieldPath& fp,
+    sbe::value::SortDirection direction,
+    FieldIndex level,
+    PlanNodeId planNodeId,
+    sbe::value::SlotIdGenerator* slotIdGenerator,
+    const MakeSortKeyFn& makeSortKey) {
+    invariant(level < fp.getPathLength());
+
+    const bool isLeafField = (level == fp.getPathLength() - 1u);
+
+    auto [fieldSlot, fromBranch] = [&]() {
+        if (level > 0) {
+            // Generate a call to getField() to read the field at the current level and bind it to
+            // 'fieldSlot'. According to MQL's sorting semantics, if the field doesn't exist we
+            // should use Null as the sort key.
+            auto getFieldExpr = makeFunction("getField"_sd,
+                                             sbe::makeE<sbe::EVariable>(inputSlot),
+                                             sbe::makeE<sbe::EConstant>(fp.getFieldName(level)));
+
+            if (isLeafField) {
+                // Wrapping the field access with makeFillEmptyNull() is only necessary for the
+                // leaf field. For non-leaf fields, if the field doesn't exist then Nothing will
+                // propagate through the TraverseStage and afterward it will be converted to Null
+                // by a projection (see below).
+                getFieldExpr = makeFillEmptyNull(std::move(getFieldExpr));
+            }
+
+            auto fieldSlot{slotIdGenerator->generate()};
+            return std::make_pair(
+                fieldSlot,
+                sbe::makeProjectStage(
+                    std::move(inputStage), planNodeId, fieldSlot, std::move(getFieldExpr)));
+        }
+
+        return std::make_pair(inputSlot, std::move(inputStage));
+    }();
+
+    // Generate the 'in' branch for the TraverseStage that we're about to construct.
+    auto [innerSlot, innerBranch] = [&, fieldSlot = fieldSlot, &fromBranch = fromBranch]() {
+        if (isLeafField) {
+            // Base case: Genereate a ProjectStage to evaluate the predicate.
+            auto innerSlot{slotIdGenerator->generate()};
+            return std::make_pair(innerSlot,
+                                  sbe::makeProjectStage(makeLimitCoScanTree(planNodeId),
+                                                        planNodeId,
+                                                        innerSlot,
+                                                        makeSortKey(fieldSlot)));
+        } else {
+            // Recursive case.
+            return generateSortKeyTraversal(makeLimitCoScanTree(planNodeId),
+                                            fieldSlot,
+                                            fp,
+                                            direction,
+                                            level + 1,
+                                            planNodeId,
+                                            slotIdGenerator,
+                                            makeSortKey);
+        }
+    }();
+
+    // Generate the traverse stage for the current nested level. The fold expression uses
+    // well-ordered comparison (cmp3w) to produce the minimum element (if 'direction' is
+    // Ascending) or the maximum element (if 'direction' is Descending).
+    auto traverseSlot{slotIdGenerator->generate()};
+    auto outputSlot{slotIdGenerator->generate()};
+    auto op = (direction == sbe::value::SortDirection::Ascending) ? sbe::EPrimBinary::less
+                                                                  : sbe::EPrimBinary::greater;
+
+    auto outputStage = sbe::makeS<sbe::TraverseStage>(
+        std::move(fromBranch),
+        std::move(innerBranch),
+        fieldSlot,
+        traverseSlot,
+        innerSlot,
+        sbe::makeSV(),
+        sbe::makeE<sbe::EIf>(makeBinaryOp(op,
+                                          makeBinaryOp(sbe::EPrimBinary::cmp3w,
+                                                       makeVariable(innerSlot),
+                                                       makeVariable(traverseSlot)),
+                                          makeConstant(sbe::value::TypeTags::NumberInt64,
+                                                       sbe::value::bitcastFrom<int64_t>(0))),
+                             makeVariable(innerSlot),
+                             makeVariable(traverseSlot)),
+        nullptr,
+        planNodeId,
+        1);
+
+    // According to MQL's sorting semantics, when a leaf field is an empty array we should use
+    // Undefined as the sort key, and when a non-leaf field is an empty array or doesn't exist
+    // we should use Null as the sort key.
+    return {outputSlot,
+            sbe::makeProjectStage(std::move(outputStage),
+                                  planNodeId,
+                                  outputSlot,
+                                  isLeafField ? makeFillEmptyUndefined(makeVariable(traverseSlot))
+                                              : makeFillEmptyNull(makeVariable(traverseSlot)))};
+}
+
+/**
+ * Given a field path, this function will return an expression that will be true if evaluating the
+ * field path involves array traversal at any level of the path (including the leaf field).
+ */
+std::unique_ptr<sbe::EExpression> generateArrayCheckForSortHelper(
+    std::unique_ptr<sbe::EExpression> inputExpr,
+    const FieldPath& fp,
+    FieldIndex level,
+    sbe::value::FrameIdGenerator* frameIdGenerator) {
+    invariant(level < fp.getPathLength());
+
+    auto fieldExpr = makeFillEmptyNull(makeFunction(
+        "getField"_sd, std::move(inputExpr), sbe::makeE<sbe::EConstant>(fp.getFieldName(level))));
+
+    if (level == fp.getPathLength() - 1u) {
+        return makeFunction("isArray"_sd, std::move(fieldExpr));
+    } else {
+        auto frameId = frameIdGenerator->generate();
+        return sbe::makeE<sbe::ELocalBind>(
+            frameId,
+            sbe::makeEs(std::move(fieldExpr)),
+            makeBinaryOp(sbe::EPrimBinary::logicOr,
+                         makeFunction("isArray"_sd, makeVariable(frameId, 0)),
+                         generateArrayCheckForSortHelper(
+                             makeVariable(frameId, 0), fp, level + 1, frameIdGenerator)));
+    }
+}
+
+/**
+ * Given a field path and a slot that holds the top-level field's value from that path, this
+ * function will return an expression that will be true if evaluating the field path involves array
+ * traversal at any level of the path (including the leaf field).
+ */
+std::unique_ptr<sbe::EExpression> generateArrayCheckForSort(
+    sbe::value::SlotId inputSlot,
+    const FieldPath& fp,
+    sbe::value::FrameIdGenerator* frameIdGenerator) {
+    if (fp.getPathLength() == 1) {
+        return makeFunction("isArray"_sd, makeVariable(inputSlot));
+    } else {
+        return makeBinaryOp(
+            sbe::EPrimBinary::logicOr,
+            makeFunction("isArray"_sd, makeVariable(inputSlot)),
+            generateArrayCheckForSortHelper(makeVariable(inputSlot), fp, 1, frameIdGenerator));
+    }
+}
+}  // namespace
+
 std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> SlotBasedStageBuilder::buildSort(
     const QuerySolutionNode* root, const PlanStageReqs& reqs) {
-    using namespace std::literals;
     invariant(!reqs.getIndexKeyBitset());
 
     const auto sn = static_cast<const SortNode*>(root);
     auto sortPattern = SortPattern{sn->pattern, _cq.getExpCtx()};
 
+    tassert(5037001,
+            "QueryPlannerAnalysis should not produce a SortNode with an empty sort pattern",
+            sortPattern.size() > 0);
+
     // The child must produce all of the slots required by the parent of this SortNode. In addition
     // to that, the child must always produce a 'resultSlot' because it's needed by the sort logic
     // below.
     auto childReqs = reqs.copy().set(kResult);
     auto [inputStage, outputs] = build(sn->children[0], childReqs);
 
+    auto collatorSlot = _data.env->getSlotIfExists("collator"_sd);
+
     sbe::value::SlotVector orderBy;
     std::vector<sbe::value::SortDirection> direction;
-    sbe::value::SlotMap<std::unique_ptr<sbe::EExpression>> projectMap;
+    StringDataSet prefixSet;
+    bool hasPartsWithCommonPrefix = false;
 
     for (const auto& part : sortPattern) {
-        uassert(5073801, "Sorting by expression not supported", !part.expression);
-        uassert(5073802,
-                "Sorting by dotted paths not supported",
-                part.fieldPath && part.fieldPath->getPathLength() == 1);
-
-        // Slot holding the sort key.
-        auto sortFieldVar{_slotIdGenerator.generate()};
-        orderBy.push_back(sortFieldVar);
+        // getExecutor() should never call into buildSlotBasedExecutableTree() when the query
+        // contains $meta, so this assertion should always be true.
+        tassert(5037002, "Sort with $meta is not supported in SBE", part.fieldPath);
+
+        if (!hasPartsWithCommonPrefix) {
+            auto [_, prefixWasNotPresent] = prefixSet.insert(part.fieldPath->getFieldName(0));
+            hasPartsWithCommonPrefix = !prefixWasNotPresent;
+        }
+
+        // Record the direction for this part of the sort pattern
         direction.push_back(part.isAscending ? sbe::value::SortDirection::Ascending
                                              : sbe::value::SortDirection::Descending);
+    }
 
-        // Generate projection to get the value of the sort key. Ideally, this should be
-        // tracked by a 'reference tracker' at higher level.
-        auto fieldName = part.fieldPath->getFieldName(0);
+    if (!hasPartsWithCommonPrefix) {
+        sbe::value::SlotMap<std::unique_ptr<sbe::EExpression>> projectMap;
 
-        auto getSortFieldExpr = makeFunction("getField"_sd,
-                                             sbe::makeE<sbe::EVariable>(outputs.get(kResult)),
-                                             sbe::makeE<sbe::EConstant>(fieldName));
+        for (const auto& part : sortPattern) {
+            // Get the top-level field for this sort part. If the field doesn't exist, according to
+            // MQL's sorting semantics we should use Null.
+            auto getFieldExpr = makeFillEmptyNull(
+                makeFunction("getField"_sd,
+                             makeVariable(outputs.get(kResult)),
+                             sbe::makeE<sbe::EConstant>(part.fieldPath->getFieldName(0))));
+
+            auto fieldSlot{_slotIdGenerator.generate()};
+            projectMap.emplace(fieldSlot, std::move(getFieldExpr));
 
-        if (auto collatorSlot = _data.env->getSlotIfExists("collator"_sd); collatorSlot) {
-            getSortFieldExpr = makeFunction("collComparisonKey"_sd,
-                                            std::move(getSortFieldExpr),
-                                            sbe::makeE<sbe::EVariable>(*collatorSlot));
+            orderBy.push_back(fieldSlot);
         }
 
-        // According to MQL semantics, missing values are treated as nulls during sorting.
-        getSortFieldExpr = makeFunction("fillEmpty"_sd,
-                                        std::move(getSortFieldExpr),
-                                        makeConstant(sbe::value::TypeTags::Null, 0));
+        inputStage = sbe::makeS<sbe::ProjectStage>(
+            std::move(inputStage), std::move(projectMap), root->nodeId());
+
+        auto failOnParallelArrays = [&]() -> std::unique_ptr<mongo::sbe::EExpression> {
+            auto parallelArraysError = sbe::makeE<sbe::EFail>(
+                ErrorCodes::BadValue, "cannot sort with keys that are parallel arrays");
+
+            if (sortPattern.size() < 2) {
+                // If the sort pattern only has one part, we don't need to generate a "parallel
+                // arrays" check.
+                return {};
+            } else if (sortPattern.size() == 2) {
+                // If the sort pattern has two parts, we can generate a simpler expression to
+                // perform the "parallel arrays" check.
+                auto makeIsNotArrayCheck = [&](sbe::value::SlotId slot, const FieldPath& fp) {
+                    return makeNot(generateArrayCheckForSort(slot, fp, &_frameIdGenerator));
+                };
+
+                return makeBinaryOp(
+                    sbe::EPrimBinary::logicOr,
+                    makeIsNotArrayCheck(orderBy[0], *sortPattern[0].fieldPath),
+                    makeBinaryOp(sbe::EPrimBinary::logicOr,
+                                 makeIsNotArrayCheck(orderBy[1], *sortPattern[1].fieldPath),
+                                 std::move(parallelArraysError)));
+            } else {
+                // If the sort pattern has three or more parts, we generate an expression to
+                // perform the "parallel arrays" check that works (and scales well) for an
+                // arbitrary number of sort pattern parts.
+                auto makeIsArrayCheck = [&](sbe::value::SlotId slot, const FieldPath& fp) {
+                    return makeBinaryOp(sbe::EPrimBinary::cmp3w,
+                                        generateArrayCheckForSort(slot, fp, &_frameIdGenerator),
+                                        makeConstant(sbe::value::TypeTags::Boolean, false));
+                };
+
+                auto numArraysExpr = makeIsArrayCheck(orderBy[0], *sortPattern[0].fieldPath);
+                for (size_t idx = 1; idx < sortPattern.size(); ++idx) {
+                    numArraysExpr =
+                        makeBinaryOp(sbe::EPrimBinary::add,
+                                     std::move(numArraysExpr),
+                                     makeIsArrayCheck(orderBy[idx], *sortPattern[idx].fieldPath));
+                }
 
-        projectMap.emplace(sortFieldVar, std::move(getSortFieldExpr));
-    }
+                return makeBinaryOp(
+                    sbe::EPrimBinary::logicOr,
+                    makeBinaryOp(sbe::EPrimBinary::lessEq,
+                                 std::move(numArraysExpr),
+                                 makeConstant(sbe::value::TypeTags::NumberInt32, 1)),
+                    std::move(parallelArraysError));
+            }
+        }();
 
-    inputStage =
-        sbe::makeS<sbe::ProjectStage>(std::move(inputStage), std::move(projectMap), root->nodeId());
-
-    // Generate traversals to pick the min/max element from arrays.
-    for (size_t idx = 0; idx < orderBy.size(); ++idx) {
-        auto resultVar{_slotIdGenerator.generate()};
-        auto innerVar{_slotIdGenerator.generate()};
-
-        auto innerBranch = sbe::makeProjectStage(
-            sbe::makeS<sbe::LimitSkipStage>(
-                sbe::makeS<sbe::CoScanStage>(root->nodeId()), 1, boost::none, root->nodeId()),
-            root->nodeId(),
-            innerVar,
-            sbe::makeE<sbe::EVariable>(orderBy[idx]));
-
-        auto op = direction[idx] == sbe::value::SortDirection::Ascending
-            ? sbe::EPrimBinary::less
-            : sbe::EPrimBinary::greater;
-        auto minmax =
-            sbe::makeE<sbe::EIf>(makeBinaryOp(op,
-                                              makeBinaryOp(sbe::EPrimBinary::cmp3w,
-                                                           sbe::makeE<sbe::EVariable>(innerVar),
-                                                           sbe::makeE<sbe::EVariable>(resultVar)),
-                                              makeConstant(sbe::value::TypeTags::NumberInt64,
-                                                           sbe::value::bitcastFrom<int64_t>(0))),
-                                 sbe::makeE<sbe::EVariable>(innerVar),
-                                 sbe::makeE<sbe::EVariable>(resultVar));
-
-        inputStage = sbe::makeS<sbe::TraverseStage>(std::move(inputStage),
-                                                    std::move(innerBranch),
-                                                    orderBy[idx],
-                                                    resultVar,
-                                                    innerVar,
-                                                    sbe::makeSV(),
-                                                    std::move(minmax),
-                                                    nullptr,
-                                                    root->nodeId(),
-                                                    boost::none);
-        orderBy[idx] = resultVar;
-    }
+        if (failOnParallelArrays) {
+            inputStage = sbe::makeProjectStage(std::move(inputStage),
+                                               root->nodeId(),
+                                               _slotIdGenerator.generate(),
+                                               std::move(failOnParallelArrays));
+        }
 
-    if (auto recordIdSlot = outputs.getIfExists(kRecordId); recordIdSlot) {
-        // Break ties with record id if available.
-        orderBy.push_back(*recordIdSlot);
-        // This is arbitrary.
-        direction.push_back(sbe::value::SortDirection::Ascending);
+        for (size_t idx = 0; idx < orderBy.size(); ++idx) {
+            auto makeSortKey = [&](sbe::value::SlotId inputSlot) {
+                return !collatorSlot ? makeVariable(inputSlot)
+                                     : makeFunction("collComparisonKey"_sd,
+                                                    makeVariable(inputSlot),
+                                                    makeVariable(*collatorSlot));
+            };
+
+            // Call generateSortKeyTraversal() to build a series of TraverseStages that will
+            // traverse this part's field path and produce the corresponding sort key. We pass
+            // in the 'makeSortKey' lambda, which will be applied on each leaf field's value
+            // to apply the current collation (if there is one).
+            sbe::value::SlotId sortKeySlot;
+            std::tie(sortKeySlot, inputStage) =
+                generateSortKeyTraversal(std::move(inputStage),
+                                         orderBy[idx],
+                                         *sortPattern[idx].fieldPath,
+                                         direction[idx],
+                                         0,
+                                         root->nodeId(),
+                                         &_slotIdGenerator,
+                                         makeSortKey);
+
+            orderBy[idx] = sortKeySlot;
+        }
+    } else {
+        // Handle the case where two or more parts of the sort pattern have a common prefix.
+        orderBy = _slotIdGenerator.generateMultiple(1);
+        direction = {sbe::value::SortDirection::Ascending};
+
+        auto sortSpecExpr =
+            makeConstant(sbe::value::TypeTags::sortSpec,
+                         sbe::value::bitcastFrom<sbe::value::SortSpec*>(
+                             new sbe::value::SortSpec(sn->pattern, _cq.getCollator())));
+
+        inputStage = sbe::makeProjectStage(std::move(inputStage),
+                                           root->nodeId(),
+                                           orderBy[0],
+                                           makeFunction("generateSortKey",
+                                                        std::move(sortSpecExpr),
+                                                        makeVariable(outputs.get(kResult))));
     }
 
-    auto forwardingReqs = reqs.copy().set(kResult).clear(kRecordId);
-
     auto values = sbe::makeSV();
-    outputs.forEachSlot(forwardingReqs, [&](auto&& slot) { values.push_back(slot); });
+    outputs.forEachSlot(childReqs, [&](auto&& slot) { values.push_back(slot); });
 
     inputStage =
         sbe::makeS<sbe::SortStage>(std::move(inputStage),
diff --git a/src/mongo/db/query/sbe_stage_builder_helpers.cpp b/src/mongo/db/query/sbe_stage_builder_helpers.cpp
index bf60eb6d0f6..519e6121e68 100644
--- a/src/mongo/db/query/sbe_stage_builder_helpers.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_helpers.cpp
@@ -203,10 +203,12 @@ std::unique_ptr<sbe::EExpression> makeFillEmptyFalse(std::unique_ptr<sbe::EExpre
                                                    sbe::value::bitcastFrom<bool>(false)));
 }
 
-std::unique_ptr<sbe::EExpression> makeVariable(sbe::value::SlotId slotId,
-                                               boost::optional<sbe::FrameId> frameId) {
-    return frameId ? sbe::makeE<sbe::EVariable>(*frameId, slotId)
-                   : sbe::makeE<sbe::EVariable>(slotId);
+std::unique_ptr<sbe::EExpression> makeVariable(sbe::value::SlotId slotId) {
+    return sbe::makeE<sbe::EVariable>(slotId);
+}
+
+std::unique_ptr<sbe::EExpression> makeVariable(sbe::FrameId frameId, sbe::value::SlotId slotId) {
+    return sbe::makeE<sbe::EVariable>(frameId, slotId);
 }
 
 std::unique_ptr<sbe::EExpression> makeFillEmptyNull(std::unique_ptr<sbe::EExpression> e) {
@@ -215,6 +217,13 @@ std::unique_ptr<sbe::EExpression> makeFillEmptyNull(std::unique_ptr<sbe::EExpres
         "fillEmpty"_sd, std::move(e), sbe::makeE<sbe::EConstant>(sbe::value::TypeTags::Null, 0));
 }
 
+std::unique_ptr<sbe::EExpression> makeFillEmptyUndefined(std::unique_ptr<sbe::EExpression> e) {
+    using namespace std::literals;
+    return makeFunction("fillEmpty"_sd,
+                        std::move(e),
+                        sbe::makeE<sbe::EConstant>(sbe::value::TypeTags::bsonUndefined, 0));
+}
+
 std::unique_ptr<sbe::EExpression> makeNothingArrayCheck(
     std::unique_ptr<sbe::EExpression> isArrayInput, std::unique_ptr<sbe::EExpression> otherwise) {
     using namespace std::literals;
diff --git a/src/mongo/db/query/sbe_stage_builder_helpers.h b/src/mongo/db/query/sbe_stage_builder_helpers.h
index 659ef029ac1..48e9f723a1d 100644
--- a/src/mongo/db/query/sbe_stage_builder_helpers.h
+++ b/src/mongo/db/query/sbe_stage_builder_helpers.h
@@ -202,16 +202,22 @@ inline auto makeConstant(StringData str) {
     return sbe::makeE<sbe::EConstant>(tag, value);
 }
 
-std::unique_ptr<sbe::EExpression> makeVariable(sbe::value::SlotId slotId,
-                                               boost::optional<sbe::FrameId> frameId = {});
+std::unique_ptr<sbe::EExpression> makeVariable(sbe::value::SlotId slotId);
+
+std::unique_ptr<sbe::EExpression> makeVariable(sbe::FrameId frameId, sbe::value::SlotId slotId);
 
 /**
- * Check if expression returns Nothing and return null if so. Otherwise, return the
- * expression.
+ * Check if expression returns Nothing and return null if so. Otherwise, return the expression.
  */
 std::unique_ptr<sbe::EExpression> makeFillEmptyNull(std::unique_ptr<sbe::EExpression> e);
 
 /**
+ * Check if expression returns Nothing and return bsonUndefined if so. Otherwise, return the
+ * expression.
+ */
+std::unique_ptr<sbe::EExpression> makeFillEmptyUndefined(std::unique_ptr<sbe::EExpression> e);
+
+/**
  * Check if expression returns an array and return Nothing if so. Otherwise, return the expression.
  */
 std::unique_ptr<sbe::EExpression> makeNothingArrayCheck(