diff --git a/src/mongo/db/pipeline/expression.cpp b/src/mongo/db/pipeline/expression.cpp
index 253e130c227..634a69111cf 100644
--- a/src/mongo/db/pipeline/expression.cpp
+++ b/src/mongo/db/pipeline/expression.cpp
@@ -4369,6 +4369,51 @@ const char* ExpressionPow::getOpName() const {
     return "$pow";
 }

+/* ------------------------- ExpressionInternalOwningShard ------------------------------ */
+
+Value ExpressionInternalOwningShard::evaluate(const Document& root, Variables* variables) const {
+    Value input(_children[0]->evaluate(root, variables));
+
+    if (input.nullish()) {
+        return Value(BSONNULL);
+    }
+
+    uassert(6868600, "$_internalOwningShard can only be used if pushed down to a shard", !_expCtx->inMongos);
+
+    uassert(6868686,
+            str::stream() << "The argument to $_intenralOwningShard must be a string, but was of type: "
+                          << typeName(input.getType()),
+            input.isString());
+
+    // TODO collectionless agg?
+    // TODO if we don't take any input and just do the primary nss then will there be confusion with union and join
+    // TODO any way to mark an expression as can't be pushed down to agg.
+    // lookups and unions
+
+    // Allways use the tenantId from the original query.
+    NamespaceString ns(_expCtx->ns.tenantId(), input.getStringData());
+    // check involved namespaces to avoid auth check without having security problem of using this
+    // on a collection that you don't have read access to.
+    uassert(68686001, "can only get sharding info for a collection involved in the query",
+            expCtx->isInvolvedForeignNamespace(ns) || ns == expCtx->ns);
+
+    auto css = CollectionShardingState::getSharedForLockFreeReads(_expCtx->opCtx, ns);
+
+    // Will throw StaleConfig if we wouldn't be able to get a shardVerions.
+    css.checkShardVersionOrThrow(_expCtx->opCtx);
+
+    BSONObjBuilder bob;
+
+    css.appendShardVersion(&bob);
+
+    return Document(bob.obj());
+}
+
+REGISTER_STABLE_EXPRESSION(_internalOwningShard, ExpressionInternalOwningShard::parse);
+const char* ExpressionReverseArray::getOpName() const {
+    return "$_internalOwningShard";
+}
+
 /* ------------------------- ExpressionRange ------------------------------ */

 Value ExpressionRange::evaluate(const Document& root, Variables* variables) const {
diff --git a/src/mongo/db/pipeline/expression.h b/src/mongo/db/pipeline/expression.h
index a406d233c03..701cef297f2 100644
--- a/src/mongo/db/pipeline/expression.h
+++ b/src/mongo/db/pipeline/expression.h
@@ -2580,6 +2580,23 @@ private:
     const char* getOpName() const final;
 };

+class ExpressionInternalOwningShard final : public ExpressionFixedArity<ExpressionReverseArray, 1> {
+public:
+    explicit ExpressionInternalOwningShard(ExpressionContext* const expCtx)
+        : ExpressionFixedArity<ExpressionInternalOwningShard, 1>(expCtx) {}
+
+    Value evaluate(const Document& root, Variables* variables) const final;
+    const char* getOpName() const final;
+
+    void acceptVisitor(ExpressionMutableVisitor* visitor) final {
+        return visitor->visit(this);
+    }
+
+    void acceptVisitor(ExpressionConstVisitor* visitor) const final {
+        return visitor->visit(this);
+    }
+};
+

 class ExpressionRange final : public ExpressionRangedArity<ExpressionRange, 2, 3> {
 public:
diff --git a/src/mongo/db/pipeline/expression_context.h b/src/mongo/db/pipeline/expression_context.h
index 8629311709a..dbb6c214add 100644
--- a/src/mongo/db/pipeline/expression_context.h
+++ b/src/mongo/db/pipeline/expression_context.h
@@ -283,6 +283,17 @@ public:
         return it->second;
     };

+    /**
+     * Checks to see if nss is in the list of resolvedNamespaces from the originating query.
+     */
+    bool isInvolvedForeignNamespace(const NamespaceString& nss) const {
+        // TODO we could totally just remove the invariant of the above function or make
+        // resolvedNamespaces public to avoid having to keep writing slightly different getter
+        // methods.
+        auto it = _resolvedNamespaces.find(nss.coll());
+        return it != _resolvedNamespaces.end();
+    }
+
     /**
      * Returns true if there are no namespaces in the query other than the namespace the query was
      * issued against. eg if there is no $out, $lookup ect. If namespaces have not yet been resolved