SERVER-63279 Allow DocumentSourceInternalUnpackBucket to interoperate with DocumentSource::pushMatchBefore

8 files changed, 71 insertions, 265 deletions

diff --git a/src/mongo/db/exec/bucket_unpacker.cpp b/src/mongo/db/exec/bucket_unpacker.cpp
index 3d7c814a20f..3c8eecbae06 100644
--- a/src/mongo/db/exec/bucket_unpacker.cpp
+++ b/src/mongo/db/exec/bucket_unpacker.cpp
@@ -212,6 +212,7 @@ std::unique_ptr<MatchExpression> createComparisonPredicate(
     ExpressionContext::CollationMatchesDefault collationMatchesDefault,
     boost::intrusive_ptr<ExpressionContext> pExpCtx,
     bool haveComputedMetaField,
+    bool includeMetaField,
     bool assumeNoMixedSchemaData,
     IneligiblePredicatePolicy policy) {
     using namespace timeseries;
@@ -255,6 +256,9 @@ std::unique_ptr<MatchExpression> createComparisonPredicate(
         if (haveComputedMetaField)
             return handleIneligible(policy, matchExpr, "can't handle a computed meta field");
 
+        if (!includeMetaField)
+            return handleIneligible(policy, matchExpr, "cannot handle an excluded meta field");
+
         auto result = matchExpr->shallowClone();
         expression::applyRenamesToExpression(
             result.get(),
@@ -447,6 +451,7 @@ std::unique_ptr<MatchExpression> BucketSpec::createPredicatesOnBucketLevelField(
     ExpressionContext::CollationMatchesDefault collationMatchesDefault,
     const boost::intrusive_ptr<ExpressionContext>& pExpCtx,
     bool haveComputedMetaField,
+    bool includeMetaField,
     bool assumeNoMixedSchemaData,
     IneligiblePredicatePolicy policy) {
 
@@ -463,6 +468,7 @@ std::unique_ptr<MatchExpression> BucketSpec::createPredicatesOnBucketLevelField(
                                                                 collationMatchesDefault,
                                                                 pExpCtx,
                                                                 haveComputedMetaField,
+                                                                includeMetaField,
                                                                 assumeNoMixedSchemaData,
                                                                 policy)) {
                 andMatchExpr->add(std::move(child));
@@ -494,6 +500,7 @@ std::unique_ptr<MatchExpression> BucketSpec::createPredicatesOnBucketLevelField(
                                                             collationMatchesDefault,
                                                             pExpCtx,
                                                             haveComputedMetaField,
+                                                            includeMetaField,
                                                             assumeNoMixedSchemaData,
                                                             policy);
             if (child) {
@@ -522,6 +529,7 @@ std::unique_ptr<MatchExpression> BucketSpec::createPredicatesOnBucketLevelField(
             collationMatchesDefault,
             pExpCtx,
             haveComputedMetaField,
+            includeMetaField,
             assumeNoMixedSchemaData,
             policy);
     } else if (matchExpr->matchType() == MatchExpression::GEO) {
@@ -575,6 +583,7 @@ std::unique_ptr<MatchExpression> BucketSpec::createPredicatesOnBucketLevelField(
                                                    collationMatchesDefault,
                                                    pExpCtx,
                                                    haveComputedMetaField,
+                                                   includeMetaField,
                                                    assumeNoMixedSchemaData,
                                                    policy);
 
@@ -604,6 +613,7 @@ BSONObj BucketSpec::pushdownPredicate(
     ExpressionContext::CollationMatchesDefault collationMatchesDefault,
     const BSONObj& predicate,
     bool haveComputedMetaField,
+    bool includeMetaField,
     bool assumeNoMixedSchemaData,
     IneligiblePredicatePolicy policy) {
 
@@ -647,6 +657,7 @@ BSONObj BucketSpec::pushdownPredicate(
               collationMatchesDefault,
               expCtx,
               haveComputedMetaField,
+              includeMetaField,
               assumeNoMixedSchemaData,
               policy)
         : nullptr;
diff --git a/src/mongo/db/exec/bucket_unpacker.h b/src/mongo/db/exec/bucket_unpacker.h
index d5e63d37c75..ac2cbf2f252 100644
--- a/src/mongo/db/exec/bucket_unpacker.h
+++ b/src/mongo/db/exec/bucket_unpacker.h
@@ -144,6 +144,7 @@ public:
         ExpressionContext::CollationMatchesDefault collationMatchesDefault,
         const boost::intrusive_ptr<ExpressionContext>& pExpCtx,
         bool haveComputedMetaField,
+        bool includeMetaField,
         bool assumeNoMixedSchemaData,
         IneligiblePredicatePolicy policy);
 
@@ -172,6 +173,7 @@ public:
         ExpressionContext::CollationMatchesDefault collationMatchesDefault,
         const BSONObj& predicate,
         bool haveComputedMetaField,
+        bool includeMetaField,
         bool assumeNoMixedSchemaData,
         IneligiblePredicatePolicy policy);
 
diff --git a/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp b/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp
index ce134b24b22..6dccf7eaf37 100644
--- a/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp
+++ b/src/mongo/db/pipeline/document_source_internal_unpack_bucket.cpp
@@ -519,20 +519,11 @@ DocumentSourceInternalUnpackBucket::createPredicatesOnBucketLevelField(
         pExpCtx->collationMatchesDefault,
         pExpCtx,
         haveComputedMetaField(),
+        _bucketUnpacker.includeMetaField(),
         _assumeNoMixedSchemaData,
         BucketSpec::IneligiblePredicatePolicy::kIgnore);
 }
 
-std::pair<boost::intrusive_ptr<DocumentSourceMatch>, boost::intrusive_ptr<DocumentSourceMatch>>
-DocumentSourceInternalUnpackBucket::splitMatchOnMetaAndRename(
-    boost::intrusive_ptr<DocumentSourceMatch> match) {
-    if (auto&& metaField = _bucketUnpacker.bucketSpec().metaField()) {
-        return std::move(*match).extractMatchOnFieldsAndRemainder(
-            {*metaField}, {{*metaField, timeseries::kBucketMetaFieldName.toString()}});
-    }
-    return {nullptr, match};
-}
-
 std::pair<BSONObj, bool> DocumentSourceInternalUnpackBucket::extractProjectForPushDown(
     DocumentSource* src) const {
     if (auto nextProject = dynamic_cast<DocumentSourceSingleDocumentTransformation*>(src);
@@ -752,6 +743,10 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi
         Pipeline::optimizeEndOfPipeline(itr, container);
         if (std::next(itr) == container->end()) {
             return container->end();
+        } else {
+            // Kick back out to optimizing this stage again a level up, so any matches that were
+            // moved to directly after this stage can be moved before it if possible.
+            return itr;
         }
     }
     {
@@ -777,25 +772,6 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi
         }
     }
 
-    // Attempt to push predicates on the metaField past $_internalUnpackBucket.
-    if (auto nextMatch = dynamic_cast<DocumentSourceMatch*>(std::next(itr)->get());
-        nextMatch && !haveComputedMetaField) {
-        auto [metaMatch, remainingMatch] = splitMatchOnMetaAndRename(nextMatch);
-
-        // The old $match can be removed and potentially replaced with 'remainingMatch'.
-        container->erase(std::next(itr));
-        if (remainingMatch) {
-            container->insert(std::next(itr), remainingMatch);
-        }
-
-        // 'metaMatch' can be pushed down and given a chance to optimize with other stages.
-        if (metaMatch) {
-            container->insert(itr, metaMatch);
-            return std::prev(itr) == container->begin() ? std::prev(itr)
-                                                        : std::prev(std::prev(itr));
-        }
-    }
-
     // Attempt to map predicates on bucketed fields to predicates on the control field.
     if (auto nextMatch = dynamic_cast<DocumentSourceMatch*>(std::next(itr)->get());
         nextMatch && !_triedBucketLevelFieldsPredicatesPushdown) {
@@ -853,4 +829,15 @@ Pipeline::SourceContainer::iterator DocumentSourceInternalUnpackBucket::doOptimi
 
     return container->end();
 }
+
+DocumentSource::GetModPathsReturn DocumentSourceInternalUnpackBucket::getModifiedPaths() const {
+    if (_bucketUnpacker.includeMetaField()) {
+        StringMap<std::string> renames;
+        renames.emplace(*_bucketUnpacker.bucketSpec().metaField(),
+                        timeseries::kBucketMetaFieldName);
+        return {GetModPathsReturn::Type::kAllExcept, std::set<std::string>{}, std::move(renames)};
+    }
+    return {GetModPathsReturn::Type::kAllPaths, std::set<std::string>{}, {}};
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/document_source_internal_unpack_bucket.h b/src/mongo/db/pipeline/document_source_internal_unpack_bucket.h
index bebe67b4a20..32079ed2ceb 100644
--- a/src/mongo/db/pipeline/document_source_internal_unpack_bucket.h
+++ b/src/mongo/db/pipeline/document_source_internal_unpack_bucket.h
@@ -80,15 +80,17 @@ public:
     }
 
     StageConstraints constraints(Pipeline::SplitState pipeState) const final {
-        return {StreamType::kStreaming,
-                PositionRequirement::kNone,
-                HostTypeRequirement::kNone,
-                DiskUseRequirement::kNoDiskUse,
-                FacetRequirement::kNotAllowed,
-                TransactionRequirement::kAllowed,
-                LookupRequirement::kAllowed,
-                UnionRequirement::kAllowed,
-                ChangeStreamRequirement::kDenylist};
+        StageConstraints constraints{StreamType::kStreaming,
+                                     PositionRequirement::kNone,
+                                     HostTypeRequirement::kNone,
+                                     DiskUseRequirement::kNoDiskUse,
+                                     FacetRequirement::kNotAllowed,
+                                     TransactionRequirement::kAllowed,
+                                     LookupRequirement::kAllowed,
+                                     UnionRequirement::kAllowed,
+                                     ChangeStreamRequirement::kDenylist};
+        constraints.canSwapWithMatch = true;
+        return constraints;
     }
 
     DepsTracker::State getDependencies(DepsTracker* deps) const final {
@@ -188,6 +190,8 @@ public:
     std::pair<bool, Pipeline::SourceContainer::iterator> rewriteGroupByMinMax(
         Pipeline::SourceContainer::iterator itr, Pipeline::SourceContainer* container);
 
+    GetModPathsReturn getModifiedPaths() const final override;
+
 private:
     GetNextResult doGetNext() final;
     bool haveComputedMetaField() const;
diff --git a/src/mongo/db/pipeline/document_source_internal_unpack_bucket_test/split_match_on_meta_and_rename_test.cpp b/src/mongo/db/pipeline/document_source_internal_unpack_bucket_test/split_match_on_meta_and_rename_test.cpp
index 8f31d177a53..ba4f31adf17 100644
--- a/src/mongo/db/pipeline/document_source_internal_unpack_bucket_test/split_match_on_meta_and_rename_test.cpp
+++ b/src/mongo/db/pipeline/document_source_internal_unpack_bucket_test/split_match_on_meta_and_rename_test.cpp
@@ -41,210 +41,6 @@ namespace {
 
 using InternalUnpackBucketSplitMatchOnMetaAndRename = AggregationContextFixture;
 
-TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename, DoesNotSplitWhenNoMetaFieldSpecified) {
-    auto unpack = DocumentSourceInternalUnpackBucket::createFromBsonInternal(
-        fromjson("{$_internalUnpackBucket: { exclude: [], timeField: 'foo', bucketMaxSpanSeconds: "
-                 "3600}}")
-            .firstElement(),
-        getExpCtx());
-    auto matchToSplit = DocumentSourceMatch::create(fromjson("{meta: {$gt: 1}}"), getExpCtx());
-
-    auto [metaOnlyMatch, remainingMatch] =
-        dynamic_cast<DocumentSourceInternalUnpackBucket*>(unpack.get())
-            ->splitMatchOnMetaAndRename(matchToSplit.get());
-
-    // Can't split when there is no metaField specified in the stage.
-    ASSERT_FALSE(metaOnlyMatch);
-    ASSERT_TRUE(remainingMatch);
-    ASSERT_BSONOBJ_EQ(matchToSplit->getQuery(), remainingMatch->getQuery());
-}
-
-TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename, DoesNotSplitWhenNoMatchOnMetaField) {
-    auto unpack = DocumentSourceInternalUnpackBucket::createFromBsonInternal(
-        fromjson("{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'myMeta', "
-                 "bucketMaxSpanSeconds: 3600}}")
-            .firstElement(),
-        getExpCtx());
-    auto matchToSplit = DocumentSourceMatch::create(fromjson("{a: {$gt: 1}}"), getExpCtx());
-
-    auto [metaOnlyMatch, remainingMatch] =
-        dynamic_cast<DocumentSourceInternalUnpackBucket*>(unpack.get())
-            ->splitMatchOnMetaAndRename(matchToSplit.get());
-
-    // Can't split when the match does not reference the metaField.
-    ASSERT_FALSE(metaOnlyMatch);
-    ASSERT_TRUE(remainingMatch);
-    ASSERT_BSONOBJ_EQ(matchToSplit->getQuery(), remainingMatch->getQuery());
-}
-
-TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename, SplitsWhenEntireMatchIsOnMetaField) {
-    auto unpack = DocumentSourceInternalUnpackBucket::createFromBsonInternal(
-        fromjson("{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'myMeta', "
-                 "bucketMaxSpanSeconds: 3600}}")
-            .firstElement(),
-        getExpCtx());
-    auto matchToSplit = DocumentSourceMatch::create(
-        fromjson("{$or: [{myMeta: {$gt: 1}}, {'myMeta.a': {$lt: 1}}]}"), getExpCtx());
-
-    auto [metaOnlyMatch, remainingMatch] =
-        dynamic_cast<DocumentSourceInternalUnpackBucket*>(unpack.get())
-            ->splitMatchOnMetaAndRename(matchToSplit.get());
-
-    // Can split and rename when the match is entirely on the metaField.
-    ASSERT_TRUE(metaOnlyMatch);
-    ASSERT_BSONOBJ_EQ(fromjson("{$or: [{meta: {$gt: 1}}, {'meta.a': {$lt: 1}}]}"),
-                      metaOnlyMatch->getQuery());
-    ASSERT_FALSE(remainingMatch);
-}
-
-TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename,
-       SplitsWhenIndependentPartOfMatchIsOnMetaField) {
-    auto unpack = DocumentSourceInternalUnpackBucket::createFromBsonInternal(
-        fromjson("{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'myMeta', "
-                 "bucketMaxSpanSeconds: 3600}}")
-            .firstElement(),
-        getExpCtx());
-    auto matchToSplit = DocumentSourceMatch::create(
-        fromjson("{$and: [{'myMeta.a': {$gt: 1}}, {b: {$lt: 1}}]}"), getExpCtx());
-
-    auto [metaOnlyMatch, remainingMatch] =
-        dynamic_cast<DocumentSourceInternalUnpackBucket*>(unpack.get())
-            ->splitMatchOnMetaAndRename(matchToSplit.get());
-
-    // Can split and rename when an independent part of the match is on the metaField.
-    ASSERT_TRUE(metaOnlyMatch);
-    ASSERT_BSONOBJ_EQ(fromjson("{'meta.a': {$gt: 1}}"), metaOnlyMatch->getQuery());
-    ASSERT_TRUE(remainingMatch);
-    ASSERT_BSONOBJ_EQ(fromjson("{b: {$lt: 1}}"), remainingMatch->getQuery());
-}
-
-TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename,
-       DoesNotSplitsWhenDependentPartOfMatchIsOnMetaField) {
-    auto unpack = DocumentSourceInternalUnpackBucket::createFromBsonInternal(
-        fromjson("{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'meta', "
-                 "bucketMaxSpanSeconds: 3600}}")
-            .firstElement(),
-        getExpCtx());
-    auto matchToSplit = DocumentSourceMatch::create(
-        fromjson("{$or: [{'meta.a': {$gt: 1}}, {metaXYZ: {$lt: 1}}]}"), getExpCtx());
-
-    auto [metaOnlyMatch, remainingMatch] =
-        dynamic_cast<DocumentSourceInternalUnpackBucket*>(unpack.get())
-            ->splitMatchOnMetaAndRename(matchToSplit.get());
-
-    // Can't split when the part of the match that is on the metaField is dependent on the rest.
-    // Even though 'metaXYZ' is prefixed by 'meta', it's not a subfield. The presence of a top-level
-    // $or means this match cannot be correctly split into two matches.
-    ASSERT_FALSE(metaOnlyMatch);
-    ASSERT_TRUE(remainingMatch);
-    ASSERT_BSONOBJ_EQ(matchToSplit->getQuery(), remainingMatch->getQuery());
-}
-
-TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename, SplitsWhenSharedPrefixOfMetaIsNotSubfield) {
-    auto unpack = DocumentSourceInternalUnpackBucket::createFromBsonInternal(
-        fromjson("{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'myMeta', "
-                 "bucketMaxSpanSeconds: 3600}}")
-            .firstElement(),
-        getExpCtx());
-    auto matchToSplit = DocumentSourceMatch::create(
-        fromjson("{$and: [{myMeta: {$gt: 1}}, {myMetaXYZ: {$lt: 1}}]}"), getExpCtx());
-
-    auto [metaOnlyMatch, remainingMatch] =
-        dynamic_cast<DocumentSourceInternalUnpackBucket*>(unpack.get())
-            ->splitMatchOnMetaAndRename(matchToSplit.get());
-
-    // Can split and rename when an independent part of the match is on the metaField. Even though
-    // 'myMetaXYZ' is prefixed by 'myMeta', it's not a subfield, so it should not be pushed down.
-    ASSERT_TRUE(metaOnlyMatch);
-    ASSERT_BSONOBJ_EQ(fromjson("{meta: {$gt: 1}}"), metaOnlyMatch->getQuery());
-    ASSERT_TRUE(remainingMatch);
-    ASSERT_BSONOBJ_EQ(fromjson("{myMetaXYZ: {$lt: 1}}"), remainingMatch->getQuery());
-}
-
-TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename, SplitsAndRenamesWithExpr) {
-    auto unpack = DocumentSourceInternalUnpackBucket::createFromBsonInternal(
-        fromjson("{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'myMeta', "
-                 "bucketMaxSpanSeconds: 3600}}")
-            .firstElement(),
-        getExpCtx());
-    auto matchToSplit =
-        DocumentSourceMatch::create(fromjson("{$expr: {$eq: ['$myMeta.a', 2]}}"), getExpCtx());
-
-    auto [metaOnlyMatch, remainingMatch] =
-        dynamic_cast<DocumentSourceInternalUnpackBucket*>(unpack.get())
-            ->splitMatchOnMetaAndRename(matchToSplit.get());
-
-    // Can split and rename when the $match includes a $expr.
-    ASSERT_TRUE(metaOnlyMatch);
-    ASSERT_BSONOBJ_EQ(fromjson("{$expr: {$eq: ['$meta.a', {$const: 2}]}}"),
-                      metaOnlyMatch->getQuery());
-    ASSERT_FALSE(remainingMatch);
-}
-
-TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename, SplitsAndRenamesWithType) {
-    auto unpack = DocumentSourceInternalUnpackBucket::createFromBsonInternal(
-        fromjson("{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'myMeta', "
-                 "bucketMaxSpanSeconds: 3600}}")
-            .firstElement(),
-        getExpCtx());
-    auto matchToSplit =
-        DocumentSourceMatch::create(fromjson("{myMeta: {$type: [4]}}"), getExpCtx());
-
-    auto [metaOnlyMatch, remainingMatch] =
-        dynamic_cast<DocumentSourceInternalUnpackBucket*>(unpack.get())
-            ->splitMatchOnMetaAndRename(matchToSplit.get());
-
-    // Can split and rename when the $match includes a $type.
-    ASSERT_TRUE(metaOnlyMatch);
-    ASSERT_BSONOBJ_EQ(fromjson("{meta: {$type: [4]}}"), metaOnlyMatch->getQuery());
-    ASSERT_FALSE(remainingMatch);
-}
-
-TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename, SplitsAndRenamesWhenMultiplePredicates) {
-    auto unpack = DocumentSourceInternalUnpackBucket::createFromBsonInternal(
-        fromjson("{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'myMeta', "
-                 "bucketMaxSpanSeconds: 3600}}")
-            .firstElement(),
-        getExpCtx());
-    auto matchToSplit = DocumentSourceMatch::create(
-        fromjson("{myMeta: {$gte: 0, $lte: 5}, l: {$type: [4]}}"), getExpCtx());
-
-    auto [metaOnlyMatch, remainingMatch] =
-        dynamic_cast<DocumentSourceInternalUnpackBucket*>(unpack.get())
-            ->splitMatchOnMetaAndRename(matchToSplit.get());
-
-    // Can split and rename when the $match includes multiple predicates.
-    ASSERT_TRUE(metaOnlyMatch);
-    ASSERT_BSONOBJ_EQ(fromjson("{$and: [{meta: {$gte: 0}}, {meta: {$lte: 5}}]}"),
-                      metaOnlyMatch->getQuery());
-    ASSERT_TRUE(remainingMatch);
-    ASSERT_BSONOBJ_EQ(fromjson("{l: {$type: [4]}}"), remainingMatch->getQuery());
-}
-
-TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename, SplitsAndRenamesWhenSeveralFieldReferences) {
-    auto unpack = DocumentSourceInternalUnpackBucket::createFromBsonInternal(
-        fromjson("{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'myMeta', "
-                 "bucketMaxSpanSeconds: 3600}}")
-            .firstElement(),
-        getExpCtx());
-    auto matchToSplit = DocumentSourceMatch::create(
-        fromjson("{$and: [{myMeta: {$type: [3]}}, {'myMeta.a': {$gte: "
-                 "0}}, {'myMeta.b': {$type: [4]}}, {a: {$in: ['$b', '$c']}}]}"),
-        getExpCtx());
-
-    auto [metaOnlyMatch, remainingMatch] =
-        dynamic_cast<DocumentSourceInternalUnpackBucket*>(unpack.get())
-            ->splitMatchOnMetaAndRename(matchToSplit.get());
-
-    // Can split and rename when the $match includes several field references.
-    ASSERT_TRUE(metaOnlyMatch);
-    ASSERT_BSONOBJ_EQ(fromjson("{$and: [{meta: {$type: [3]}}, {'meta.a': {$gte: 0}}, "
-                               "{'meta.b': {$type: [4]}}]}"),
-                      metaOnlyMatch->getQuery());
-    ASSERT_TRUE(remainingMatch);
-    ASSERT_BSONOBJ_EQ(fromjson("{a: {$in: ['$b', '$c']}}"), remainingMatch->getQuery());
-}
-
 TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename, OptimizeSplitsMatchAndMapsControlPredicates) {
     auto unpack = fromjson(
         "{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'myMeta', "
@@ -294,6 +90,30 @@ TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename, OptimizeMovesMetaMatchBefo
 }
 
 TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename,
+       OptimizeDoesNotMoveMetaMatchBeforeUnpackWithExclusionOnMeta) {
+    auto unpack = fromjson(
+        "{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'myMeta', "
+        "bucketMaxSpanSeconds: 3600}}");
+    auto unpackExcluded = fromjson(
+        "{$_internalUnpackBucket: { include: ['_id', 'data'], timeField: 'foo', metaField: "
+        "'myMeta', "
+        "bucketMaxSpanSeconds: 3600}}");
+    auto pipeline = Pipeline::parse(makeVector(unpack,
+                                               fromjson("{$project: {data: 1}}"),
+                                               fromjson("{$match: {myMeta: {$gte: 0}}}")),
+                                    getExpCtx());
+    ASSERT_EQ(3u, pipeline->getSources().size());
+
+    pipeline->optimizePipeline();
+
+    // The $match on meta is not moved before $_internalUnpackBucket since the field is excluded.
+    auto serialized = pipeline->serializeToBson();
+    ASSERT_EQ(2u, serialized.size());
+    ASSERT_BSONOBJ_EQ(unpackExcluded, serialized[0]);
+    ASSERT_BSONOBJ_EQ(fromjson("{$match: {myMeta: {$gte: 0}}}"), serialized[1]);
+}
+
+TEST_F(InternalUnpackBucketSplitMatchOnMetaAndRename,
        OptimizeDoesNotErrorOnFailedSplitOfMetaMatch) {
     auto unpack = fromjson(
         "{$_internalUnpackBucket: { exclude: [], timeField: 'foo', metaField: 'myMeta', "
diff --git a/src/mongo/db/pipeline/document_source_match.cpp b/src/mongo/db/pipeline/document_source_match.cpp
index c0ee9639ff5..10346eb6a46 100644
--- a/src/mongo/db/pipeline/document_source_match.cpp
+++ b/src/mongo/db/pipeline/document_source_match.cpp
@@ -416,12 +416,6 @@ DocumentSourceMatch::splitSourceBy(const std::set<std::string>& fields,
 }
 
 pair<intrusive_ptr<DocumentSourceMatch>, intrusive_ptr<DocumentSourceMatch>>
-DocumentSourceMatch::extractMatchOnFieldsAndRemainder(const std::set<std::string>& fields,
-                                                      const StringMap<std::string>& renames) && {
-    return std::move(*this).splitSourceByFunc(fields, renames, expression::isOnlyDependentOn);
-}
-
-pair<intrusive_ptr<DocumentSourceMatch>, intrusive_ptr<DocumentSourceMatch>>
 DocumentSourceMatch::splitSourceByFunc(const std::set<std::string>& fields,
                                        const StringMap<std::string>& renames,
                                        expression::ShouldSplitExprFunc func) && {
diff --git a/src/mongo/db/pipeline/document_source_match.h b/src/mongo/db/pipeline/document_source_match.h
index 2d9c0c712dd..241b0b265f3 100644
--- a/src/mongo/db/pipeline/document_source_match.h
+++ b/src/mongo/db/pipeline/document_source_match.h
@@ -200,23 +200,6 @@ public:
     std::pair<boost::intrusive_ptr<DocumentSourceMatch>, boost::intrusive_ptr<DocumentSourceMatch>>
     splitSourceBy(const std::set<std::string>& fields, const StringMap<std::string>& renames) &&;
 
-    /**
-     * Attempt to split this $match into two stages, where the first is ONLY dependent upon paths
-     * from 'fields', and where applying them in sequence is equivalent to applying this stage once.
-     *
-     * Will return two intrusive_ptrs to new $match stages, where the first pointer is dependent on
-     * 'fields' and the second is the remainder. Either pointer may be null, so be sure to check the
-     * return value.
-     *
-     * The 'renames' structure maps from a field to an alias that should be used in the dependent
-     * portion of the match. For example, suppose that we split by fields "a" with the rename "a" =>
-     * "c". The match {$match: {a: "foo", b: "bar", z: "baz"}} will split into {$match: {c: "foo"}}
-     * and {$match: {b: "bar", z: "baz"}}.
-     */
-    std::pair<boost::intrusive_ptr<DocumentSourceMatch>, boost::intrusive_ptr<DocumentSourceMatch>>
-    extractMatchOnFieldsAndRemainder(const std::set<std::string>& fields,
-                                     const StringMap<std::string>& renames) &&;
-
     boost::optional<DistributedPlanLogic> distributedPlanLogic() final {
         return boost::none;
     }
diff --git a/src/mongo/db/timeseries/timeseries_commands_conversion_helper.cpp b/src/mongo/db/timeseries/timeseries_commands_conversion_helper.cpp
index 961344afd0e..abe386549c5 100644
--- a/src/mongo/db/timeseries/timeseries_commands_conversion_helper.cpp
+++ b/src/mongo/db/timeseries/timeseries_commands_conversion_helper.cpp
@@ -124,6 +124,10 @@ CreateIndexesCommand makeTimeseriesCreateIndexesCommand(OperationContext* opCtx,
                 // computed fields.
                 bool haveComputedMetaField = false;
 
+                // partialFilterExpression is evaluated against a collection, so there are no
+                // exclusions
+                bool includeMetaField = options.getMetaField().has_value();
+
                 // As part of building the index, we verify that the collection does not contain
                 // any mixed-schema buckets. So by the time the index is visible to the query
                 // planner, this will be true.
@@ -135,6 +139,7 @@ CreateIndexesCommand makeTimeseriesCreateIndexesCommand(OperationContext* opCtx,
                                                   collationMatchesDefault,
                                                   pred,
                                                   haveComputedMetaField,
+                                                  includeMetaField,
                                                   assumeNoMixedSchemaData,
                                                   BucketSpec::IneligiblePredicatePolicy::kError);
                 builder.append(IndexDescriptor::kPartialFilterExprFieldName, bucketPred);