feat(eap-items): Route indexed-column filters from options-automator config

sentry-options/schemas/snuba/schema.json

@@ -838,6 +838,25 @@
       "default": 1785110400,
       "description": "Unix timestamp of the earliest weekly date bucket the v2 co-occurring-attributes tables hold data for; TraceItemAttributeNames only reads v2 for requests whose range starts at or after it, and reads v1 otherwise. Defaults to 2026-07-27 00:00 UTC, the Monday of the week the v2 tables and their materialized view were created. Must be a Monday: the date column is bucketed weekly with toMonday() and the query rounds its lower bound down to the previous Monday, so a mid-week value would let a request read a bucket v2 never populated. Lower it only to a bucket v2 has been backfilled to."
     },
+    "indexed_columns": {
+      "type": "object",
+      "additionalProperties": {
+        "type": "object",
+        "additionalProperties": {
+          "type": "object",
+          "properties": {
+            "indexed_column_name": {
+              "type": "string"
+            },
+            "indexed_start_date": {
+              "type": "string"
+            }
+          }
+        }
+      },
+      "default": {},
+      "description": "Nested dict: unindexed eap_items column or attribute (e.g. sentry.op) -> item type (span, metric, log, ...) -> the bloom-filter-indexed column holding a copy of it for that item type. One indexed column can hold a different attribute per item type (indexed_name is sentry.op for spans, sentry.metric_name for metrics), so filters on the attribute for any item type without an entry keep reading the attribute map. Indexed columns can't be backfilled, so indexed_start_date is the date (YYYY-MM-DD, from 00:00 UTC) from which the indexed column is fully populated for that item type. Equality and IN filters read indexed_column_name instead only for requests starting at or after it. Never set it before the column was fully populated, otherwise older rows are silently dropped. Empty (the default) disables the rewrite."
+    },
     "eap_items_indexed_name_start_timestamp": {
       "type": "integer",
       "default": 1790121600,

snuba/web/rpc/common/common.py

@@ -2,9 +2,10 @@
 from collections.abc import Callable, Iterable
 from dataclasses import replace
 from datetime import UTC, datetime, timedelta
-from typing import Any, TypeVar
+from typing import Any, TypedDict, TypeVar
 
 from google.protobuf.message import Message as ProtobufMessage
+from google.protobuf.timestamp_pb2 import Timestamp as ProtobufTimestamp
 from sentry_protos.snuba.v1.request_common_pb2 import RequestMeta, TraceItemType
 from sentry_protos.snuba.v1.trace_item_attribute_pb2 import AttributeKey, AttributeValue
 from sentry_protos.snuba.v1.trace_item_filter_pb2 import (
@@ -26,6 +27,7 @@
     array_element_column,
     coalesced_attribute_names,
     first_present_value,
+    get_trace_item_type_name,
     key_existence_conditions,
     sentry_column,
     type_array_to_membership_array_expression_from_typed_columns,
@@ -57,7 +59,7 @@
     Literal,
     SubscriptableReference,
 )
-from snuba.state.sentry_options import get_option
+from snuba.state.sentry_options import get_mapped_option, get_option
 from snuba.web.rpc.common.exceptions import BadSnubaRPCRequestException
 
 
@@ -1007,30 +1009,56 @@ def _any_attribute_filter_to_expression(
     return positive_expr
 
 
-_INDEXED_NAME_KEY_BY_ITEM_TYPE: dict[TraceItemType.ValueType, str] = {
-    TraceItemType.TRACE_ITEM_TYPE_SPAN: "sentry.op",
-    TraceItemType.TRACE_ITEM_TYPE_METRIC: "sentry.metric_name",
-}
+class IndexedColumn(TypedDict):
+    indexed_column_name: str
+    indexed_start_date: str
+
 
-# Rows written before this cutoff may have an empty indexed_name.
-INDEXED_NAME_START_TIMESTAMP_OPTION = "eap_items_indexed_name_start_timestamp"
-# Fail closed if the cutoff can't be read.
-_INDEXED_NAME_START_TIMESTAMP_FALLBACK = 2**63 - 1
+INDEXED_COLUMNS_OPTION = "indexed_columns"
 
 
-def use_indexed_name_for_request(meta: RequestMeta) -> bool:
-    start_timestamp = get_option(
-        INDEXED_NAME_START_TIMESTAMP_OPTION, _INDEXED_NAME_START_TIMESTAMP_FALLBACK
+def indexed_column_for(
+    start_timestamp: ProtobufTimestamp | None,
+    item_type: TraceItemType.ValueType,
+    unindexed_column: str,
+) -> str | None:
+    """
+    Indexed columns can't be backfilled, so each one only holds data from its
+    indexed_start_date (YYYY-MM-DD) onwards. An indexed column can hold a
+    different attribute per item type (indexed_name is sentry.op for spans,
+    sentry.metric_name for metrics), so each entry only applies to its own item_type.
+    Set per region in options-automator.
+    """
+    if start_timestamp is None:
+        return None
+
+    indexed_column: dict[str, IndexedColumn] = get_mapped_option(
+        INDEXED_COLUMNS_OPTION, unindexed_column, {}
     )
-    return meta.HasField("start_timestamp") and meta.start_timestamp.seconds >= start_timestamp
+
+    try:
+        type = get_trace_item_type_name(item_type)
+    except ValueError:
+        # This is only helper function and should never break the request execution
+        return None
+
+    indexed_type = indexed_column.get(type)
+    if indexed_type is None:
+        return None
+
+    indexed_start = datetime.fromisoformat(indexed_type["indexed_start_date"]).replace(tzinfo=UTC)
+    if start_timestamp.ToDatetime(tzinfo=UTC) < indexed_start:
+        return None
+
+    return indexed_type["indexed_column_name"]
 
 
 def trace_item_filters_to_expression(
     item_type: TraceItemType.ValueType,
     item_filter: TraceItemFilter,
     attribute_key_to_expression: Callable[[AttributeKey], Expression],
     membership_as_has: bool = False,
-    use_indexed_name: bool = False,
+    start_timestamp: ProtobufTimestamp | None = None,
 ) -> Expression:
     """
     Trace Item Filters are things like (span.id=12345 AND start_timestamp >= "june 4th, 2024")
@@ -1044,7 +1072,8 @@ def trace_item_filters_to_expression(
         ``IN`` set leaks an unstable ``__set_*`` identifier into the result-block column
         name and breaks mixed-version distributed reads (see ``_in_or_has``). Leave the
         default for WHERE clauses, where the prepared ``IN`` set drives pruning.
-    :param use_indexed_name: see ``use_indexed_name_for_request``.
+    :param start_timestamp: request start; enables reading indexed columns, see
+        ``indexed_column_for``.
     :return:
 
     Array predicates always read the typed ``attributes_array_*`` map columns: an
@@ -1062,7 +1091,7 @@ def trace_item_filters_to_expression(
                 filters[0],
                 attribute_key_to_expression,
                 membership_as_has,
-                use_indexed_name,
+                start_timestamp,
             )
         return and_cond(
             *(
@@ -1071,7 +1100,7 @@ def trace_item_filters_to_expression(
                     x,
                     attribute_key_to_expression,
                     membership_as_has,
-                    use_indexed_name,
+                    start_timestamp,
                 )
                 for x in filters
             )
@@ -1087,7 +1116,7 @@ def trace_item_filters_to_expression(
                 filters[0],
                 attribute_key_to_expression,
                 membership_as_has,
-                use_indexed_name,
+                start_timestamp,
             )
         return or_cond(
             *(
@@ -1096,7 +1125,7 @@ def trace_item_filters_to_expression(
                     x,
                     attribute_key_to_expression,
                     membership_as_has,
-                    use_indexed_name,
+                    start_timestamp,
                 )
                 for x in filters
             )
@@ -1113,7 +1142,7 @@ def trace_item_filters_to_expression(
                     filters[0],
                     attribute_key_to_expression,
                     membership_as_has,
-                    use_indexed_name,
+                    start_timestamp,
                 )
             )
         return not_cond(
@@ -1124,7 +1153,7 @@ def trace_item_filters_to_expression(
                         x,
                         attribute_key_to_expression,
                         membership_as_has,
-                        use_indexed_name,
+                        start_timestamp,
                     )
                     for x in filters
                 )
@@ -1155,17 +1184,16 @@ def trace_item_filters_to_expression(
             v_expression = _attribute_value_to_expression(v)
 
         if (
-            use_indexed_name
-            and k.type == AttributeKey.Type.TYPE_STRING
-            and k.name == _INDEXED_NAME_KEY_BY_ITEM_TYPE.get(item_type)
+            k.type == AttributeKey.Type.TYPE_STRING
             and not v.is_null
             and not item_filter.comparison_filter.ignore_case
+            and (index_name := indexed_column_for(start_timestamp, item_type, k.name)) is not None
         ):
             if op == ComparisonFilter.OP_EQUALS and v.val_str:
-                return f.equals(column("indexed_name"), v_expression)
+                return f.equals(column(index_name), v_expression)
             values = v.val_str_array.values
             if op == ComparisonFilter.OP_IN and values and "" not in values:
-                return _in_or_has(column("indexed_name"), v_expression, as_has=membership_as_has)
+                return _in_or_has(column(index_name), v_expression, as_has=membership_as_has)
 
         # `sentry.timestamp` is a normalized column that `attribute_key_to_expression`
         # maps to `CAST(timestamp, 'Float64')`. Wrapping the primary-key/partition

snuba/web/rpc/common/cross_item_queries.py

@@ -31,7 +31,6 @@
     base_conditions_and,
     trace_item_filters_to_expression,
     treeify_or_and_conditions,
-    use_indexed_name_for_request,
 )
 
 # 50 million trace ids * 16 bytes per id = a limit of 1gigabyte memory usage per cross item query
@@ -145,7 +144,6 @@ def get_trace_ids_sql_for_cross_item_query(
                 for trace_filter in trace_filters
             ]
 
-        use_indexed_name = use_indexed_name_for_request(request_meta)
         for trace_filter in converted_trace_filters:
             item_type_cond = f.equals(column("item_type"), trace_filter.item_type)
             filter_expressions.append(
@@ -155,7 +153,7 @@ def get_trace_ids_sql_for_cross_item_query(
                         trace_filter.item_type,
                         trace_filter.filter,
                         attribute_key_to_expression,
-                        use_indexed_name=use_indexed_name,
+                        start_timestamp=request_meta.start_timestamp,
                     ),
                 )
             )
@@ -167,7 +165,7 @@ def get_trace_ids_sql_for_cross_item_query(
                         trace_filter.filter,
                         attribute_key_to_expression,
                         membership_as_has=True,
-                        use_indexed_name=use_indexed_name,
+                        start_timestamp=request_meta.start_timestamp,
                     ),
                 )
             )

snuba/web/rpc/v1/endpoint_export_trace_items.py

@@ -43,7 +43,6 @@
     trace_item_filters_to_expression,
     treeify_or_and_conditions,
     typed_array_map_selected_expressions,
-    use_indexed_name_for_request,
 )
 from snuba.web.rpc.common.debug_info import setup_trace_query_settings
 from snuba.web.rpc.common.exceptions import BadSnubaRPCRequestException
@@ -348,7 +347,7 @@ def _build_query(
                 meta.trace_item_type,
                 in_msg.filter,
                 attribute_key_to_expression,
-                use_indexed_name=use_indexed_name_for_request(meta),
+                start_timestamp=meta.start_timestamp,
             ),
             *page_token_filter,
             *item_type_filter,

snuba/web/rpc/v1/endpoint_get_traces.py

@@ -47,7 +47,6 @@
     base_conditions_and,
     trace_item_filters_to_expression,
     treeify_or_and_conditions,
-    use_indexed_name_for_request,
 )
 from snuba.web.rpc.common.cross_item_queries import (
     CROSS_ITEM_DISTRIBUTED_PRODUCT_MODE,
@@ -579,7 +578,6 @@ def _get_trace_item_filter_expressions(
         for trace_filter in filters:
             filters_by_item_type[trace_filter.item_type].append(trace_filter.filter)
 
-        use_indexed_name = use_indexed_name_for_request(request_meta)
         for item_type in filters_by_item_type:
             filter_expressions_by_item_type[item_type] = and_cond(
                 f.equals(column("item_type"), item_type),
@@ -592,7 +590,7 @@ def _get_trace_item_filter_expressions(
                     ),
                     attribute_key_to_expression,
                     membership_as_has=True,
-                    use_indexed_name=use_indexed_name,
+                    start_timestamp=request_meta.start_timestamp,
                 ),
             )
 
@@ -617,7 +615,7 @@ def _get_trace_ids_for_single_item_query(
                 ),
             ),
             attribute_key_to_expression,
-            use_indexed_name=use_indexed_name_for_request(request.meta),
+            start_timestamp=request.meta.start_timestamp,
         )
         selected_columns: list[SelectedExpression] = [
             SelectedExpression(

snuba/web/rpc/v1/endpoint_time_series.py

@@ -54,7 +54,6 @@
     base_conditions_and,
     trace_item_filters_to_expression,
     treeify_or_and_conditions,
-    use_indexed_name_for_request,
     use_sampling_factor,
     valid_sampling_factor_conditions,
 )
@@ -423,7 +422,7 @@ def build_query(
                 request.meta.trace_item_type,
                 request.filter,
                 _get_attribute_key_to_expression_function(request.meta),
-                use_indexed_name=use_indexed_name_for_request(request.meta),
+                start_timestamp=request.meta.start_timestamp,
             ),
             valid_sampling_factor_conditions(),
             *item_type_conds,

snuba/web/rpc/v1/endpoint_trace_item_details.py

@@ -36,7 +36,6 @@
     trace_item_filters_to_expression,
     treeify_or_and_conditions,
     typed_array_map_selected_expressions,
-    use_indexed_name_for_request,
 )
 from snuba.web.rpc.common.debug_info import (
     extract_response_meta,
@@ -109,7 +108,7 @@ def _build_query(request: TraceItemDetailsRequest) -> Query:
                 request.meta.trace_item_type,
                 request.filter,
                 attribute_key_to_expression,
-                use_indexed_name=use_indexed_name_for_request(request.meta),
+                start_timestamp=request.meta.start_timestamp,
             ),
         ),
         limit=1,

snuba/web/rpc/v1/endpoint_trace_item_stats.py

@@ -49,7 +49,6 @@
     base_conditions_and,
     trace_item_filters_to_expression,
     treeify_or_and_conditions,
-    use_indexed_name_for_request,
 )
 from snuba.web.rpc.common.debug_info import (
     extract_response_meta,
@@ -240,7 +239,7 @@ def _build_attr_distribution_query(
         in_msg.meta.trace_item_type,
         in_msg.filter,
         (attribute_key_to_expression),
-        use_indexed_name=use_indexed_name_for_request(in_msg.meta),
+        start_timestamp=in_msg.meta.start_timestamp,
     )
     item_type_filter = f.equals(column("item_type"), in_msg.meta.trace_item_type)
     query = Query(

snuba/web/rpc/v1/endpoint_trace_item_table.py

@@ -78,7 +78,6 @@
     trace_item_filters_to_expression,
     treeify_or_and_conditions,
     typed_array_select_subcolumn_name,
-    use_indexed_name_for_request,
     use_sampling_factor,
     valid_sampling_factor_conditions,
 )
@@ -722,7 +721,7 @@ def build_query(
                 request.meta.trace_item_type,
                 request.filter,
                 attribute_key_to_expression,
-                use_indexed_name=use_indexed_name_for_request(request.meta),
+                start_timestamp=request.meta.start_timestamp,
             ),
             valid_sampling_factor_conditions(),
             *item_type_conds,

snuba/web/rpc/v1/heatmap_builder.py

@@ -35,7 +35,6 @@
     base_conditions_and,
     trace_item_filters_to_expression,
     treeify_or_and_conditions,
-    use_indexed_name_for_request,
 )
 from snuba.web.rpc.common.debug_info import setup_trace_query_settings
 from snuba.web.rpc.common.exceptions import BadSnubaRPCRequestException
@@ -126,7 +125,7 @@ def _get_min_max_bucketsize_y(
             in_msg.meta.trace_item_type,
             filter,
             (attribute_key_to_expression),
-            use_indexed_name=use_indexed_name_for_request(in_msg.meta),
+            start_timestamp=in_msg.meta.start_timestamp,
         )
         condition = base_conditions_and(in_msg.meta, filter_expression)
         min_max_query = Query(
@@ -218,7 +217,7 @@ def _build_heatmap_query(
             self.in_msg.meta.trace_item_type,
             filter,
             (attribute_key_to_expression),
-            use_indexed_name=use_indexed_name_for_request(self.in_msg.meta),
+            start_timestamp=self.in_msg.meta.start_timestamp,
         )
         condition = base_conditions_and(self.in_msg.meta, filter_expression)
         bucket_index_y = f.least(