feat(replays): Introduce more memory efficient search system (#54527)

## Why Is This Necessary?

**Background**

Replays are aggregated by their replay_id. Replay_id is high
cardinality. On our largest customers these aggregations OOM no matter
what your query contains. For example, this query...

```sql
SELECT min(any_column) FROM replays GROUP BY replay_id LIMIT 1
```

...will fail on large projects. So we append two settings to the end of
our query.

```sql
SETTINGS max_rows_to_group_by='1000000', group_by_overflow_mode='any'
```

The details don't matter too much but basically we trade accuracy for
resource usage. Does this solve the problem? In the query I showed above
- yes. However, more complex queries still breach the limit. So what can
we do to solve this problem?

**Proposal**

For queries which **require** aggregation to answer filter and sort
questions, we must do two things:

1. `SELECT` as few columns as possible. The fewer column aggregations we
hold in-memory the less likely we are to breach our limit.
2. When our query asks us to operate on a memory intensive column, we
avoid pulling raw column data into an aggregation set and instead deduce
the outcome of a query with a memory-sensitive proxy.
- `replays/usecases/query/conditions/aggregate.py` is a good source to
see how this is performed.

**Solution**

At its core this PR adds two new, useful features. We query an
aggregated set with a limited `SELECT` statement (this means we have to
run a _second_ aggregation query to back-fill the missing data) and we
filter our dataset with memory-sensitive search aggregations.

On the first point, this means we're following the lead of our sub-query
optimization except we can answer every search/sort operation a user is
able to ask.

## What Does This Pull Actually Accomplish

1. Minimizes the `SELECT` in our searching and sorting aggregation
query. We only select by `replay_id`.
2. We no longer aggregate columns into memory. Instead of we analyze
values on a row-wise basis and aggregate a bit representing the outcome
of the condition. `1` for true and `0` for false. If a condition returns
a sum greater than `0` we know at least one row (for a given aggregation
key) contained a truthy response.
- Given a query with 10 filters we would aggregate 10 condition bits
(UInt64). The number of condition bits is multiplied by the number of
unique aggregation keys (maximum of 1 million). So the total memory
usage of a complex query is now _80 MB_ (not considering the size of
replay_id).
4. We utilize indexes where available which makes our look-ups
significantly faster.
5. We separate our sorting, searching, and selection configurations from
one another. They each serve different purposes and by configuring each
independently we can have fine grained control over what is queried, how
we allow it to be filtered, and how we display it to the user.

---------

Co-authored-by: Josh Ferge <josh.ferge@sentry.io>
Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>

src/sentry/conf/server.py

@@ -1602,6 +1602,8 @@ SENTRY_FEATURES = {
     "organizations:session-replay-issue-emails": False,
     "organizations:session-replay-weekly-email": False,
     "organizations:session-replay-trace-table": False,
+    # Enable optimized serach feature.
+    "organizations:session-replay-optimized-search": False,
     # Enable rage click and dead click columns in replay list.
     "organizations:replay-rage-click-dead-click-columns": False,
     # Enable experimental error and rage/dead click cards in replay list.

src/sentry/features/__init__.py

@@ -192,6 +192,7 @@ default_manager.add("organizations:session-replay-slack-new-issue", Organization
 default_manager.add("organizations:session-replay-weekly-email", OrganizationFeature, FeatureHandlerStrategy.INTERNAL)
 default_manager.add("organizations:session-replay-issue-emails", OrganizationFeature, FeatureHandlerStrategy.INTERNAL)
 default_manager.add("organizations:session-replay-trace-table", OrganizationFeature, FeatureHandlerStrategy.REMOTE)
+default_manager.add("organizations:session-replay-optimized-search", OrganizationFeature, FeatureHandlerStrategy.REMOTE)
 default_manager.add("organizations:set-grouping-config", OrganizationFeature, FeatureHandlerStrategy.INTERNAL)
 default_manager.add("organizations:slack-overage-notifications", OrganizationFeature, FeatureHandlerStrategy.REMOTE)
 default_manager.add("organizations:sdk-crash-detection", OrganizationFeature, FeatureHandlerStrategy.INTERNAL)

src/sentry/replays/lib/new_query/conditions.py

@@ -0,0 +1,271 @@
+"""Condition visitor module.
+
+Each visitor (class) represents a unique data-type that is being acted upon.  For example, a class
+titled "StringScalar" points to a column, function, or aggregation whose end product is of type
+String.  The expression can then be interacted with using generic string operations.
+
+Visitor classes are polymorphic on their data type, their methods are polymorphic on the operator,
+and their values are polymorphic on data-type and operator.
+
+It's important to note that condition visitors in this module (and elsewhere) do not assume the
+origin of the data.  The same visitor may be applied to aggregated output as easily as its applied
+to rows.
+
+Every condition visitor must define a method for every operator supported by the caller.  A full
+list of supported operations can be found in the "GenericBase" visitor.
+"""
+from __future__ import annotations
+
+from typing import Any, TypeVar
+from uuid import UUID
+
+from snuba_sdk import Condition, Function, Identifier, Lambda, Op
+from snuba_sdk.expressions import Expression
+
+from sentry.replays.lib.new_query.errors import OperatorNotSupported
+from sentry.replays.lib.new_query.utils import to_uuid, to_uuids
+
+T = TypeVar("T")
+
+
+class GenericBase:
+    @staticmethod
+    def visit_eq(expression: Expression, value: Any) -> Condition:
+        not_supported()
+
+    @staticmethod
+    def visit_neq(expression: Expression, value: Any) -> Condition:
+        not_supported()
+
+    @staticmethod
+    def visit_gt(expression: Expression, value: Any) -> Condition:
+        not_supported()
+
+    @staticmethod
+    def visit_gte(expression: Expression, value: Any) -> Condition:
+        not_supported()
+
+    @staticmethod
+    def visit_lt(expression: Expression, value: Any) -> Condition:
+        not_supported()
+
+    @staticmethod
+    def visit_lte(expression: Expression, value: Any) -> Condition:
+        not_supported()
+
+    @staticmethod
+    def visit_match(expression: Expression, value: Any) -> Condition:
+        not_supported()
+
+    @staticmethod
+    def visit_not_match(expression: Expression, value: Any) -> Condition:
+        not_supported()
+
+    @staticmethod
+    def visit_in(expression: Expression, value: list[Any]) -> Condition:
+        not_supported()
+
+    @staticmethod
+    def visit_not_in(expression: Expression, value: list[Any]) -> Condition:
+        not_supported()
+
+
+class BooleanScalar(GenericBase):
+    """Boolean scalar condition class."""
+
+    @staticmethod
+    def visit_eq(expression: Expression, value: bool) -> Condition:
+        return Condition(expression, Op.EQ, value)
+
+    @staticmethod
+    def visit_neq(expression: Expression, value: bool) -> Condition:
+        return Condition(expression, Op.NEQ, value)
+
+
+class IntegerScalar(GenericBase):
+    """Integer scalar condition class."""
+
+    @staticmethod
+    def visit_eq(expression: Expression, value: int) -> Condition:
+        return Condition(expression, Op.EQ, value)
+
+    @staticmethod
+    def visit_neq(expression: Expression, value: int) -> Condition:
+        return Condition(expression, Op.NEQ, value)
+
+    @staticmethod
+    def visit_gt(expression: Expression, value: int) -> Condition:
+        return Condition(expression, Op.GT, value)
+
+    @staticmethod
+    def visit_gte(expression: Expression, value: int) -> Condition:
+        return Condition(expression, Op.GTE, value)
+
+    @staticmethod
+    def visit_lt(expression: Expression, value: int) -> Condition:
+        return Condition(expression, Op.LT, value)
+
+    @staticmethod
+    def visit_lte(expression: Expression, value: int) -> Condition:
+        return Condition(expression, Op.LTE, value)
+
+    @staticmethod
+    def visit_in(expression: Expression, value: list[int]) -> Condition:
+        return Condition(expression, Op.IN, value)
+
+    @staticmethod
+    def visit_not_in(expression: Expression, value: list[int]) -> Condition:
+        return Condition(expression, Op.NOT_IN, value)
+
+
+class StringScalar(GenericBase):
+    """String scalar condition class."""
+
+    @staticmethod
+    def visit_eq(expression: Expression, value: str) -> Condition:
+        return Condition(expression, Op.EQ, value)
+
+    @staticmethod
+    def visit_neq(expression: Expression, value: str) -> Condition:
+        return Condition(expression, Op.NEQ, value)
+
+    @staticmethod
+    def visit_match(expression: Expression, value: str) -> Condition:
+        v = f"(?i){value[1:-1]}"
+        return Condition(Function("match", parameters=[expression, v]), Op.EQ, 1)
+
+    @staticmethod
+    def visit_not_match(expression: Expression, value: str) -> Condition:
+        v = f"(?i){value[1:-1]}"
+        return Condition(Function("match", parameters=[expression, v]), Op.EQ, 0)
+
+    @staticmethod
+    def visit_in(expression: Expression, value: list[str]) -> Condition:
+        return Condition(expression, Op.IN, value)
+
+    @staticmethod
+    def visit_not_in(expression: Expression, value: list[str]) -> Condition:
+        return Condition(expression, Op.NOT_IN, value)
+
+
+class UUIDScalar(GenericBase):
+    """UUID scalar condition class."""
+
+    @staticmethod
+    def visit_eq(expression: Expression, value: UUID) -> Condition:
+        return Condition(expression, Op.EQ, to_uuid(value))
+
+    @staticmethod
+    def visit_neq(expression: Expression, value: UUID) -> Condition:
+        return Condition(expression, Op.NEQ, to_uuid(value))
+
+    @staticmethod
+    def visit_in(expression: Expression, value: list[UUID]) -> Condition:
+        return Condition(expression, Op.IN, to_uuids(value))
+
+    @staticmethod
+    def visit_not_in(expression: Expression, value: list[UUID]) -> Condition:
+        return Condition(expression, Op.NOT_IN, to_uuids(value))
+
+
+class IPv4Scalar(GenericBase):
+    """IPv4 scalar condition class."""
+
+    @staticmethod
+    def visit_eq(expression: Expression, value: str) -> Condition:
+        return Condition(expression, Op.EQ, Function("IPv4StringToNum", parameters=[value]))
+
+    @staticmethod
+    def visit_neq(expression: Expression, value: str) -> Condition:
+        return Condition(expression, Op.NEQ, Function("IPv4StringToNum", parameters=[value]))
+
+    @staticmethod
+    def visit_in(expression: Expression, value: list[str]) -> Condition:
+        values = [Function("IPv4StringToNum", parameters=[v]) for v in value]
+        return Condition(expression, Op.IN, values)
+
+    @staticmethod
+    def visit_not_in(expression: Expression, value: list[str]) -> Condition:
+        values = [Function("IPv4StringToNum", parameters=[v]) for v in value]
+        return Condition(expression, Op.NOT_IN, values)
+
+
+class GenericArray(GenericBase):
+    @staticmethod
+    def visit_eq(expression: Expression, value: Any) -> Condition:
+        return Condition(Function("has", parameters=[expression, value]), Op.EQ, 1)
+
+    @staticmethod
+    def visit_neq(expression: Expression, value: Any) -> Condition:
+        return Condition(Function("has", parameters=[expression, value]), Op.EQ, 0)
+
+    @staticmethod
+    def visit_in(expression: Expression, value: list[Any]) -> Condition:
+        return Condition(Function("hasAny", parameters=[expression, value]), Op.EQ, 1)
+
+    @staticmethod
+    def visit_not_in(expression: Expression, value: list[Any]) -> Condition:
+        return Condition(Function("hasAny", parameters=[expression, value]), Op.EQ, 0)
+
+
+class IntegerArray(GenericArray):
+    """Integer array condition class."""
+
+
+class StringArray(GenericArray):
+    """String array condition class."""
+
+    @staticmethod
+    def visit_match(expression: Expression, value: str) -> Condition:
+        v = f"(?i){value[1:-1]}"
+        return Condition(
+            Function(
+                "arrayExists",
+                parameters=[
+                    Lambda(["item"], Function("match", parameters=[Identifier("item"), v])),
+                    expression,
+                ],
+            ),
+            Op.EQ,
+            1,
+        )
+
+    @staticmethod
+    def visit_not_match(expression: Expression, value: str) -> Condition:
+        v = f"(?i){value[1:-1]}"
+        return Condition(
+            Function(
+                "arrayExists",
+                parameters=[
+                    Lambda(["item"], Function("match", parameters=[Identifier("item"), v])),
+                    expression,
+                ],
+            ),
+            Op.EQ,
+            0,
+        )
+
+
+class UUIDArray(GenericArray):
+    """UUID array condition class."""
+
+    @staticmethod
+    def visit_eq(expression: Expression, value: UUID) -> Condition:
+        return GenericArray.visit_eq(expression, to_uuid(value))
+
+    @staticmethod
+    def visit_neq(expression: Expression, value: UUID) -> Condition:
+        return GenericArray.visit_neq(expression, to_uuid(value))
+
+    @staticmethod
+    def visit_in(expression: Expression, value: list[UUID]) -> Condition:
+        return GenericArray.visit_in(expression, to_uuids(value))
+
+    @staticmethod
+    def visit_not_in(expression: Expression, value: list[UUID]) -> Condition:
+        return GenericArray.visit_not_in(expression, to_uuids(value))
+
+
+def not_supported() -> None:
+    """Raise not supported exception."""
+    raise OperatorNotSupported("Not supported.")

src/sentry/replays/lib/new_query/errors.py

@@ -0,0 +1,6 @@
+class OperatorNotSupported(Exception):
+    pass
+
+
+class CouldNotParseValue(Exception):
+    pass

src/sentry/replays/lib/new_query/fields.py

@@ -0,0 +1,148 @@
+"""Field interface module.
+
+Fields are the contact point to the wider search ecosystem at Sentry.  They act as an interface
+to bridge the two worlds.  Fields are responsible for accepting a SearchFilter, verifying its
+operator is valid, verifying the value's data-type is valid, and finally calling into the
+condition system to return a condition clause for the query.
+
+Fields also contain the means to determine the expression being filtered.  Whereas a Condition
+visitor doesn't care what expression is given to it the Field instance certainly does.  One of its
+core responsibilities is to pass the correct expression to be filtered against.
+
+Fields are polymorphic on the source they target and the data-type of the field in the API
+response.  Note just because a field appears as an array or as a scalar does not mean it is
+filtered in that way.  The field's job is to translate the display format to the expression
+format.
+"""
+from __future__ import annotations
+
+import datetime
+from typing import Callable, Generic, Type
+from uuid import UUID
+
+from snuba_sdk import Column, Condition, Function
+from snuba_sdk.expressions import Expression
+
+from sentry.api.event_search import SearchFilter
+from sentry.replays.lib.new_query.conditions import GenericBase, T
+
+
+class BaseField(Generic[T]):
+    def __init__(self, parse: Callable[[str], T], query: Type[GenericBase]) -> None:
+        self.parse = parse
+        self.query = query
+
+    def apply(self, search_filter: SearchFilter) -> Condition:
+        raise NotImplementedError
+
+    def _apply_wildcard(self, expression: Expression, operator: str, value: T) -> Condition:
+        if operator == "=":
+            visitor = self.query.visit_match
+        elif operator == "!=":
+            visitor = self.query.visit_not_match
+        else:
+            raise Exception(f"Unsupported wildcard search operator: '{operator}'")
+
+        return visitor(expression, value)
+
+    def _apply_composite(self, expression: Expression, operator: str, value: list[T]) -> Condition:
+        if operator == "IN":
+            visitor = self.query.visit_in
+        elif operator == "NOT IN":
+            visitor = self.query.visit_not_in
+        else:
+            raise Exception(f"Unsupported composite search operator: '{operator}'")
+
+        return visitor(expression, value)
+
+    def _apply_scalar(self, expression: Expression, operator: str, value: T) -> Condition:
+        if operator == "=":
+            visitor = self.query.visit_eq
+        elif operator == "!=":
+            visitor = self.query.visit_neq
+        elif operator == ">":
+            visitor = self.query.visit_gt
+        elif operator == ">=":
+            visitor = self.query.visit_gte
+        elif operator == "<":
+            visitor = self.query.visit_lt
+        elif operator == "<=":
+            visitor = self.query.visit_lte
+        else:
+            raise Exception(f"Unsupported search operator: '{operator}'")
+
+        return visitor(expression, value)
+
+
+class ColumnField(BaseField[T]):
+    """Column fields target one column."""
+
+    def __init__(
+        self, column_name: str, parse_fn: Callable[[str], T], query_type: Type[GenericBase]
+    ) -> None:
+        self.column_name = column_name
+        self.parse = parse_fn
+        self.query = query_type
+
+    def apply(self, search_filter: SearchFilter) -> Condition:
+        """Apply a search operation against any named expression.
+
+        A named expression can be a column name or an expression alias.
+        """
+        operator = search_filter.operator
+        value = search_filter.value.value
+
+        # We need to check if the value is a scalar to determine the path we should follow.  This
+        # is not as simple as asking for isinstance(value, list).  Array values are provided to us
+        # as "Sequence" types.  Sequence[str] is the same type as str.  So we can't check for
+        # Sequence in the isinstance check.  We have to explicitly check for all the possible
+        # scalar values and then use the else to apply array based filtering techniques.
+        if isinstance(value, (str, int, float, datetime.datetime)):
+            # We don't care that the SearchFilter typed the value for us. We'll determine what we
+            # want to parse it to.  There's too much polymorphism if we have to consider coercing
+            # this data-type to that data-type in the parse step.
+            parsed_value = self.parse(str(value))
+
+            if search_filter.value.is_wildcard():
+                applicable = self._apply_wildcard
+            else:
+                applicable = self._apply_scalar
+
+            return applicable(self.expression, operator, parsed_value)
+        else:
+            # Again the types contained within the list are coerced to string to be re-coerced
+            # back into their correct data-type.
+            parsed_values = [self.parse(str(v)) for v in value]
+            return self._apply_composite(self.expression, operator, parsed_values)
+
+    @property
+    def expression(self) -> Column:
+        return Column(self.column_name)
+
+
+class StringColumnField(ColumnField[str]):
+    """String type conditional column field."""
+
+
+class UUIDColumnField(ColumnField[UUID]):
+    """UUID type conditional column field."""
+
+
+class CountField(ColumnField[int]):
+    @property
+    def expression(self) -> Function:
+        return Function("count", parameters=[Column(self.column_name)])
+
+
+class SumField(ColumnField[int]):
+    @property
+    def expression(self) -> Function:
+        return Function("sum", parameters=[Column(self.column_name)])
+
+
+class SumLengthField(ColumnField[int]):
+    @property
+    def expression(self) -> Function:
+        return Function(
+            "sum", parameters=[Function("length", parameters=[Column(self.column_name)])]
+        )

src/sentry/replays/lib/new_query/parsers.py

@@ -0,0 +1,36 @@
+"""Parser module.
+
+Functions in this module coerce external types to internal types.  Else they die.
+"""
+import uuid
+
+from sentry.replays.lib.new_query.errors import CouldNotParseValue
+
+
+def parse_float(value: str) -> float:
+    """Coerce to float or fail."""
+    try:
+        return float(value)
+    except ValueError:
+        raise CouldNotParseValue("Failed to parse float.")
+
+
+def parse_int(value: str) -> int:
+    """Coerce to int or fail."""
+    return int(parse_float(value))
+
+
+def parse_str(value: str) -> str:
+    """Coerce to str or fail."""
+    return value
+
+
+def parse_uuid(value: str) -> uuid.UUID:
+    try:
+        return uuid.UUID(value)
+    except ValueError:
+        # Return an empty uuid. This emulates current behavior where inability to parse a UUID
+        # leads to an empty result-set rather than an error.
+        #
+        # TODO: Probably raise an error here...
+        return uuid.UUID("00000000000000000000000000000000")

src/sentry/replays/lib/new_query/utils.py

@@ -0,0 +1,51 @@
+"""Query utility module."""
+from __future__ import annotations
+
+from uuid import UUID
+
+from snuba_sdk import Condition, Function, Op
+
+
+def to_uuid(value: UUID) -> Function:
+    return Function("toUUID", parameters=[str(value)])
+
+
+def to_uuids(value: list[UUID]) -> list[Function]:
+    return [to_uuid(v) for v in value]
+
+
+def contains(condition: Condition) -> Condition:
+    """Return true if any of the rows in the aggregation set match the condition."""
+    return Condition(
+        Function("sum", parameters=[translate_condition_to_function(condition)]), Op.NEQ, 0
+    )
+
+
+def does_not_contain(condition: Condition) -> Condition:
+    """Return true if none of the rows in the aggregation set match the condition."""
+    return Condition(
+        Function("sum", parameters=[translate_condition_to_function(condition)]), Op.EQ, 0
+    )
+
+
+# Work-around for https://github.com/getsentry/snuba-sdk/issues/115
+def translate_condition_to_function(condition: Condition) -> Function:
+    """Transforms infix operations to prefix operations."""
+    if condition.op == Op.EQ:
+        return Function("equals", parameters=[condition.lhs, condition.rhs])
+    elif condition.op == Op.NEQ:
+        return Function("notEquals", parameters=[condition.lhs, condition.rhs])
+    elif condition.op == Op.GT:
+        return Function("greater", parameters=[condition.lhs, condition.rhs])
+    elif condition.op == Op.GTE:
+        return Function("greaterOrEquals", parameters=[condition.lhs, condition.rhs])
+    elif condition.op == Op.LT:
+        return Function("less", parameters=[condition.lhs, condition.rhs])
+    elif condition.op == Op.LTE:
+        return Function("lessOrEquals", parameters=[condition.lhs, condition.rhs])
+    elif condition.op == Op.IN:
+        return Function("in", parameters=[condition.lhs, condition.rhs])
+    elif condition.op == Op.NOT_IN:
+        return Function("notIn", parameters=[condition.lhs, condition.rhs])
+    else:
+        raise Exception(f"Didn't understand operation: {condition.op}")

src/sentry/replays/query.py

@@ -23,6 +23,7 @@ from snuba_sdk import (
 from snuba_sdk.expressions import Expression
 from snuba_sdk.orderby import Direction, OrderBy
 
+from sentry import features
 from sentry.api.event_search import ParenExpression, SearchConfig, SearchFilter
 from sentry.models.organization import Organization
 from sentry.replays.lib.query import (
@@ -38,6 +39,7 @@ from sentry.replays.lib.query import (
     generate_valid_conditions,
     get_valid_sort_commands,
 )
+from sentry.replays.usecases.query import query_using_aggregated_search
 from sentry.utils.snuba import raw_snql_query
 
 MAX_PAGE_SIZE = 100
@@ -73,6 +75,19 @@ def query_replays_collection(
 
     paginators = make_pagination_values(limit, offset)
 
+    if features.has("organizations:session-replay-optimized-search", organization, actor=actor):
+        return query_using_aggregated_search(
+            fields,
+            search_filters,
+            environment,
+            sort,
+            paginators,
+            organization,
+            project_ids,
+            start,
+            end,
+        )
+
     # Attempt to eager return with subquery.
 
     try:

src/sentry/replays/usecases/query/__init__.py

@@ -0,0 +1,320 @@
+"""Query use-case module.
+
+For now, this is the search and sort entry-point.  Some of this code may be moved to
+replays/query.py when the pre-existing query module is deprecated.
+
+There are two important functions in this module: "search_filter_to_condition" and
+"query_using_aggregated_search".  "search_filter_to_condition" is responsible for transforming a
+SearchFilter into a Condition.  This is the only entry-point into the Field system.
+
+"query_using_aggregated_search" is the request processing engine.  It accepts raw data from an
+external source, makes decisions around what to query and when, and is responsible for returning
+intelligible output for the "post_process" module.  More information on its implementation can be
+found in the function.
+"""
+from __future__ import annotations
+
+from collections import namedtuple
+from datetime import datetime, timedelta
+from typing import Union, cast
+
+from rest_framework.exceptions import ParseError
+from snuba_sdk import (
+    And,
+    Column,
+    Condition,
+    Direction,
+    Entity,
+    Function,
+    Granularity,
+    Op,
+    Or,
+    OrderBy,
+    Query,
+    Request,
+)
+from snuba_sdk.expressions import Expression
+
+from sentry.api.event_search import ParenExpression, SearchFilter, SearchKey, SearchValue
+from sentry.models.organization import Organization
+from sentry.replays.lib.new_query.errors import CouldNotParseValue, OperatorNotSupported
+from sentry.replays.lib.new_query.fields import ColumnField
+from sentry.replays.usecases.query.fields import ComputedField, TagField
+from sentry.utils.snuba import raw_snql_query
+
+
+def handle_search_filters(
+    search_config: dict[str, Union[ColumnField, ComputedField, TagField]],
+    search_filters: list[Union[SearchFilter, str, ParenExpression]],
+) -> list[Condition]:
+    """Convert search filters to snuba conditions."""
+    result: list[Condition] = []
+    look_back = None
+    for search_filter in search_filters:
+        # SearchFilters are transformed into Conditions and appended to the result set.  If they
+        # are top level filters they are implicitly AND'ed in the WHERE/HAVING clause.  Otherwise
+        # explicit operators are used.
+        if isinstance(search_filter, SearchFilter):
+            try:
+                condition = search_filter_to_condition(search_config, search_filter)
+            except OperatorNotSupported:
+                raise ParseError(f"Invalid operator specified for `{search_filter.key.name}`")
+            except CouldNotParseValue:
+                raise ParseError(f"Could not parse value for `{search_filter.key.name}`")
+
+            if look_back == "AND":
+                look_back = None
+                attempt_compressed_condition(result, condition, And)
+            elif look_back == "OR":
+                look_back = None
+                attempt_compressed_condition(result, condition, Or)
+            else:
+                result.append(condition)
+        # ParenExpressions are recursively computed.  If more than one condition is returned then
+        # those conditions are AND'ed.
+        elif isinstance(search_filter, ParenExpression):
+            conditions = handle_search_filters(search_config, search_filter.children)
+            if len(conditions) < 2:
+                result.extend(conditions)
+            else:
+                result.append(And(conditions))
+        # String types are limited to AND and OR... I think?  In the case where its not a valid
+        # look-back it is implicitly ignored.
+        elif isinstance(search_filter, str):
+            look_back = search_filter
+
+    return result
+
+
+def attempt_compressed_condition(
+    result: list[Expression],
+    condition: Condition,
+    condition_type: Union[And, Or],
+):
+    """Unnecessary query optimization.
+
+    Improves legibility for query debugging. Clickhouse would flatten these nested OR statements
+    internally anyway.
+
+    (block OR block) OR block => (block OR block OR block)
+    """
+    if isinstance(result[-1], condition_type):
+        result[-1].conditions.append(condition)
+    else:
+        result.append(condition_type([result.pop(), condition]))
+
+
+def search_filter_to_condition(
+    search_config: dict[str, Union[ColumnField, ComputedField, TagField]],
+    search_filter: SearchFilter,
+) -> Condition:
+    # The field-name is whatever the API says it is.  We take it at face value.
+    field_name = search_filter.key.name
+
+    # If the field-name is in the search config then we can apply the search filter and return a
+    # result.  If its not then its a tag and the same operation is performed only with a few more
+    # steps.
+    field = search_config.get(field_name)
+    if isinstance(field, (ColumnField, ComputedField)):
+        return field.apply(search_filter)
+
+    if field is None:
+        # Tags are represented with an "*" field by convention.  We could name it `tags` and
+        # update our search config to point to this field-name.
+        field = cast(TagField, search_config["*"])
+
+    # Tags that are namespaced are stripped.
+    if field_name.startswith("tags["):
+        field_name = field_name[5:-1]
+
+    # The field_name in this case does not represent a column_name but instead it represents a
+    # dynamic value in the tags.key array.  For this reason we need to pass it into our "apply"
+    # function.
+    return field.apply(field_name, search_filter)
+
+
+# Everything below here will move to replays/query.py once we deprecate the old query behavior.
+# Leaving it here for now so this is easier to review/remove.
+from sentry.replays.usecases.query.configs.aggregate import search_config as agg_search_config
+from sentry.replays.usecases.query.configs.aggregate_sort import sort_config as agg_sort_config
+
+Paginators = namedtuple("Paginators", ("limit", "offset"))
+
+
+def query_using_aggregated_search(
+    fields: list[str],
+    search_filters: list[Union[SearchFilter, str, ParenExpression]],
+    environments: list[str],
+    sort: str | None,
+    pagination: Paginators | None,
+    organization: Organization | None,
+    project_ids: list[int],
+    period_start: datetime,
+    period_stop: datetime,
+):
+    tenant_ids = _make_tenant_id(organization)
+
+    if sort is None:
+        sorting = [OrderBy(_get_sort_column("started_at"), Direction.DESC)]
+    elif sort.startswith("-"):
+        sorting = [OrderBy(_get_sort_column(sort[1:]), Direction.DESC)]
+    else:
+        sorting = [OrderBy(_get_sort_column(sort), Direction.ASC)]
+
+    # Environments is provided to us outside of the ?query= url parameter. It's stil filtered like
+    # the values in that parameter so let's shove it inside and process it like any other filter.
+    if environments:
+        search_filters.append(
+            SearchFilter(SearchKey("environment"), "IN", SearchValue(environments))
+        )
+
+    # First aggregation step.
+
+    simple_aggregation_query = make_simple_aggregation_query(
+        search_filters=search_filters,
+        orderby=sorting,
+        project_ids=project_ids,
+        period_start=period_start,
+        period_stop=period_stop,
+    )
+    if pagination:
+        simple_aggregation_query = simple_aggregation_query.set_limit(pagination.limit)
+        simple_aggregation_query = simple_aggregation_query.set_offset(pagination.offset)
+
+    # The simple aggregation query does not select by anything other than replay_id.  Every filter
+    # is applies is ephemeral and calculated for its own purpose.  These filters _should_ be
+    # optimized to be memory sensitive in cases where its required.
+    #
+    # This query will aggregate the entire data-set contained within the period's start and end
+    # times.
+    simple_aggregation_response = raw_snql_query(
+        Request(
+            dataset="replays",
+            app_id="replay-backend-web",
+            query=simple_aggregation_query,
+            tenant_ids=tenant_ids,
+        ),
+        "replays.query.browse_simple_aggregation",
+    )
+
+    # These replay_ids are ordered by the OrderBy expression in the query above.
+    replay_ids = [row["replay_id"] for row in simple_aggregation_response.get("data", [])]
+
+    # The final aggregation step.  Here we pass the replay_ids as the only filter.  In this step
+    # we select everything and use as much memory as we need to complete the operation.
+    #
+    # If this step runs out of memory your pagination size is about 1,000,000 rows too large.
+    # That's a joke.  This will complete very quickly at normal pagination sizes.
+    results = raw_snql_query(
+        Request(
+            dataset="replays",
+            app_id="replay-backend-web",
+            query=make_full_aggregation_query(
+                fields=fields,
+                replay_ids=replay_ids,
+                project_ids=project_ids,
+                period_start=period_start,
+                period_end=period_stop,
+            ),
+            tenant_ids=tenant_ids,
+        ),
+        "replays.query.browse_points",
+    )["data"]
+
+    # A weird snuba-ism.  You can't sort by an aggregation that is also present in the select.
+    # Even if the aggregations are different.  So we have to fallback to sorting on the
+    # application server.
+    #
+    # For example these are all examples of valid SQL that are not possible with Snuba.
+    #
+    #   SELECT any(os_name)
+    #   FROM replays_local
+    #   GROUP BY replay_id
+    #   ORDER BY any(os_name)
+    #
+    #   SELECT anyIf(os_name, notEmpty(os_name))
+    #   FROM replays_local
+    #   GROUP BY replay_id
+    #   ORDER BY any(os_name)
+    ordered_results = [None] * len(replay_ids)
+    replay_id_to_index = {replay_id: index for index, replay_id in enumerate(replay_ids)}
+    for result in results:
+        index = replay_id_to_index[result["replay_id"]]
+        ordered_results[index] = result
+
+    return ordered_results
+
+
+def make_simple_aggregation_query(
+    search_filters: list[Union[SearchFilter, str, ParenExpression]],
+    orderby: list[OrderBy],
+    project_ids: list[int],
+    period_start: datetime,
+    period_stop: datetime,
+) -> Query:
+    # This is our entry-point to the SearchFilter to Condition transformation process.  We do not
+    # filter at any other step in this process.
+    having: list[Condition] = handle_search_filters(agg_search_config, search_filters)
+
+    return Query(
+        match=Entity("replays"),
+        select=[Column("replay_id")],
+        where=[
+            Condition(Column("project_id"), Op.IN, project_ids),
+            Condition(Column("timestamp"), Op.LT, period_stop),
+            Condition(Column("timestamp"), Op.GTE, period_start),
+        ],
+        having=having,
+        orderby=orderby,
+        groupby=[Column("replay_id")],
+        granularity=Granularity(3600),
+    )
+
+
+def make_full_aggregation_query(
+    fields: list[str],
+    replay_ids: list[str],
+    project_ids: list[int],
+    period_start: datetime,
+    period_end: datetime,
+) -> Query:
+    """Return a query to fetch every replay in the set."""
+    from sentry.replays.query import QUERY_ALIAS_COLUMN_MAP, select_from_fields
+
+    def _select_from_fields() -> list[Union[Column, Function]]:
+        if fields:
+            return select_from_fields(list(set(fields)))
+        else:
+            return list(QUERY_ALIAS_COLUMN_MAP.values())
+
+    return Query(
+        match=Entity("replays"),
+        select=_select_from_fields(),
+        where=[
+            Condition(Column("project_id"), Op.IN, project_ids),
+            # Replay-ids were pre-calculated so no having clause and no aggregating significant
+            # amounts of data.
+            Condition(Column("replay_id"), Op.IN, replay_ids),
+            # We can scan an extended time range to account for replays which span either end of
+            # the range.  These timestamps are an optimization and could be removed with minimal
+            # performance impact.  It's a point query.  Its super fast.
+            Condition(Column("timestamp"), Op.GTE, period_start - timedelta(hours=1)),
+            Condition(Column("timestamp"), Op.LT, period_end + timedelta(hours=1)),
+        ],
+        groupby=[Column("project_id"), Column("replay_id")],
+        granularity=Granularity(3600),
+    )
+
+
+def _get_sort_column(column_name: str) -> Function:
+    try:
+        return agg_sort_config[column_name]
+    except KeyError:
+        raise ParseError(f"The field `{column_name}` is not a sortable field.")
+
+
+def _make_tenant_id(organization: Organization | None) -> dict[str, int]:
+    if organization is None:
+        return {}
+    else:
+        return {"organization_id": organization.id}