fix(querier): page raw results by the full row order key

The raw cursor held the last row's millisecond, so the next page dropped
or repeated rows inside that millisecond. The cursor now holds the
timestamp in nanoseconds and the tie-break columns (id for logs, trace_id
and span_id for traces), and the next page takes the rows strictly after
that row. A traces order of timestamp alone gets the tie-break columns.

A cursor in the millisecond format still pages the old way, so a page held
across an upgrade keeps working. An offset with a cursor, an undecodable
cursor, and a cursor on a raw query not ordered by timestamp first are
rejected instead of ignored.

Assisted-by: Claude Opus 5.5

pkg/querier/builder_query.go

@@ -3,8 +3,10 @@ package querier
 import (
 	"context"
 	"encoding/base64"
+	"encoding/json"
 	"fmt"
 	"log/slog"
+	"slices"
 	"strconv"
 	"strings"
 	"time"
@@ -255,6 +257,10 @@ func (q *builderQuery[T]) Execute(ctx context.Context) (*qbtypes.Result, error)
 	if q.kind == qbtypes.RequestTypeRaw && q.isWindowList() {
 		return q.executeWindowList(ctx)
 	}
+	if q.kind == qbtypes.RequestTypeRaw && strings.TrimSpace(q.spec.Cursor) != "" {
+		return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "cursor needs a raw query ordered by timestamp first").
+			WithSuggestions("order logs by timestamp then id, both in one direction, and traces by timestamp")
+	}
 
 	fromMS, toMS := q.fromMS, q.toMS
 	if q.spec.Signal == telemetrytypes.SignalTraces || q.spec.Signal == telemetrytypes.SignalLogs {
@@ -517,20 +523,50 @@ func (q *builderQuery[T]) executeWindowList(ctx context.Context) (*qbtypes.Resul
 	isAsc := len(q.spec.Order) > 0 &&
 		strings.ToLower(string(q.spec.Order[0].Direction.StringValue())) == "asc"
 
+	// A cursor names one row only when the order ends in the signal's tie-break
+	// columns. Logs already order by timestamp then id. A traces order of
+	// timestamp alone gets trace_id and span_id.
+	rowKeyed := q.spec.Signal == telemetrytypes.SignalLogs || len(q.spec.Order) == 1
+	if q.spec.Signal == telemetrytypes.SignalTraces && rowKeyed {
+		direction := q.spec.Order[0].Direction
+		q.spec.Order = append(slices.Clip(q.spec.Order),
+			qbtypes.OrderBy{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: tracestelemetryschema.SpanTraceIDColumn, FieldContext: telemetrytypes.FieldContextSpan}}, Direction: direction},
+			qbtypes.OrderBy{Key: qbtypes.OrderByKey{TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{Name: tracestelemetryschema.SpanSpanIDColumn, FieldContext: telemetrytypes.FieldContextSpan}}, Direction: direction},
+		)
+	}
+
 	fromMS, toMS := q.fromMS, q.toMS
 
 	// Adjust [fromMS,toMS] window if a cursor was supplied
 	if cur := strings.TrimSpace(q.spec.Cursor); cur != "" {
-		if ts, err := decodeCursor(cur); err == nil {
+		after, legacyMS, err := decodeCursor(cur)
+		if err != nil {
+			return nil, err
+		}
+		switch {
+		case after == nil:
+			// a cursor issued before the row key moves the window edge as it always did
 			if isAsc {
-				if uint64(ts) >= fromMS {
-					fromMS = uint64(ts + 1)
+				if uint64(legacyMS) >= fromMS {
+					fromMS = uint64(legacyMS + 1)
 				}
 			} else { // DESC
-				if uint64(ts) <= toMS {
-					toMS = uint64(ts - 1)
+				if uint64(legacyMS) <= toMS {
+					toMS = uint64(legacyMS - 1)
 				}
 			}
+		case !rowKeyed || (q.spec.Signal == telemetrytypes.SignalLogs && after.ID == "") || (q.spec.Signal == telemetrytypes.SignalTraces && (after.TraceID == "" || after.SpanID == "")):
+			return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "cursor does not belong to this query").
+				WithSuggestions("send the nextCursor with the same signal and order as the query that returned it")
+		default:
+			q.spec.StartAfter = after
+			// the window keeps the whole boundary millisecond; StartAfter cuts inside it
+			boundaryMS := after.TimeUnixNano / uint64(time.Millisecond)
+			if isAsc {
+				fromMS = max(fromMS, boundaryMS)
+			} else {
+				toMS = min(toMS, boundaryMS+1)
+			}
 		}
 	}
 
@@ -637,8 +673,29 @@ func (q *builderQuery[T]) executeWindowList(ctx context.Context) (*qbtypes.Resul
 
 	nextCursor := ""
 	if len(rows) == reqLimit {
-		lastTS := rows[len(rows)-1].Timestamp.UnixMilli()
-		nextCursor = encodeCursor(lastTS)
+		last := rows[len(rows)-1]
+		if rowKeyed {
+			key := qbtypes.RawRowKey{TimeUnixNano: uint64(last.Timestamp.UnixNano())}
+			var ok bool
+			if q.spec.Signal == telemetrytypes.SignalLogs {
+				key.ID, ok = last.Data["id"].(string)
+			} else {
+				key.TraceID, ok = last.Data["trace_id"].(string)
+				if ok {
+					key.SpanID, ok = last.Data["span_id"].(string)
+				}
+			}
+			if !ok {
+				return nil, errors.NewInternalf(errors.CodeInternal, "the last %s row has no order key for the cursor", q.spec.Signal.StringValue())
+			}
+			data, err := json.Marshal(key)
+			if err != nil {
+				return nil, errors.WrapInternalf(err, errors.CodeInternal, "failed to encode the cursor")
+			}
+			nextCursor = base64.RawURLEncoding.EncodeToString(data)
+		} else {
+			nextCursor = encodeCursor(last.Timestamp.UnixMilli())
+		}
 	}
 
 	return &qbtypes.Result{
@@ -662,10 +719,23 @@ func encodeCursor(tsMilli int64) string {
 	return base64.StdEncoding.EncodeToString([]byte(strconv.FormatInt(tsMilli, 10)))
 }
 
-func decodeCursor(cur string) (int64, error) {
-	b, err := base64.StdEncoding.DecodeString(cur)
-	if err != nil {
-		return 0, err
+// decodeCursor reads a raw nextCursor. A cursor issued before the row key
+// holds only the last row's millisecond; it comes back as legacyMS.
+func decodeCursor(cursor string) (*qbtypes.RawRowKey, int64, error) {
+	if data, err := base64.StdEncoding.DecodeString(cursor); err == nil {
+		if legacyMS, err := strconv.ParseInt(string(data), 10, 64); err == nil {
+			return nil, legacyMS, nil
+		}
 	}
-	return strconv.ParseInt(string(b), 10, 64)
+
+	after := &qbtypes.RawRowKey{}
+	data, err := base64.RawURLEncoding.DecodeString(cursor)
+	if err == nil {
+		err = json.Unmarshal(data, after)
+	}
+	if err != nil || after.TimeUnixNano == 0 {
+		return nil, 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid cursor %q", cursor).
+			WithSuggestions("send the nextCursor of the previous page unchanged")
+	}
+	return after, 0, nil
 }

pkg/querybuilder/rows_after.go

@@ -0,0 +1,20 @@
+package querybuilder
+
+import "github.com/huandu/go-sqlbuilder"
+
+// RowsAfter returns the condition for the rows strictly after one row in a
+// lexicographic order over columns, with every column in the same direction.
+// values holds that row's value for each column. The caller keeps its time
+// window, so granule pruning does not depend on this condition.
+func RowsAfter(sb *sqlbuilder.SelectBuilder, columns []string, values []any, descending bool) string {
+	compare := sb.G
+	if descending {
+		compare = sb.L
+	}
+	last := len(columns) - 1
+	condition := compare(columns[last], values[last])
+	for i := last - 1; i >= 0; i-- {
+		condition = sb.Or(compare(columns[i], values[i]), sb.And(sb.E(columns[i], values[i]), condition))
+	}
+	return condition
+}

pkg/statementbuilder/logsstatementbuilder/statement_builder.go

@@ -393,6 +393,13 @@ func (b *logQueryStatementBuilder) buildListQuery(
 		return nil, err
 	}
 
+	if after := query.StartAfter; after != nil {
+		sb.Where(querybuilder.RowsAfter(sb,
+			[]string{logstelemetryschema.LogsV2TimestampColumn, logstelemetryschema.LogsV2IDColumn},
+			[]any{fmt.Sprintf("%d", after.TimeUnixNano), after.ID},
+			query.Order[0].Direction == qbtypes.OrderDirectionDesc))
+	}
+
 	// Add order by
 	for _, orderBy := range query.Order {
 

pkg/statementbuilder/tracesstatementbuilder/statement_builder.go

@@ -404,6 +404,13 @@ func (b *traceQueryStatementBuilder) buildListQuery(
 		return nil, err
 	}
 
+	if after := query.StartAfter; after != nil {
+		sb.Where(querybuilder.RowsAfter(sb,
+			[]string{tracestelemetryschema.SpanTimestampColumn, tracestelemetryschema.SpanTraceIDColumn, tracestelemetryschema.SpanSpanIDColumn},
+			[]any{fmt.Sprintf("%d", after.TimeUnixNano), after.TraceID, after.SpanID},
+			query.Order[0].Direction == qbtypes.OrderDirectionDesc))
+	}
+
 	// Add order by
 	for _, orderBy := range query.Order {
 		expr, err := querybuilder.ResolveColumn(ctx, info, b.storage, &orderBy.Key.TelemetryFieldKey, telemetrytypes.FieldDataTypeUnspecified, keys)

pkg/types/querybuildertypes/querybuildertypesv5/builder_query.go

@@ -53,6 +53,20 @@ type QueryBuilderQuery[T any] struct {
 	// ShiftBy is extracted from timeShift function for internal use
 	// This field is not serialized to JSON
 	ShiftBy int64 `json:"-"`
+
+	// StartAfter is the row a raw page continues from, decoded from Cursor.
+	// The page holds only rows strictly after it in the requested order.
+	StartAfter *RawRowKey `json:"-"`
+}
+
+// RawRowKey is the full order key of one raw row: the timestamp, then the
+// signal's tie-break columns, id for logs, trace_id and span_id for traces.
+// It is the content of a raw nextCursor.
+type RawRowKey struct {
+	TimeUnixNano uint64 `json:"timeUnixNano"`
+	ID           string `json:"id,omitempty"`
+	TraceID      string `json:"traceId,omitempty"`
+	SpanID       string `json:"spanId,omitempty"`
 }
 
 // PrepareJSONSchema pins `signal` to the single value implied by the aggregation
@@ -143,6 +157,11 @@ func (q QueryBuilderQuery[T]) Copy() QueryBuilderQuery[T] {
 		c.BucketOptions = &bucketOptionsCopy
 	}
 
+	if q.StartAfter != nil {
+		startAfterCopy := *q.StartAfter
+		c.StartAfter = &startAfterCopy
+	}
+
 	return c
 }
 

pkg/types/querybuildertypes/querybuildertypesv5/validation.go

@@ -437,6 +437,13 @@ func (q *QueryBuilderQuery[T]) validateLimitAndPagination(cfg validationConfig)
 		)
 	}
 
+	// the cursor already starts after the last row returned, so an offset
+	// would skip rows nobody has seen
+	if q.Cursor != "" && q.Offset > 0 {
+		return errors.NewInvalidInputf(errors.CodeInvalidInput, "offset cannot be combined with cursor, got offset %d", q.Offset).
+			WithSuggestions("send offset 0 with the cursor; the cursor continues after the last row of the previous page")
+	}
+
 	return nil
 }
 

tests/fixtures/querier.py

@@ -82,6 +82,7 @@ class BuilderQuery:
     query_type: str = "builder_query"
     limit: int | None = None
     offset: int | None = None
+    cursor: str | None = None
     filter_expression: str | None = None
     having_expression: str | None = None
     select_fields: list[TelemetryFieldKey] | None = None
@@ -101,6 +102,8 @@ class BuilderQuery:
             spec["limit"] = self.limit
         if self.offset is not None:
             spec["offset"] = self.offset
+        if self.cursor:
+            spec["cursor"] = self.cursor
         if self.filter_expression:
             spec["filter"] = {"expression": self.filter_expression}
         if self.having_expression:

tests/integration/tests/querierlogs/01_list.py

@@ -9,7 +9,12 @@ from fixtures import types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logs import Logs
 from fixtures.querier import (
+    BuilderQuery,
+    OrderBy,
+    RequestType,
+    TelemetryFieldKey,
     assert_identical_query_response,
+    get_rows,
     make_query_request,
 )
 
@@ -498,3 +503,103 @@ def test_logs_list_with_order_by(
     ids = [row["data"]["attributes_string"].get("id", "") for row in rows]
 
     assert ids == expected_order
+
+
+@pytest.mark.parametrize("direction", ["desc", "asc"])
+def test_logs_list_cursor_pages_every_row_once(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+    direction: str,
+) -> None:
+    """
+    Setup:
+    7 logs: one alone in its millisecond at each end, 5 inside one millisecond,
+    3 of them at the same instant.
+
+    Tests:
+    Paging by nextCursor with limit 2 returns every log once, ordered by
+    timestamp then id, although page boundaries fall inside the shared
+    millisecond and inside the shared instant.
+    """
+    now = datetime.now(tz=UTC).replace(microsecond=0)
+    shared = now - timedelta(seconds=5) + timedelta(microseconds=300)
+    instants = [
+        now - timedelta(seconds=6),
+        shared,
+        shared + timedelta(microseconds=1),
+        shared + timedelta(microseconds=2),
+        shared + timedelta(microseconds=2),
+        shared + timedelta(microseconds=2),
+        now - timedelta(seconds=4),
+    ]
+    logs = [Logs(timestamp=instant, resources={"service.name": "cursor-service"}, body=f"log-{i}") for i, instant in enumerate(instants)]
+    insert_logs(logs)
+    expected = [log.id for log in sorted(logs, key=lambda log: (int(log.timestamp), log.id), reverse=direction == "desc")]
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    seen: list[str] = []
+    cursor = None
+    for _ in range(len(logs) + 1):
+        response = make_query_request(
+            signoz,
+            token,
+            start_ms=int((now - timedelta(minutes=1)).timestamp() * 1000),
+            end_ms=int((now + timedelta(seconds=1)).timestamp() * 1000),
+            request_type=RequestType.RAW,
+            queries=[
+                BuilderQuery(
+                    signal="logs",
+                    limit=2,
+                    cursor=cursor,
+                    filter_expression="service.name = 'cursor-service'",
+                    order=[OrderBy(TelemetryFieldKey("timestamp"), direction), OrderBy(TelemetryFieldKey("id"), direction)],
+                ).to_dict()
+            ],
+        )
+        assert response.status_code == HTTPStatus.OK, response.text
+        seen += [row["data"]["id"] for row in get_rows(response)]
+        cursor = response.json()["data"]["data"]["results"][0].get("nextCursor")
+        if not cursor:
+            break
+
+    assert seen == expected
+
+
+@pytest.mark.parametrize(
+    "order,offset,cursor,message",
+    [
+        pytest.param(["timestamp", "id"], 2, "MTc5MzA0MDgzNjc5NQ==", "offset cannot be combined with cursor, got offset 2", id="offset_with_cursor"),
+        pytest.param(["timestamp"], 0, "MTc5MzA0MDgzNjc5NQ==", "cursor needs a raw query ordered by timestamp first", id="order_without_id"),
+        pytest.param(["timestamp", "id"], 0, "not-a-cursor", 'invalid cursor "not-a-cursor"', id="invalid_cursor"),
+    ],
+)
+def test_logs_list_cursor_rejections(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    order: list[str],
+    offset: int,
+    cursor: str,
+    message: str,
+) -> None:
+    now = datetime.now(tz=UTC)
+    response = make_query_request(
+        signoz,
+        get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD),
+        start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
+        end_ms=int(now.timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            BuilderQuery(
+                signal="logs",
+                limit=2,
+                offset=offset,
+                cursor=cursor,
+                order=[OrderBy(TelemetryFieldKey(name), "desc") for name in order],
+            ).to_dict()
+        ],
+    )
+    assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
+    assert message in response.json()["error"]["message"], response.text

tests/integration/tests/queriertraces/01_list.py

@@ -1,3 +1,4 @@
+import base64
 from collections.abc import Callable
 from datetime import UTC, datetime, timedelta
 from http import HTTPStatus
@@ -917,6 +918,136 @@ def test_traces_list_pagination(
     assert [row["data"]["span_id"] for row in rows] == expected_ids
 
 
+@pytest.mark.parametrize("direction", ["desc", "asc"])
+def test_traces_list_cursor_pages_every_row_once(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+    direction: str,
+) -> None:
+    """
+    Setup:
+    7 spans: one alone in its millisecond at each end, 5 inside one millisecond,
+    3 of them at the same instant.
+
+    Tests:
+    Paging by nextCursor with limit 2 returns every span once, ordered by
+    timestamp then trace_id and span_id, although page boundaries fall inside
+    the shared millisecond and inside the shared instant.
+    """
+    now = datetime.now(tz=UTC).replace(microsecond=0)
+    shared = now - timedelta(seconds=5) + timedelta(microseconds=300)
+    instants = [
+        now - timedelta(seconds=6),
+        shared,
+        shared + timedelta(microseconds=1),
+        shared + timedelta(microseconds=2),
+        shared + timedelta(microseconds=2),
+        shared + timedelta(microseconds=2),
+        now - timedelta(seconds=4),
+    ]
+    spans = [
+        Traces(
+            timestamp=instant,
+            trace_id=TraceIdGenerator.trace_id(),
+            span_id=TraceIdGenerator.span_id(),
+            name=f"span-{i}",
+            resources={"service.name": "cursor-service"},
+        )
+        for i, instant in enumerate(instants)
+    ]
+    insert_traces(spans)
+    expected = [span.span_id for span in sorted(spans, key=lambda span: (span.timestamp, span.trace_id, span.span_id), reverse=direction == "desc")]
+
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    seen: list[str] = []
+    cursor = None
+    for _ in range(len(spans) + 1):
+        response = make_query_request(
+            signoz,
+            token,
+            start_ms=int((now - timedelta(minutes=1)).timestamp() * 1000),
+            end_ms=int((now + timedelta(seconds=1)).timestamp() * 1000),
+            request_type=RequestType.RAW,
+            queries=[
+                BuilderQuery(
+                    signal="traces",
+                    limit=2,
+                    cursor=cursor,
+                    filter_expression="service.name = 'cursor-service'",
+                    order=[OrderBy(TelemetryFieldKey("timestamp"), direction)],
+                ).to_dict()
+            ],
+        )
+        assert response.status_code == HTTPStatus.OK, response.text
+        seen += [row["data"]["span_id"] for row in get_rows(response)]
+        cursor = response.json()["data"]["data"]["results"][0].get("nextCursor")
+        if not cursor:
+            break
+
+    assert seen == expected
+
+
+@pytest.mark.parametrize(
+    "direction,boundary,expected",
+    [
+        pytest.param("desc", 2, [1, 0], id="desc"),
+        pytest.param("asc", 1, [2, 3], id="asc"),
+    ],
+)
+def test_traces_list_millisecond_cursor_still_pages(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_traces: Callable[[list[Traces]], None],
+    direction: str,
+    boundary: int,
+    expected: list[int],
+) -> None:
+    """
+    Setup:
+    4 spans 10 ms apart.
+
+    Tests:
+    A cursor in the earlier format, the base64 of a millisecond, still continues
+    from that millisecond as it did, so a page a client holds before an upgrade
+    keeps working.
+    """
+    now = datetime.now(tz=UTC).replace(microsecond=0)
+    spans = [
+        Traces(
+            timestamp=now - timedelta(seconds=5) + timedelta(milliseconds=10 * i),
+            trace_id=TraceIdGenerator.trace_id(),
+            span_id=TraceIdGenerator.span_id(),
+            name=f"span-{i}",
+            resources={"service.name": "legacy-cursor-service"},
+        )
+        for i in range(4)
+    ]
+    insert_traces(spans)
+    cursor = base64.b64encode(str(int(spans[boundary].timestamp.timestamp() * 1000)).encode()).decode()
+
+    response = make_query_request(
+        signoz,
+        get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD),
+        start_ms=int((now - timedelta(minutes=1)).timestamp() * 1000),
+        end_ms=int((now + timedelta(seconds=1)).timestamp() * 1000),
+        request_type=RequestType.RAW,
+        queries=[
+            BuilderQuery(
+                signal="traces",
+                limit=10,
+                cursor=cursor,
+                filter_expression="service.name = 'legacy-cursor-service'",
+                order=[OrderBy(TelemetryFieldKey("timestamp"), direction)],
+            ).to_dict()
+        ],
+    )
+    assert response.status_code == HTTPStatus.OK, response.text
+    assert [row["data"]["span_id"] for row in get_rows(response)] == [spans[i].span_id for i in expected]
+
+
 # ============================================================================
 # Time window
 # ============================================================================