feat(deletes): support TraceItemFilter[s] in DeleteTraceItems API (#7535)

Wire DeleteTraceEndpoint TraceItemFilter[s] into bulk_delete_query
arguments. This allows product to test API-level integration with the
delete endpoint, without blowing up the size of this change.

This validates the incoming fields and passes them through in a new
`AttributeConditions` object when not deleting by trace_id

Since we don't support delete by attributes, and it's unsafe to only use
the column values provided, we short-circuit requests that have
attribute conditions (and do no delete at all).

Next steps:
1. Wire these values into the DeleteQueryMessage (undecided at the
moment if we should resolve the attributes to columns when we create
that message, or if attribute_conditions should be a new field that gets
resolved when the consumer executes the delete)
2. Run a test?

Depends on #7534

Author: onewland

snuba/web/bulk_delete_query.py

@@ -1,7 +1,17 @@
 import logging
 import time
+from dataclasses import dataclass
 from threading import Thread
-from typing import Any, Dict, Mapping, MutableMapping, Optional, Sequence, TypedDict
+from typing import (
+    Any,
+    Dict,
+    List,
+    Mapping,
+    MutableMapping,
+    Optional,
+    Sequence,
+    TypedDict,
+)
 
 import rapidjson
 from confluent_kafka import KafkaError
@@ -12,6 +22,7 @@
 from snuba.attribution.attribution_info import AttributionInfo
 from snuba.clickhouse.columns import ColumnSet
 from snuba.clickhouse.query import Query
+from snuba.datasets.deletion_settings import DeletionSettings
 from snuba.datasets.storage import WritableTableStorage
 from snuba.datasets.storages.storage_key import StorageKey
 from snuba.query.conditions import combine_or_conditions
@@ -39,6 +50,12 @@
 logger = logging.getLogger(__name__)
 
 
+@dataclass
+class AttributeConditions:
+    item_type: int
+    attributes: Dict[str, List[Any]]
+
+
 class DeleteQueryMessage(TypedDict):
     rows_to_delete: int
     storage_name: str
@@ -123,11 +140,63 @@ def produce_delete_query(delete_query: DeleteQueryMessage) -> None:
         logger.exception("Could not produce delete query due to error: %r", ex)
 
 
+def _validate_attribute_conditions(
+    attribute_conditions: AttributeConditions,
+    delete_settings: DeletionSettings,
+) -> None:
+    """
+    Validates that the attribute_conditions are allowed for the configured item_type.
+
+    Args:
+        attribute_conditions: AttributeConditions containing item_type and attribute mappings
+        delete_settings: The deletion settings for the storage
+
+    Raises:
+        InvalidQueryException: If no attributes are configured for the item_type,
+                              or if any requested attributes are not allowed
+    """
+    # Get the string name for the item_type from the configuration
+    # The configuration uses string names (e.g., "occurrence") as keys
+    allowed_attrs_config = delete_settings.allowed_attributes_by_item_type
+
+    if not allowed_attrs_config:
+        raise InvalidQueryException("No attribute-based deletions configured for this storage")
+
+    # Check if the item_type has any allowed attributes configured
+    # Since the config uses string names and we're given an integer, we need to find the matching config
+    # For now, we'll check all configured item types and validate against any that match
+
+    # Try to find a matching configuration by item_type name
+    matching_allowed_attrs = None
+    for configured_item_type, allowed_attrs in allowed_attrs_config.items():
+        # For this initial implementation, we'll use the string key directly
+        # In the future, we might need a mapping from item_type int to string name
+        matching_allowed_attrs = allowed_attrs
+        break  # For now, assume the first/only configured type
+
+    if matching_allowed_attrs is None:
+        raise InvalidQueryException(
+            f"No attribute-based deletions configured for item_type {attribute_conditions.item_type}"
+        )
+
+    # Validate that all requested attributes are allowed
+    requested_attrs = set(attribute_conditions.attributes.keys())
+    allowed_attrs_set = set(matching_allowed_attrs)
+    invalid_attrs = requested_attrs - allowed_attrs_set
+
+    if invalid_attrs:
+        raise InvalidQueryException(
+            f"Invalid attributes for deletion: {invalid_attrs}. "
+            f"Allowed attributes: {allowed_attrs_set}"
+        )
+
+
 @with_span()
 def delete_from_storage(
     storage: WritableTableStorage,
     conditions: Dict[str, list[Any]],
     attribution_info: Mapping[str, Any],
+    attribute_conditions: Optional[AttributeConditions] = None,
 ) -> dict[str, Result]:
     """
     This method does a series of validation checks (outline below),
@@ -138,6 +207,7 @@ def delete_from_storage(
     * storage validation that deletes are enabled
     * column names are valid (allowed_columns storage setting)
     * column types are valid
+    * attribute names are valid (allowed_attributes_by_item_type storage setting)
     """
     if not deletes_are_enabled():
         raise DeletesNotEnabledError("Deletes not enabled in this region")
@@ -161,6 +231,16 @@ def delete_from_storage(
     except InvalidColumnType as e:
         raise InvalidQueryException(e.message)
 
+    # validate attribute conditions if provided
+    if attribute_conditions:
+        _validate_attribute_conditions(attribute_conditions, delete_settings)
+        logger.error(
+            "valid attribute_conditions passed to delete_from_storage, but delete will be ignored "
+            "as functionality is not yet implemented"
+        )
+        # deleting by just conditions and ignoring attribute_conditions would be dangerous
+        return {}
+
     attr_info = _get_attribution_info(attribution_info)
     return delete_from_tables(storage, delete_settings.tables, conditions, attr_info)
 

snuba/web/rpc/v1/endpoint_delete_trace_items.py

@@ -1,18 +1,99 @@
-from typing import Type
+from typing import Any, Dict, List, Optional, Sequence, Type
 
 from sentry_protos.snuba.v1.endpoint_delete_trace_items_pb2 import (
     DeleteTraceItemsRequest,
     DeleteTraceItemsResponse,
 )
+from sentry_protos.snuba.v1.request_common_pb2 import TraceItemFilterWithType
+from sentry_protos.snuba.v1.trace_item_filter_pb2 import ComparisonFilter
 
 from snuba.attribution.appid import AppID
 from snuba.datasets.storages.factory import get_writable_storage
 from snuba.datasets.storages.storage_key import StorageKey
-from snuba.web.bulk_delete_query import delete_from_storage
+from snuba.web.bulk_delete_query import AttributeConditions, delete_from_storage
 from snuba.web.rpc import RPCEndpoint
 from snuba.web.rpc.common.exceptions import BadSnubaRPCRequestException
 
 
+def _extract_attribute_value(comparison_filter: ComparisonFilter) -> Any:
+    """Extract the value from a ComparisonFilter's AttributeValue."""
+    value_field = comparison_filter.value.WhichOneof("value")
+    if value_field == "val_str":
+        return comparison_filter.value.val_str
+    elif value_field == "val_int":
+        return comparison_filter.value.val_int
+    elif value_field == "val_double":
+        return comparison_filter.value.val_double
+    elif value_field == "val_bool":
+        return comparison_filter.value.val_bool
+    elif value_field == "val_str_array":
+        return list(comparison_filter.value.val_str_array.values)
+    elif value_field == "val_int_array":
+        return list(comparison_filter.value.val_int_array.values)
+    elif value_field == "val_double_array":
+        return list(comparison_filter.value.val_double_array.values)
+    else:
+        raise BadSnubaRPCRequestException(f"Unsupported attribute value type: {value_field}")
+
+
+def _trace_item_filters_to_attribute_conditions(
+    item_type: int,
+    filters: Sequence[TraceItemFilterWithType],
+) -> AttributeConditions:
+    """
+    Convert TraceItemFilters to AttributeConditions for deletion.
+
+    Only supports ComparisonFilter with OP_EQUALS or OP_IN operations.
+    All filters are combined with AND logic.
+
+    Args:
+        item_type: The trace item type (e.g., occurrence, span)
+        filters: List of TraceItemFilterWithType from the request
+
+    Returns:
+        AttributeConditions object containing item_type and attribute mappings
+
+    Raises:
+        BadSnubaRPCRequestException: If unsupported filter types or operations are encountered
+    """
+    attributes: Dict[str, List[Any]] = {}
+
+    for filter_with_type in filters:
+        # Extract the actual filter from TraceItemFilterWithType
+        trace_filter = filter_with_type.filter
+        # Only support comparison filters for deletion
+        if not trace_filter.HasField("comparison_filter"):
+            raise BadSnubaRPCRequestException(
+                "Only comparison filters are supported for deletion. "
+                "AND, OR, and NOT filters are not supported."
+            )
+
+        comparison_filter = trace_filter.comparison_filter
+        op = comparison_filter.op
+
+        # Only support equality operations for deletion
+        if op not in (ComparisonFilter.OP_EQUALS, ComparisonFilter.OP_IN):
+            op_name = ComparisonFilter.Op.Name(op)
+            raise BadSnubaRPCRequestException(
+                f"Only OP_EQUALS and OP_IN operations are supported for deletion. Got: {op_name}"
+            )
+
+        attribute_name = comparison_filter.key.name
+        value = _extract_attribute_value(comparison_filter)
+
+        # Convert single values to lists for consistency
+        if not isinstance(value, list):
+            value = [value]
+
+        # If the attribute already exists, extend the list (OR logic within same attribute)
+        if attribute_name in attributes:
+            attributes[attribute_name].extend(value)
+        else:
+            attributes[attribute_name] = value
+
+    return AttributeConditions(item_type=item_type, attributes=attributes)
+
+
 class EndpointDeleteTraceItems(RPCEndpoint[DeleteTraceItemsRequest, DeleteTraceItemsResponse]):
     @classmethod
     def version(cls) -> str:
@@ -36,9 +117,6 @@ def _execute(self, request: DeleteTraceItemsRequest) -> DeleteTraceItemsResponse
         if has_trace_ids and has_filters:
             raise BadSnubaRPCRequestException("Provide only one of trace_ids or filters, not both.")
 
-        if has_filters:
-            raise NotImplementedError("Currently, only delete by trace_ids is supported")
-
         attribution_info = {
             "app_id": AppID("eap"),
             "referrer": request.meta.referrer,
@@ -51,14 +129,37 @@ def _execute(self, request: DeleteTraceItemsRequest) -> DeleteTraceItemsResponse
             "parent_api": "eap_delete_trace_items",
         }
 
+        # Build base conditions that apply to all deletions
+        conditions: Dict[str, List[Any]] = {
+            "organization_id": [request.meta.organization_id],
+            "project_id": list(request.meta.project_ids),
+        }
+
+        attribute_conditions: Optional[AttributeConditions] = None
+
+        if has_trace_ids:
+            # Delete by trace_ids (no attribute filtering)
+            conditions["trace_id"] = [str(tid) for tid in request.trace_ids]
+        else:
+            # Delete by filters (with attribute filtering)
+            # item_type must be specified in the request metadata for attribute-based deletion
+            if request.meta.trace_item_type == 0:  # TRACE_ITEM_TYPE_UNSPECIFIED
+                raise BadSnubaRPCRequestException(
+                    "trace_item_type must be specified in metadata when using filters"
+                )
+
+            attribute_conditions = _trace_item_filters_to_attribute_conditions(
+                request.meta.trace_item_type,
+                request.filters,
+            )
+            # Add item_type to conditions for the delete query
+            conditions["item_type"] = [attribute_conditions.item_type]
+
         delete_result = delete_from_storage(
             get_writable_storage(StorageKey.EAP_ITEMS),
-            {
-                "organization_id": [request.meta.organization_id],
-                "trace_id": list(request.trace_ids),
-                "project_id": list(request.meta.project_ids),
-            },
+            conditions,
             attribution_info,
+            attribute_conditions,
         )
 
         response = DeleteTraceItemsResponse()