diff --git a/zerver/tests/test_message_fetch.py b/zerver/tests/test_message_fetch.py
index 5cfcad040e..fdcdb875fe 100644
--- a/zerver/tests/test_message_fetch.py
+++ b/zerver/tests/test_message_fetch.py
@@ -2505,6 +2505,22 @@ class GetOldMessagesTest(ZulipTestCase):
             '<p><span class="highlight">KEYWORDMATCH</span> and should work</p>',
         )
 
+        narrow = [
+            dict(operator="search", operand="KEYWORDMATCH"),
+            dict(operator="search", operand="work"),
+        ]
+
+        raw_params = dict(msg_ids=msg_ids, narrow=narrow)
+        params = {k: orjson.dumps(v).decode() for k, v in raw_params.items()}
+        result = self.client_get("/json/messages/matches_narrow", params)
+        messages = self.assert_json_success(result)["messages"]
+        self.assert_length(list(messages.keys()), 1)
+        message = messages[str(good_id)]
+        self.assertEqual(
+            message["match_content"],
+            '<p><span class="highlight">KEYWORDMATCH</span> and should <span class="highlight">work</span></p>',
+        )
+
     @override_settings(USING_PGROONGA=False)
     def test_get_messages_with_search(self) -> None:
         self.login("cordelia")
diff --git a/zerver/views/message_fetch.py b/zerver/views/message_fetch.py
index 0752feb82a..6c5ac66eeb 100644
--- a/zerver/views/message_fetch.py
+++ b/zerver/views/message_fetch.py
@@ -11,8 +11,8 @@ from zerver.context_processors import get_valid_realm_from_request
 from zerver.lib.exceptions import JsonableError, MissingAuthenticationError
 from zerver.lib.message import get_first_visible_message_id, messages_for_ids
 from zerver.lib.narrow import (
-    NarrowBuilder,
     OptionalNarrowListT,
+    add_narrow_conditions,
     fetch_messages,
     is_spectator_compatible,
     is_web_public_narrow,
@@ -246,7 +246,7 @@ def messages_in_narrow_backend(
     # This query is limited to messages the user has access to because they
     # actually received them, as reflected in `zerver_usermessage`.
     query = (
-        select(column("message_id", Integer), topic_column_sa(), column("rendered_content", Text))
+        select(column("message_id", Integer))
         .where(
             and_(
                 column("user_profile_id", Integer) == literal(user_profile.id),
@@ -263,22 +263,28 @@ def messages_in_narrow_backend(
         )
     )
 
-    builder = NarrowBuilder(user_profile, column("message_id", Integer), user_profile.realm)
-    if narrow is not None:
-        for term in narrow:
-            query = builder.add_term(query, term)
+    inner_msg_id_col = column("message_id", Integer)
+    query, is_search = add_narrow_conditions(
+        user_profile=user_profile,
+        inner_msg_id_col=inner_msg_id_col,
+        query=query,
+        narrow=narrow,
+        is_web_public_query=False,
+        realm=user_profile.realm,
+    )
+
+    if not is_search:
+        # `add_narrow_conditions` adds the following columns only if narrow has search operands.
+        query = query.add_columns(topic_column_sa(), column("rendered_content", Text))
 
     search_fields = {}
     with get_sqlalchemy_connection() as sa_conn:
-        for row in sa_conn.execute(query).fetchall():
-            message_id = row._mapping["message_id"]
-            topic_name = row._mapping[DB_TOPIC_NAME]
-            rendered_content = row._mapping["rendered_content"]
-            if "content_matches" in row._mapping:
-                content_matches = row._mapping["content_matches"]
-                topic_matches = row._mapping["topic_matches"]
-            else:
-                content_matches = topic_matches = []
+        for row in sa_conn.execute(query).mappings():
+            message_id = row["message_id"]
+            topic_name: str = row[DB_TOPIC_NAME]
+            rendered_content: str = row["rendered_content"]
+            content_matches = row.get("content_matches", [])
+            topic_matches = row.get("topic_matches", [])
             search_fields[str(message_id)] = get_search_fields(
                 rendered_content,
                 topic_name,