diff --git a/snuba/web/rpc/common/pagination.py b/snuba/web/rpc/common/pagination.py index f9c3693ec0..8c381a7cf5 100644 --- a/snuba/web/rpc/common/pagination.py +++ b/snuba/web/rpc/common/pagination.py @@ -2,6 +2,9 @@ This file contains functionality to encode and decode custom page tokens """ +from collections.abc import Mapping +from typing import Final + from google.protobuf.timestamp_pb2 import Timestamp from sentry_protos.snuba.v1.endpoint_trace_item_table_pb2 import ( TraceItemColumnValues, @@ -15,15 +18,57 @@ TraceItemFilter, ) +from snuba.protos.common import NORMALIZED_COLUMNS_EAP_ITEMS from snuba.query.dsl import Functions as f from snuba.query.dsl import column, literal -from snuba.query.expressions import Expression +from snuba.query.expressions import Expression, OptionalScalarType from snuba.web.rpc.common.common import ( attribute_key_to_expression, semver_sort_key, ) +from snuba.web.rpc.common.exceptions import BadSnubaRPCRequestException from snuba.web.rpc.storage_routing.routing_strategies.storage_routing import TimeWindow +# Value an absent map-backed attribute sorts as, per attribute type. The page boundary +# compares the ORDER BY values as a tuple, and a NULL element makes the whole comparison +# NULL, which drops the row instead of paginating past it. A stored value equal to the +# sentinel ties with an absent one, which is harmless: the trailing `sentry.item_id` +# element breaks the tie. +_NULL_ORDERING_SENTINELS: Final[Mapping[AttributeKey.Type.ValueType, OptionalScalarType]] = { + AttributeKey.Type.TYPE_BOOLEAN: False, + AttributeKey.Type.TYPE_DOUBLE: 0.0, + AttributeKey.Type.TYPE_FLOAT: 0.0, + AttributeKey.Type.TYPE_INT: 0, + AttributeKey.Type.TYPE_STRING: "", +} + + +def null_safe_ordering_expression( + expression: Expression, attr_type: AttributeKey.Type.ValueType +) -> Expression: + """Make an absent map-backed attribute sort as its type's zero value. + + Apply this to the ORDER BY and to the page boundary of the same column, or the two + disagree on where absent keys sort and pagination skips or repeats rows. A type with no + sentinel is returned unchanged, which covers the columns that need none: normalized + columns and `timestamp` (never NULL, and their page token carries no type), and arrays + (rejected from ORDER BY upstream). + """ + if attr_type not in _NULL_ORDERING_SENTINELS: + return expression + return f.ifNull(expression, literal(_NULL_ORDERING_SENTINELS[attr_type])) + + +def _comparison_value_expression(comparison_filter: ComparisonFilter) -> Expression: + value = comparison_filter.value + if value.is_null or value.WhichOneof("value") == "val_null": + if comparison_filter.key.type not in _NULL_ORDERING_SENTINELS: + raise BadSnubaRPCRequestException( + f"page token column {comparison_filter.key.name} is null and has no type to sort it by" + ) + return literal(_NULL_ORDERING_SENTINELS[comparison_filter.key.type]) + return literal(getattr(value, str(value.WhichOneof("value")))) + class FlexibleTimeWindowPageWithFilters: _TIME_WINDOW_PREFIX = "sentry__time_window" @@ -79,6 +124,9 @@ def get_filters(self) -> Expression | None: # Parallel to column_names: True when that column's ORDER BY used # SORT_SEMVER, so the boundary comparison must use the semver key too. column_is_semver: list[bool] = [] + # Parallel to column_names: the attribute type, which tells the boundary + # comparison how a map-backed column with an absent key was sorted. + column_types: list[AttributeKey.Type.ValueType] = [] for filter in self.page_token.filter_offset.and_filter.filters: if not filter.HasField("comparison_filter"): @@ -107,33 +155,30 @@ def get_filters(self) -> Expression | None: ) column_names.append("timestamp") column_is_semver.append(False) + column_types.append(AttributeKey.Type.TYPE_UNSPECIFIED) else: # strip the matching prefix (and the dot) to recover the alias prefix = self._SEMVER_FILTER_PREFIX if is_semver else self._FILTER_PREFIX column_names.append(key_name[len(prefix) + 1 :]) column_is_semver.append(is_semver) - column_values.append( - literal( - getattr( - filter.comparison_filter.value, - str(filter.comparison_filter.value.WhichOneof("value")), - ) - ) - ) + column_types.append(filter.comparison_filter.key.type) + column_values.append(_comparison_value_expression(filter.comparison_filter)) # Assumes everything in the ORDER BY is ordered by DESC if column_names: col_exprs = [] val_exprs = [] - for c_name, c_value, is_semver in zip( - column_names, column_values, column_is_semver, strict=True + for c_name, c_value, is_semver, c_type in zip( + column_names, column_values, column_is_semver, column_types, strict=True ): + # An absent map-backed key sorts as its type's zero value, as in ORDER BY. + col_expr = null_safe_ordering_expression(column(c_name), c_type) # For SORT_SEMVER columns, apply the same semver key on both sides # so the page-boundary comparison uses the same ordering as ORDER BY. if is_semver: - col_exprs.append(semver_sort_key(column(c_name))) + col_exprs.append(semver_sort_key(col_expr)) val_exprs.append(semver_sort_key(c_value)) else: - col_exprs.append(column(c_name)) + col_exprs.append(col_expr) val_exprs.append(c_value) res = f.less(f.tuple(*col_exprs), f.tuple(*val_exprs)) return res @@ -234,11 +279,24 @@ def create( ) prefix = cls._SEMVER_FILTER_PREFIX if is_semver else cls._FILTER_PREFIX + # Only a map-backed attribute reads as NULL when its key is absent, + # and `last_result_value` is then null; get_filters uses this type to + # sort it the way ORDER BY did. A normalized column is never NULL and + # its ORDER BY compares the raw column, so leaving the type unset is + # what keeps the two sides in step. + null_sort_type = ( + selected_key.type + if selected_key is not None + and selected_key.name not in NORMALIZED_COLUMNS_EAP_ITEMS + else AttributeKey.Type.TYPE_UNSPECIFIED + ) + filters.append( TraceItemFilter( comparison_filter=ComparisonFilter( key=AttributeKey( name=f"{prefix}.{attribute_expression.alias}", + type=null_sort_type, ), op=ComparisonFilter.OP_LESS_THAN, value=last_result_value, diff --git a/snuba/web/rpc/v1/resolvers/R_eap_items/resolver_trace_item_table.py b/snuba/web/rpc/v1/resolvers/R_eap_items/resolver_trace_item_table.py index 0f4266b49e..f28458efa5 100644 --- a/snuba/web/rpc/v1/resolvers/R_eap_items/resolver_trace_item_table.py +++ b/snuba/web/rpc/v1/resolvers/R_eap_items/resolver_trace_item_table.py @@ -6,6 +6,7 @@ import sentry_sdk from google.protobuf.json_format import MessageToDict +from sentry_protos.snuba.v1.downsampled_storage_pb2 import DownsampledStorageConfig from sentry_protos.snuba.v1.endpoint_trace_item_table_pb2 import ( AggregationComparisonFilter, AggregationFilter, @@ -76,7 +77,10 @@ extract_response_meta, ) from snuba.web.rpc.common.exceptions import BadSnubaRPCRequestException -from snuba.web.rpc.common.pagination import FlexibleTimeWindowPageWithFilters +from snuba.web.rpc.common.pagination import ( + FlexibleTimeWindowPageWithFilters, + null_safe_ordering_expression, +) from snuba.web.rpc.storage_routing.routing_strategies.storage_routing import ( RoutingDecision, TimeWindow, @@ -299,6 +303,7 @@ def _convert_order_by( groupby: list[Expression], order_by: Sequence[TraceItemTableRequest.OrderBy], request_meta: RequestMeta, + paginated_by_order_by: bool = False, ) -> Sequence[OrderBy]: res: list[OrderBy] = [] for i, x in enumerate(order_by): @@ -345,6 +350,12 @@ def _convert_order_by( # expression so an aggregation query that orders by `sentry.timestamp` stays # valid. expression = _groupby_order_by_expression(x.column.key) + if paginated_by_order_by and x.column.key.name not in NORMALIZED_COLUMNS_EAP_ITEMS: + # A map-backed attribute reads as NULL when the key is absent, and the page + # token compares this same value as a tuple element, where a NULL makes the + # whole comparison NULL and drops the row. Sort absent keys somewhere the + # comparison can also express. + expression = null_safe_ordering_expression(expression, x.column.key.type) # SORT_SEMVER: client-driven semver ordering, string columns only # (numeric/timestamp columns already sort numerically). if ( @@ -732,6 +743,15 @@ def build_query( groupby, request.order_by, request.meta, + # Flextime is the mode that pages through the ORDER BY values (see + # FlexibleTimeWindowPageWithFilters). Keyed off the request rather than off + # `time_window` because a swallowed routing failure drops the time window while + # the client's page token still compares against the ORDER BY of the page + # before it, and the two must agree on where absent keys sort. + paginated_by_order_by=( + request.meta.downsampled_storage_config.mode + == DownsampledStorageConfig.MODE_HIGHEST_ACCURACY_FLEXTIME + ), ), limitby=_convert_limit_by(request.limit_by, selected_columns), groupby=groupby, diff --git a/tests/web/rpc/test_pagination.py b/tests/web/rpc/test_pagination.py new file mode 100644 index 0000000000..82ef6da811 --- /dev/null +++ b/tests/web/rpc/test_pagination.py @@ -0,0 +1,110 @@ +import pytest +from google.protobuf.timestamp_pb2 import Timestamp +from sentry_protos.snuba.v1.endpoint_trace_item_table_pb2 import ( + Column, + TraceItemColumnValues, + TraceItemTableRequest, +) +from sentry_protos.snuba.v1.request_common_pb2 import RequestMeta, TraceItemType +from sentry_protos.snuba.v1.trace_item_attribute_pb2 import AttributeKey, AttributeValue + +from snuba.query.dsl import Functions as f +from snuba.query.dsl import column, literal +from snuba.query.expressions import Expression +from snuba.web.rpc.common.exceptions import BadSnubaRPCRequestException +from snuba.web.rpc.common.pagination import FlexibleTimeWindowPageWithFilters +from snuba.web.rpc.storage_routing.routing_strategies.storage_routing import TimeWindow + +_START = Timestamp(seconds=1_700_000_000) +_END = Timestamp(seconds=1_700_003_600) +_TIME_WINDOW = TimeWindow(start_timestamp=_START, end_timestamp=_END) + +_SEQUENCE_ALIAS = "sentry.timestamp.sequence_TYPE_INT" +_ITEM_ID_ALIAS = "sentry.item_id_TYPE_STRING" + + +def _request() -> TraceItemTableRequest: + columns = [ + Column( + key=AttributeKey(type=AttributeKey.TYPE_STRING, name="sentry.timestamp"), + label="sentry.timestamp", + ), + Column( + key=AttributeKey(type=AttributeKey.TYPE_INT, name="sentry.timestamp.sequence"), + label="sentry.timestamp.sequence", + ), + Column( + key=AttributeKey(type=AttributeKey.TYPE_STRING, name="sentry.item_id"), + label="sentry.item_id", + ), + ] + return TraceItemTableRequest( + meta=RequestMeta( + project_ids=[1], + organization_id=1, + start_timestamp=_START, + end_timestamp=_END, + trace_item_type=TraceItemType.TRACE_ITEM_TYPE_LOG, + ), + columns=columns, + order_by=[TraceItemTableRequest.OrderBy(column=col, descending=True) for col in columns], + ) + + +def _results(sequence: AttributeValue) -> list[TraceItemColumnValues]: + return [ + TraceItemColumnValues( + attribute_name="sentry.timestamp", + results=[AttributeValue(val_str="2025-10-06 14:00:00")], + ), + TraceItemColumnValues( + attribute_name="sentry.timestamp.sequence", + results=[sequence], + ), + TraceItemColumnValues( + attribute_name="sentry.item_id", + results=[AttributeValue(val_str="deadbeef")], + ), + ] + + +def _expected_filters(sequence_bookmark: int) -> Expression: + return f.less( + f.tuple( + column("timestamp"), + f.ifNull(column(_SEQUENCE_ALIAS), literal(0)), + # `sentry.item_id` is a normalized column, never NULL, so it needs no sentinel. + column(_ITEM_ID_ALIAS), + ), + f.tuple( + f.toDateTime("2025-10-06 14:00:00"), + literal(sequence_bookmark), + literal("deadbeef"), + ), + ) + + +class TestFlexibleTimeWindowPageWithFilters: + def test_compares_the_last_value_when_the_order_by_attribute_is_present(self) -> None: + page = FlexibleTimeWindowPageWithFilters.create( + _request(), _TIME_WINDOW, _results(AttributeValue(val_int=7)) + ) + + assert page.get_filters() == _expected_filters(7) + + def test_compares_the_null_sentinel_when_the_order_by_attribute_is_absent(self) -> None: + page = FlexibleTimeWindowPageWithFilters.create( + _request(), _TIME_WINDOW, _results(AttributeValue(is_null=True)) + ) + + assert page.get_filters() == _expected_filters(0) + + def test_rejects_a_null_bookmark_whose_page_token_carries_no_attribute_type(self) -> None: + page = FlexibleTimeWindowPageWithFilters.create( + _request(), _TIME_WINDOW, _results(AttributeValue(is_null=True)) + ) + for filter in page.page_token.filter_offset.and_filter.filters: + filter.comparison_filter.key.ClearField("type") + + with pytest.raises(BadSnubaRPCRequestException): + FlexibleTimeWindowPageWithFilters(page.page_token).get_filters() diff --git a/tests/web/rpc/v1/test_endpoint_trace_item_table/test_endpoint_trace_item_table.py b/tests/web/rpc/v1/test_endpoint_trace_item_table/test_endpoint_trace_item_table.py index 2663b98276..fe670e8848 100644 --- a/tests/web/rpc/v1/test_endpoint_trace_item_table/test_endpoint_trace_item_table.py +++ b/tests/web/rpc/v1/test_endpoint_trace_item_table/test_endpoint_trace_item_table.py @@ -63,6 +63,7 @@ from snuba.query import LimitBy, OrderBy, OrderByDirection from snuba.query.dsl import Functions as f from snuba.query.dsl import column as snuba_column +from snuba.query.dsl import literal from snuba.query.expressions import Expression from snuba.web import QueryException from snuba.web.rpc import RPCEndpoint @@ -4499,6 +4500,54 @@ def test_build_query_with_order_by_optimization_disabled_because_groupby() -> No ] +def test_build_query_orders_flextime_map_attributes_null_safely_without_a_time_window() -> None: + # A swallowed routing failure leaves the decision with no time window, but the client's + # page token still compares against the previous page's ORDER BY, so the null handling + # has to follow the requested mode rather than the routing outcome. + request = TraceItemTableRequest( + meta=RequestMeta( + project_ids=[1], + trace_item_type=TraceItemType.TRACE_ITEM_TYPE_LOG, + downsampled_storage_config=DownsampledStorageConfig( + mode=DownsampledStorageConfig.MODE_HIGHEST_ACCURACY_FLEXTIME + ), + ), + columns=[ + Column(key=AttributeKey(type=AttributeKey.TYPE_INT, name="sentry.timestamp.sequence")), + Column(key=AttributeKey(type=AttributeKey.TYPE_STRING, name="sentry.item_id")), + ], + order_by=[ + TraceItemTableRequest.OrderBy( + column=Column( + key=AttributeKey(type=AttributeKey.TYPE_INT, name="sentry.timestamp.sequence") + ), + descending=True, + ), + TraceItemTableRequest.OrderBy( + column=Column( + key=AttributeKey(type=AttributeKey.TYPE_STRING, name="sentry.item_id") + ), + descending=True, + ), + ], + ) + request = _apply_labels_to_columns(request) + + query = build_query(request, time_window=None) + + selected = {column.name: column.expression for column in query.get_selected_columns()} + assert query.get_orderby() == [ + OrderBy( + direction=OrderByDirection.DESC, + expression=f.ifNull(selected["sentry.timestamp.sequence"], literal(0)), + ), + OrderBy( + direction=OrderByDirection.DESC, + expression=selected["sentry.item_id"], + ), + ] + + def test_order_by_bug() -> None: start_ts = Timestamp() start_ts.FromDatetime(datetime.fromisoformat("2025-10-22T17:55:24Z")) diff --git a/tests/web/rpc/v1/test_endpoint_trace_item_table/test_trace_item_table_flex_time.py b/tests/web/rpc/v1/test_endpoint_trace_item_table/test_trace_item_table_flex_time.py index 79c2d1af62..eff0e8dba1 100644 --- a/tests/web/rpc/v1/test_endpoint_trace_item_table/test_trace_item_table_flex_time.py +++ b/tests/web/rpc/v1/test_endpoint_trace_item_table/test_trace_item_table_flex_time.py @@ -1,4 +1,5 @@ import random +from collections.abc import Callable from dataclasses import dataclass from datetime import datetime, timedelta from typing import Any @@ -47,41 +48,46 @@ class LogOutcomeDataPoint: num_logs: int -def _store_logs_and_outcomes(data_points: list[LogOutcomeDataPoint]) -> None: +def _default_log_attributes(time: datetime, index: int) -> dict[str, AnyValue]: + return { + "color": AnyValue( + string_value=random.choice( + [ + "red", + "green", + "blue", + ] + ) + ), + "location": AnyValue( + string_value=random.choice( + [ + "mobile", + "frontend", + "backend", + ] + ) + ), + "sentry.timestamp_precise": AnyValue(double_value=int(time.timestamp()) + random.random()), + } + + +def _store_logs_and_outcomes( + data_points: list[LogOutcomeDataPoint], + log_attributes: Callable[[datetime, int], dict[str, AnyValue]] = _default_log_attributes, +) -> None: items_storage = get_writable_storage(StorageKey("eap_items")) messages = [] outcome_data = [] for data_point in data_points: - for _ in range(data_point.num_logs): + for index in range(data_point.num_logs): item_id = random.randint(0, 2**128 - 1).to_bytes(16, byteorder="big") message = gen_item_message( start_timestamp=data_point.time, item_id=item_id, type=TraceItemType.TRACE_ITEM_TYPE_LOG, - attributes={ - "color": AnyValue( - string_value=random.choice( - [ - "red", - "green", - "blue", - ] - ) - ), - "location": AnyValue( - string_value=random.choice( - [ - "mobile", - "frontend", - "backend", - ] - ) - ), - "sentry.timestamp_precise": AnyValue( - double_value=int(data_point.time.timestamp()) + random.random() - ), - }, + attributes=log_attributes(data_point.time, index), project_id=_PROJECT_ID, organization_id=_ORG_ID, ) @@ -343,6 +349,86 @@ def test_paginate_within_time_window(self, eap: Any) -> None: queried_item_ids ) + def test_paginate_when_an_order_by_attribute_is_absent_from_some_items(self, eap: Any) -> None: + def log_attributes(time: datetime, index: int) -> dict[str, AnyValue]: + # Every log in an hour shares a `timestamp_precise`, so ordering falls through to + # `sentry.timestamp.sequence` — which only half of them carry, the way logs from + # an SDK that predates that counter do. + attributes = { + "color": AnyValue(string_value="red"), + "sentry.timestamp_precise": AnyValue(double_value=int(time.timestamp())), + } + if index % 2 == 0: + attributes["sentry.timestamp.sequence"] = AnyValue(int_value=index) + return attributes + + num_hours_to_query = 4 + _store_logs_and_outcomes( + [ + LogOutcomeDataPoint( + time=BASE_TIME - timedelta(hours=hour), + num_outcomes=10_000_000, + num_logs=_LOG_COUNT, + ) + for hour in range(num_hours_to_query + 1) + ], + log_attributes, + ) + + columns = [ + Column(key=AttributeKey(type=AttributeKey.TYPE_DOUBLE, name="sentry.timestamp")), + Column( + key=AttributeKey(type=AttributeKey.TYPE_DOUBLE, name="sentry.timestamp_precise") + ), + Column(key=AttributeKey(type=AttributeKey.TYPE_INT, name="sentry.timestamp.sequence")), + Column(key=AttributeKey(type=AttributeKey.TYPE_STRING, name="sentry.item_id")), + ] + order_by = [ + TraceItemTableRequest.OrderBy(column=column, descending=True) for column in columns + ] + start_timestamp = Timestamp( + seconds=int((BASE_TIME - timedelta(hours=num_hours_to_query)).timestamp()) + ) + end_timestamp = Timestamp(seconds=int(BASE_TIME.timestamp())) + + all_ids_response = EndpointTraceItemTable().execute( + _generate_table_request( + start_timestamp, + end_timestamp, + accuracy=DownsampledStorageConfig.MODE_HIGHEST_ACCURACY, + limit=3000, + columns=columns, + order_by=order_by, + ) + ) + stored_item_ids = get_item_ids_from_response(all_ids_response) + + strategy = OutcomesFlexTimeRoutingStrategy() + end_pagination = PageToken(end_pagination=True) + page_token = PageToken(offset=0) + queried_item_ids: list[str] = [] + with override_component_config(strategy, "max_items_to_query", 20_000_000): + while page_token != end_pagination: + response = EndpointTraceItemTable().execute( + _generate_table_request( + start_timestamp, + end_timestamp, + accuracy=DownsampledStorageConfig.Mode.MODE_HIGHEST_ACCURACY_FLEXTIME, + limit=_LOG_COUNT, + page_token=page_token, + columns=columns, + order_by=order_by, + ) + ) + assert isinstance(response, TraceItemTableResponse) + page_token = response.page_token + queried_item_ids.extend(get_item_ids_from_response(response)) + + assert len(set(queried_item_ids)) == len(queried_item_ids) + assert set(queried_item_ids) == set(stored_item_ids), set(stored_item_ids) - set( + queried_item_ids + ) + def test_paginate_first_page_empty(self, eap: Any) -> None: data_points = [ LogOutcomeDataPoint(