mirror of
https://github.com/SigNoz/signoz.git
synced 2026-11-05 16:00:41 +00:00
Compare commits
1 Commits
fix/body-p
...
nv/offset-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4813da9556 |
@@ -4027,7 +4027,9 @@ components:
|
||||
the query they draw.
|
||||
cursor:
|
||||
description: The opaque nextCursor value from a previous raw result, used
|
||||
to fetch the next page.
|
||||
to fetch the next page. Honored only by raw requests whose order starts
|
||||
with timestamp, for logs followed by id in the same direction; every other
|
||||
request ignores it.
|
||||
type: string
|
||||
disabled:
|
||||
description: When true, the result is excluded from the response; the query
|
||||
@@ -4064,10 +4066,11 @@ components:
|
||||
are replaced by group-by label values.
|
||||
type: string
|
||||
limit:
|
||||
description: The maximum number of rows for raw requests, or of series or
|
||||
rows for grouped aggregations, keeping the top entries by order. When
|
||||
0, aggregations are unlimited while raw requests ordered by timestamp
|
||||
return a page of 10000 rows; follow nextCursor for the rest.
|
||||
description: The maximum number of rows for raw and trace requests, or of
|
||||
series or rows for grouped scalar and time_series requests, keeping the
|
||||
top entries by order. When 0, scalar and time_series requests are unlimited,
|
||||
raw requests ordered by timestamp return a page of 10000 rows to continue
|
||||
with nextCursor, and other raw and trace requests return 100 rows.
|
||||
type: integer
|
||||
limitBy:
|
||||
allOf:
|
||||
@@ -4079,8 +4082,10 @@ components:
|
||||
operators and selectedQueryName refer to it.
|
||||
type: string
|
||||
offset:
|
||||
description: The number of rows to skip before returning results. Prefer
|
||||
cursor for raw requests.
|
||||
description: The number of rows for raw and trace requests, or of ordered
|
||||
series or rows for scalar and time_series requests, to skip before the
|
||||
page starts. On raw requests the skip counts from where a cursor resumes,
|
||||
and the skipped rows are still read, so prefer cursor for paging.
|
||||
type: integer
|
||||
order:
|
||||
description: The sort order of the result. Order by a group-by field, an
|
||||
@@ -10102,6 +10107,10 @@ components:
|
||||
description: The unique name of the formula, such as F1, used by selectedQueryName
|
||||
and by panels. Referencing a formula from another formula is not reliable.
|
||||
type: string
|
||||
offset:
|
||||
description: The number of ordered series or rows to skip before the page
|
||||
starts.
|
||||
type: integer
|
||||
order:
|
||||
description: The sort order of the formula result.
|
||||
items:
|
||||
@@ -10127,7 +10136,9 @@ components:
|
||||
the query they draw.
|
||||
cursor:
|
||||
description: The opaque nextCursor value from a previous raw result, used
|
||||
to fetch the next page.
|
||||
to fetch the next page. Honored only by raw requests whose order starts
|
||||
with timestamp, for logs followed by id in the same direction; every other
|
||||
request ignores it.
|
||||
type: string
|
||||
disabled:
|
||||
description: When true, the result is excluded from the response; the query
|
||||
@@ -10163,10 +10174,11 @@ components:
|
||||
are replaced by group-by label values.
|
||||
type: string
|
||||
limit:
|
||||
description: The maximum number of rows for raw requests, or of series or
|
||||
rows for grouped aggregations, keeping the top entries by order. When
|
||||
0, aggregations are unlimited while raw requests ordered by timestamp
|
||||
return a page of 10000 rows; follow nextCursor for the rest.
|
||||
description: The maximum number of rows for raw and trace requests, or of
|
||||
series or rows for grouped scalar and time_series requests, keeping the
|
||||
top entries by order. When 0, scalar and time_series requests are unlimited,
|
||||
raw requests ordered by timestamp return a page of 10000 rows to continue
|
||||
with nextCursor, and other raw and trace requests return 100 rows.
|
||||
type: integer
|
||||
limitBy:
|
||||
allOf:
|
||||
@@ -10178,8 +10190,10 @@ components:
|
||||
operators and selectedQueryName refer to it.
|
||||
type: string
|
||||
offset:
|
||||
description: The number of rows to skip before returning results. Prefer
|
||||
cursor for raw requests.
|
||||
description: The number of rows for raw and trace requests, or of ordered
|
||||
series or rows for scalar and time_series requests, to skip before the
|
||||
page starts. On raw requests the skip counts from where a cursor resumes,
|
||||
and the skipped rows are still read, so prefer cursor for paging.
|
||||
type: integer
|
||||
order:
|
||||
description: The sort order of the result. Order by a group-by field, an
|
||||
@@ -10241,7 +10255,9 @@ components:
|
||||
the query they draw.
|
||||
cursor:
|
||||
description: The opaque nextCursor value from a previous raw result, used
|
||||
to fetch the next page.
|
||||
to fetch the next page. Honored only by raw requests whose order starts
|
||||
with timestamp, for logs followed by id in the same direction; every other
|
||||
request ignores it.
|
||||
type: string
|
||||
disabled:
|
||||
description: When true, the result is excluded from the response; the query
|
||||
@@ -10278,10 +10294,11 @@ components:
|
||||
are replaced by group-by label values.
|
||||
type: string
|
||||
limit:
|
||||
description: The maximum number of rows for raw requests, or of series or
|
||||
rows for grouped aggregations, keeping the top entries by order. When
|
||||
0, aggregations are unlimited while raw requests ordered by timestamp
|
||||
return a page of 10000 rows; follow nextCursor for the rest.
|
||||
description: The maximum number of rows for raw and trace requests, or of
|
||||
series or rows for grouped scalar and time_series requests, keeping the
|
||||
top entries by order. When 0, scalar and time_series requests are unlimited,
|
||||
raw requests ordered by timestamp return a page of 10000 rows to continue
|
||||
with nextCursor, and other raw and trace requests return 100 rows.
|
||||
type: integer
|
||||
limitBy:
|
||||
allOf:
|
||||
@@ -10293,8 +10310,10 @@ components:
|
||||
operators and selectedQueryName refer to it.
|
||||
type: string
|
||||
offset:
|
||||
description: The number of rows to skip before returning results. Prefer
|
||||
cursor for raw requests.
|
||||
description: The number of rows for raw and trace requests, or of ordered
|
||||
series or rows for scalar and time_series requests, to skip before the
|
||||
page starts. On raw requests the skip counts from where a cursor resumes,
|
||||
and the skipped rows are still read, so prefer cursor for paging.
|
||||
type: integer
|
||||
order:
|
||||
description: The sort order of the result. Order by a group-by field, an
|
||||
@@ -10356,7 +10375,9 @@ components:
|
||||
the query they draw.
|
||||
cursor:
|
||||
description: The opaque nextCursor value from a previous raw result, used
|
||||
to fetch the next page.
|
||||
to fetch the next page. Honored only by raw requests whose order starts
|
||||
with timestamp, for logs followed by id in the same direction; every other
|
||||
request ignores it.
|
||||
type: string
|
||||
disabled:
|
||||
description: When true, the result is excluded from the response; the query
|
||||
@@ -10393,10 +10414,11 @@ components:
|
||||
are replaced by group-by label values.
|
||||
type: string
|
||||
limit:
|
||||
description: The maximum number of rows for raw requests, or of series or
|
||||
rows for grouped aggregations, keeping the top entries by order. When
|
||||
0, aggregations are unlimited while raw requests ordered by timestamp
|
||||
return a page of 10000 rows; follow nextCursor for the rest.
|
||||
description: The maximum number of rows for raw and trace requests, or of
|
||||
series or rows for grouped scalar and time_series requests, keeping the
|
||||
top entries by order. When 0, scalar and time_series requests are unlimited,
|
||||
raw requests ordered by timestamp return a page of 10000 rows to continue
|
||||
with nextCursor, and other raw and trace requests return 100 rows.
|
||||
type: integer
|
||||
limitBy:
|
||||
allOf:
|
||||
@@ -10408,8 +10430,10 @@ components:
|
||||
operators and selectedQueryName refer to it.
|
||||
type: string
|
||||
offset:
|
||||
description: The number of rows to skip before returning results. Prefer
|
||||
cursor for raw requests.
|
||||
description: The number of rows for raw and trace requests, or of ordered
|
||||
series or rows for scalar and time_series requests, to skip before the
|
||||
page starts. On raw requests the skip counts from where a cursor resumes,
|
||||
and the skipped rows are still read, so prefer cursor for paging.
|
||||
type: integer
|
||||
order:
|
||||
description: The sort order of the result. Order by a group-by field, an
|
||||
@@ -10510,8 +10534,8 @@ components:
|
||||
description: The unique name of the trace operator query.
|
||||
type: string
|
||||
offset:
|
||||
description: The number of raw results to skip before returning. Aggregation
|
||||
requests ignore it.
|
||||
description: The number of raw or trace rows, or of ordered series or rows
|
||||
for time_series and scalar requests, to skip before the page starts.
|
||||
type: integer
|
||||
order:
|
||||
description: The sort order of the matched traces; only span_count and trace_duration
|
||||
|
||||
@@ -5273,7 +5273,7 @@ export interface DashboardtypesAIBuilderQuerySpecDTO {
|
||||
bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
|
||||
/**
|
||||
* @type string
|
||||
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
|
||||
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
|
||||
*/
|
||||
cursor?: string;
|
||||
/**
|
||||
@@ -5306,7 +5306,7 @@ export interface DashboardtypesAIBuilderQuerySpecDTO {
|
||||
legend?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
|
||||
* @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
|
||||
*/
|
||||
limit?: number;
|
||||
/**
|
||||
@@ -5320,7 +5320,7 @@ export interface DashboardtypesAIBuilderQuerySpecDTO {
|
||||
name?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description The number of rows to skip before returning results. Prefer cursor for raw requests.
|
||||
* @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
|
||||
*/
|
||||
offset?: number;
|
||||
/**
|
||||
@@ -5665,7 +5665,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
|
||||
bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
|
||||
/**
|
||||
* @type string
|
||||
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
|
||||
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
|
||||
*/
|
||||
cursor?: string;
|
||||
/**
|
||||
@@ -5698,7 +5698,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
|
||||
legend?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
|
||||
* @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
|
||||
*/
|
||||
limit?: number;
|
||||
/**
|
||||
@@ -5712,7 +5712,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
|
||||
name?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description The number of rows to skip before returning results. Prefer cursor for raw requests.
|
||||
* @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
|
||||
*/
|
||||
offset?: number;
|
||||
/**
|
||||
@@ -5845,7 +5845,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
|
||||
bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
|
||||
/**
|
||||
* @type string
|
||||
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
|
||||
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
|
||||
*/
|
||||
cursor?: string;
|
||||
/**
|
||||
@@ -5878,7 +5878,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
|
||||
legend?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
|
||||
* @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
|
||||
*/
|
||||
limit?: number;
|
||||
/**
|
||||
@@ -5892,7 +5892,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
|
||||
name?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description The number of rows to skip before returning results. Prefer cursor for raw requests.
|
||||
* @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
|
||||
*/
|
||||
offset?: number;
|
||||
/**
|
||||
@@ -5944,7 +5944,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
|
||||
bucketOptions?: Querybuildertypesv5BucketOptionsDTO;
|
||||
/**
|
||||
* @type string
|
||||
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page.
|
||||
* @description The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it.
|
||||
*/
|
||||
cursor?: string;
|
||||
/**
|
||||
@@ -5977,7 +5977,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
|
||||
legend?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest.
|
||||
* @description The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows.
|
||||
*/
|
||||
limit?: number;
|
||||
/**
|
||||
@@ -5991,7 +5991,7 @@ export interface Querybuildertypesv5QueryBuilderQueryGithubComSigNozSignozPkgTyp
|
||||
name?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description The number of rows to skip before returning results. Prefer cursor for raw requests.
|
||||
* @description The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging.
|
||||
*/
|
||||
offset?: number;
|
||||
/**
|
||||
@@ -7064,6 +7064,11 @@ export interface Querybuildertypesv5QueryBuilderFormulaDTO {
|
||||
* @description The unique name of the formula, such as F1, used by selectedQueryName and by panels. Referencing a formula from another formula is not reliable.
|
||||
*/
|
||||
name?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description The number of ordered series or rows to skip before the page starts.
|
||||
*/
|
||||
offset?: number;
|
||||
/**
|
||||
* @type array,null
|
||||
* @description The sort order of the formula result.
|
||||
@@ -7146,7 +7151,7 @@ export interface Querybuildertypesv5QueryBuilderTraceOperatorDTO {
|
||||
name?: string;
|
||||
/**
|
||||
* @type integer
|
||||
* @description The number of raw results to skip before returning. Aggregation requests ignore it.
|
||||
* @description The number of raw or trace rows, or of ordered series or rows for time_series and scalar requests, to skip before the page starts.
|
||||
*/
|
||||
offset?: number;
|
||||
/**
|
||||
|
||||
@@ -182,7 +182,7 @@ func postProcessBuilderQuery[T any](
|
||||
req *qbtypes.QueryRangeRequest,
|
||||
) *qbtypes.Result {
|
||||
|
||||
result = q.applySeriesLimit(result, query.Limit, query.Order)
|
||||
result = q.applySeriesLimit(result, query.Limit, query.Offset, query.Order)
|
||||
|
||||
// Apply functions
|
||||
if len(query.Functions) > 0 {
|
||||
@@ -234,7 +234,7 @@ func postProcessMetricQuery(
|
||||
reducesToScalar := req.RequestType == qbtypes.RequestTypeScalar && config.ReduceTo != qbtypes.ReduceToUnknown
|
||||
|
||||
if !reducesToScalar {
|
||||
result = q.applySeriesLimit(result, query.Limit, query.Order)
|
||||
result = q.applySeriesLimit(result, query.Limit, query.Offset, query.Order)
|
||||
}
|
||||
|
||||
if len(query.Functions) > 0 {
|
||||
@@ -246,7 +246,7 @@ func postProcessMetricQuery(
|
||||
// Apply reduce to for scalar request type
|
||||
if reducesToScalar {
|
||||
result = q.applyMetricReduceTo(result, config.ReduceTo)
|
||||
result = q.applySeriesLimit(result, query.Limit, query.Order)
|
||||
result = q.applySeriesLimit(result, query.Limit, query.Offset, query.Order)
|
||||
if tsData, ok := result.Value.(*qbtypes.TimeSeriesData); ok {
|
||||
result.Value = convertTimeSeriesDataToScalar(tsData, tsData.QueryName)
|
||||
}
|
||||
@@ -263,7 +263,7 @@ func postProcessTraceOperator(
|
||||
req *qbtypes.QueryRangeRequest,
|
||||
) *qbtypes.Result {
|
||||
|
||||
result = q.applySeriesLimit(result, query.Limit, query.Order)
|
||||
result = q.applySeriesLimit(result, query.Limit, query.Offset, query.Order)
|
||||
|
||||
// Apply functions if any
|
||||
if len(query.Functions) > 0 {
|
||||
@@ -296,7 +296,7 @@ func (q *querier) applyMetricReduceTo(result *qbtypes.Result, reduceOp qbtypes.R
|
||||
}
|
||||
|
||||
// applySeriesLimit limits the number of series in the result.
|
||||
func (q *querier) applySeriesLimit(result *qbtypes.Result, limit int, orderBy []qbtypes.OrderBy) *qbtypes.Result {
|
||||
func (q *querier) applySeriesLimit(result *qbtypes.Result, limit, offset int, orderBy []qbtypes.OrderBy) *qbtypes.Result {
|
||||
tsData, ok := result.Value.(*qbtypes.TimeSeriesData)
|
||||
if !ok {
|
||||
return result
|
||||
@@ -305,7 +305,7 @@ func (q *querier) applySeriesLimit(result *qbtypes.Result, limit int, orderBy []
|
||||
if tsData != nil {
|
||||
for _, agg := range tsData.Aggregations {
|
||||
// Use the ApplySeriesLimit function from querybuildertypes
|
||||
agg.Series = qbtypes.ApplySeriesLimit(agg.Series, orderBy, limit)
|
||||
agg.Series = qbtypes.ApplySeriesLimit(agg.Series, orderBy, limit, offset)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -357,7 +357,7 @@ func (q *querier) applyFormulas(ctx context.Context, results map[string]*qbtypes
|
||||
case qbtypes.RequestTypeTimeSeries:
|
||||
result := q.processTimeSeriesFormula(ctx, results, formula, req)
|
||||
if result != nil {
|
||||
result = q.applySeriesLimit(result, formula.Limit, formula.Order)
|
||||
result = q.applySeriesLimit(result, formula.Limit, formula.Offset, formula.Order)
|
||||
results[name] = result
|
||||
}
|
||||
case qbtypes.RequestTypeHeatmap:
|
||||
@@ -370,7 +370,7 @@ func (q *querier) applyFormulas(ctx context.Context, results map[string]*qbtypes
|
||||
bucketFormulaOutputAsHeatmap(tsData, bucketing)
|
||||
resolveHeatmapBucketAxis(tsData, bucketing)
|
||||
}
|
||||
result = q.applySeriesLimit(result, formula.Limit, formula.Order)
|
||||
result = q.applySeriesLimit(result, formula.Limit, formula.Offset, formula.Order)
|
||||
results[name] = result
|
||||
}
|
||||
case qbtypes.RequestTypeScalar:
|
||||
@@ -651,7 +651,7 @@ func (q *querier) processScalarFormula(
|
||||
}
|
||||
|
||||
// Apply ordering (and limit) before converting to scalar format.
|
||||
formulaSeries = qbtypes.ApplySeriesLimit(formulaSeries, formula.Order, formula.Limit)
|
||||
formulaSeries = qbtypes.ApplySeriesLimit(formulaSeries, formula.Order, formula.Limit, formula.Offset)
|
||||
|
||||
// Convert back to scalar format
|
||||
scalarResult := &qbtypes.ScalarData{
|
||||
|
||||
@@ -186,6 +186,7 @@ func TestPostProcessMetricQuery_LimitRanksByReducedValue(t *testing.T) {
|
||||
name string
|
||||
reduceTo qbtypes.ReduceTo
|
||||
order []qbtypes.OrderBy
|
||||
offset int
|
||||
expected [][]any
|
||||
}{
|
||||
{
|
||||
@@ -211,6 +212,13 @@ func TestPostProcessMetricQuery_LimitRanksByReducedValue(t *testing.T) {
|
||||
reduceTo: qbtypes.ReduceToMax,
|
||||
expected: [][]any{{"payment", 40.0}, {"auth", 30.0}},
|
||||
},
|
||||
{
|
||||
name: "Max_Desc_Offset1_SkipsHighestPeak",
|
||||
reduceTo: qbtypes.ReduceToMax,
|
||||
order: orderByValue(qbtypes.OrderDirectionDesc),
|
||||
offset: 1,
|
||||
expected: [][]any{{"auth", 30.0}, {"checkout", 10.0}},
|
||||
},
|
||||
}
|
||||
|
||||
for _, testCase := range testCases {
|
||||
@@ -220,6 +228,7 @@ func TestPostProcessMetricQuery_LimitRanksByReducedValue(t *testing.T) {
|
||||
Aggregations: []qbtypes.MetricAggregation{{MetricName: "http_requests_total", ReduceTo: testCase.reduceTo}},
|
||||
Order: testCase.order,
|
||||
Limit: 2,
|
||||
Offset: testCase.offset,
|
||||
}
|
||||
req := &qbtypes.QueryRangeRequest{RequestType: qbtypes.RequestTypeScalar}
|
||||
|
||||
|
||||
17
pkg/querybuilder/offset.go
Normal file
17
pkg/querybuilder/offset.go
Normal file
@@ -0,0 +1,17 @@
|
||||
package querybuilder
|
||||
|
||||
import "github.com/huandu/go-sqlbuilder"
|
||||
|
||||
// AddOffset skips offset rows. Call it after the ORDER BY and LIMIT clauses:
|
||||
// the ClickHouse flavor renders OFFSET only inside a LIMIT, so an unlimited
|
||||
// page writes the clause itself at the current position.
|
||||
func AddOffset(sb *sqlbuilder.SelectBuilder, limit, offset int) {
|
||||
if offset <= 0 {
|
||||
return
|
||||
}
|
||||
if limit > 0 {
|
||||
sb.Offset(offset)
|
||||
return
|
||||
}
|
||||
sb.SQL("OFFSET " + sb.Var(offset))
|
||||
}
|
||||
@@ -374,7 +374,11 @@ func (b *auditQueryStatementBuilder) buildTimeSeriesQuery(
|
||||
|
||||
if query.Limit > 0 && len(query.GroupBy) > 0 {
|
||||
cteSB := sqlbuilder.NewSelectBuilder()
|
||||
cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, query, start, end, keys, true, variables)
|
||||
// the series skip happens after the query, so the top-N must still include the skipped groups
|
||||
cteQuery := query
|
||||
cteQuery.Limit += query.Offset
|
||||
cteQuery.Offset = 0
|
||||
cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, cteQuery, start, end, keys, true, variables)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -536,6 +540,8 @@ func (b *auditQueryStatementBuilder) buildScalarQuery(
|
||||
sb.Limit(query.Limit)
|
||||
}
|
||||
|
||||
querybuilder.AddOffset(sb, query.Limit, query.Offset)
|
||||
|
||||
combinedArgs := append(allGroupByArgs, allAggChArgs...)
|
||||
|
||||
mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
|
||||
|
||||
@@ -506,7 +506,11 @@ func (b *logQueryStatementBuilder) buildTimeSeriesQuery(
|
||||
if query.Limit > 0 && len(query.GroupBy) > 0 {
|
||||
// build the scalar “top/bottom-N” query in its own builder.
|
||||
cteSB := sqlbuilder.NewSelectBuilder()
|
||||
cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, query, start, end, keys, true, variables)
|
||||
// the series skip happens after the query, so the top-N must still include the skipped groups
|
||||
cteQuery := query
|
||||
cteQuery.Limit += query.Offset
|
||||
cteQuery.Offset = 0
|
||||
cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, cteQuery, start, end, keys, true, variables)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -707,6 +711,8 @@ func (b *logQueryStatementBuilder) buildScalarQuery(
|
||||
sb.Limit(query.Limit)
|
||||
}
|
||||
|
||||
querybuilder.AddOffset(sb, query.Limit, query.Offset)
|
||||
|
||||
combinedArgs := allAggChArgs
|
||||
|
||||
mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
|
||||
|
||||
@@ -654,7 +654,10 @@ func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
|
||||
cteFragments = append(cteFragments, fmt.Sprintf("__scoped_traces_total AS (%s)", totalSQL))
|
||||
cteArgs = append(cteArgs, totalArgs)
|
||||
|
||||
limitSQL, limitArgs := outerLimitSQL(query, traceAggs, groupNames, windowSeconds)
|
||||
// the series skip happens after the query, so the top-N must still include the skipped groups
|
||||
cteQuery := query
|
||||
cteQuery.Limit += query.Offset
|
||||
limitSQL, limitArgs := outerLimitSQL(cteQuery, traceAggs, groupNames, windowSeconds)
|
||||
cteFragments = append(cteFragments, fmt.Sprintf("__limit_cte AS (%s)", limitSQL))
|
||||
cteArgs = append(cteArgs, limitArgs)
|
||||
|
||||
@@ -722,6 +725,7 @@ func (b *scopedTraceStatementBuilder) buildTraceAggregationQuery(
|
||||
if query.Limit > 0 {
|
||||
sb.Limit(query.Limit)
|
||||
}
|
||||
querybuilder.AddOffset(sb, query.Limit, query.Offset)
|
||||
}
|
||||
|
||||
mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse)
|
||||
|
||||
@@ -632,7 +632,11 @@ func (b *traceQueryStatementBuilder) buildTimeSeriesQuery(
|
||||
if query.Limit > 0 && len(query.GroupBy) > 0 {
|
||||
// build the scalar “top/bottom-N” query in its own builder.
|
||||
cteSB := sqlbuilder.NewSelectBuilder()
|
||||
cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, query, start, end, keys, variables, true, true)
|
||||
// the series skip happens after the query, so the top-N must still include the skipped groups
|
||||
cteQuery := query
|
||||
cteQuery.Limit += query.Offset
|
||||
cteQuery.Offset = 0
|
||||
cteStmt, err := b.buildScalarQuery(ctx, orgID, cteSB, cteQuery, start, end, keys, variables, true, true)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -835,6 +839,8 @@ func (b *traceQueryStatementBuilder) buildScalarQuery(
|
||||
sb.Limit(query.Limit)
|
||||
}
|
||||
|
||||
querybuilder.AddOffset(sb, query.Limit, query.Offset)
|
||||
|
||||
combinedArgs := allAggChArgs
|
||||
|
||||
mainSQL, mainArgs := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
|
||||
|
||||
@@ -841,6 +841,8 @@ func (b *traceOperatorCTEBuilder) buildTraceQuery(ctx context.Context, selectFro
|
||||
sb.Limit(b.operator.Limit)
|
||||
}
|
||||
|
||||
querybuilder.AddOffset(sb, b.operator.Limit, b.operator.Offset)
|
||||
|
||||
combinedArgs := allAggChArgs
|
||||
|
||||
sql, args := sb.BuildWithFlavor(sqlbuilder.ClickHouse, combinedArgs...)
|
||||
@@ -922,6 +924,8 @@ func (b *traceOperatorCTEBuilder) buildScalarQuery(ctx context.Context, selectFr
|
||||
sb.Limit(b.operator.Limit)
|
||||
}
|
||||
|
||||
querybuilder.AddOffset(sb, b.operator.Limit, b.operator.Offset)
|
||||
|
||||
combinedArgs := allAggChArgs
|
||||
|
||||
// Add HAVING clause if specified
|
||||
|
||||
@@ -34,13 +34,13 @@ type QueryBuilderQuery[T any] struct {
|
||||
|
||||
SelectFields []telemetrytypes.TelemetryFieldKey `json:"selectFields,omitzero" description:"The fields to return for each row in raw requests, in addition to the columns always present: timestamp and id for logs, timestamp, trace_id and span_id for traces. Ignored by aggregating requests."`
|
||||
|
||||
Limit int `json:"limit,omitempty" description:"The maximum number of rows for raw requests, or of series or rows for grouped aggregations, keeping the top entries by order. When 0, aggregations are unlimited while raw requests ordered by timestamp return a page of 10000 rows; follow nextCursor for the rest."`
|
||||
Limit int `json:"limit,omitempty" description:"The maximum number of rows for raw and trace requests, or of series or rows for grouped scalar and time_series requests, keeping the top entries by order. When 0, scalar and time_series requests are unlimited, raw requests ordered by timestamp return a page of 10000 rows to continue with nextCursor, and other raw and trace requests return 100 rows."`
|
||||
|
||||
LimitBy *LimitBy `json:"limitBy,omitempty" description:"Reserved for applying limit within each combination of the given group-by keys. Accepted but currently ignored."`
|
||||
|
||||
Offset int `json:"offset,omitempty" description:"The number of rows to skip before returning results. Prefer cursor for raw requests."`
|
||||
Offset int `json:"offset,omitempty" description:"The number of rows for raw and trace requests, or of ordered series or rows for scalar and time_series requests, to skip before the page starts. On raw requests the skip counts from where a cursor resumes, and the skipped rows are still read, so prefer cursor for paging."`
|
||||
|
||||
Cursor string `json:"cursor,omitempty" description:"The opaque nextCursor value from a previous raw result, used to fetch the next page."`
|
||||
Cursor string `json:"cursor,omitempty" description:"The opaque nextCursor value from a previous raw result, used to fetch the next page. Honored only by raw requests whose order starts with timestamp, for logs followed by id in the same direction; every other request ignores it."`
|
||||
|
||||
Having *Having `json:"having,omitempty" description:"A filter applied to aggregation results, referencing aliases or expressions, for example count() > 100."`
|
||||
|
||||
|
||||
@@ -26,6 +26,8 @@ type QueryBuilderFormula struct {
|
||||
|
||||
Limit int `json:"limit,omitempty" description:"The maximum number of series or rows to return. Unlimited when 0."`
|
||||
|
||||
Offset int `json:"offset,omitempty" description:"The number of ordered series or rows to skip before the page starts."`
|
||||
|
||||
Having *Having `json:"having,omitempty" description:"Accepted but currently ignored; the computed values are returned unfiltered."`
|
||||
|
||||
Functions []Function `json:"functions,omitzero" description:"Post-processing functions applied in order to each result series."`
|
||||
@@ -94,6 +96,14 @@ func (f QueryBuilderFormula) Validate() error {
|
||||
)
|
||||
}
|
||||
|
||||
if f.Offset < 0 {
|
||||
return errors.NewInvalidInputf(
|
||||
errors.CodeInvalidInput,
|
||||
"offset must be non-negative, got %d",
|
||||
f.Offset,
|
||||
)
|
||||
}
|
||||
|
||||
// Validate expression is parseable
|
||||
if _, err := govaluate.NewEvaluableExpressionWithFunctions(f.Expression, EvalFuncs()); err != nil {
|
||||
return errors.NewInvalidInputf(
|
||||
|
||||
@@ -180,6 +180,8 @@ func (q *QueryEnvelope) GetOffset() int {
|
||||
return spec.Offset
|
||||
case QueryBuilderQuery[MetricAggregation]:
|
||||
return spec.Offset
|
||||
case QueryBuilderFormula:
|
||||
return spec.Offset
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
@@ -209,6 +209,9 @@ func (q *QueryEnvelope) SetOffset(offset int) {
|
||||
case QueryBuilderQuery[MetricAggregation]:
|
||||
spec.Offset = offset
|
||||
q.Spec = spec
|
||||
case QueryBuilderFormula:
|
||||
spec.Offset = offset
|
||||
q.Spec = spec
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -13,9 +13,8 @@ const (
|
||||
DefaultOrderByKey = "__result"
|
||||
)
|
||||
|
||||
// ApplyLimit applies limit and ordering to a list of time series
|
||||
// This function sorts the series based on the provided order criteria and applies the limit.
|
||||
func ApplySeriesLimit(series []*TimeSeries, orderBy []OrderBy, limit int) []*TimeSeries {
|
||||
// ApplySeriesLimit sorts the series by the order criteria, skips the first offset of them and keeps at most limit.
|
||||
func ApplySeriesLimit(series []*TimeSeries, orderBy []OrderBy, limit, offset int) []*TimeSeries {
|
||||
if len(series) == 0 {
|
||||
return series
|
||||
}
|
||||
@@ -65,6 +64,10 @@ func ApplySeriesLimit(series []*TimeSeries, orderBy []OrderBy, limit int) []*Tim
|
||||
return 0
|
||||
})
|
||||
|
||||
if offset > 0 {
|
||||
series = series[min(offset, len(series)):]
|
||||
}
|
||||
|
||||
// Apply limit if specified
|
||||
if limit > 0 && len(series) > limit {
|
||||
return series[:limit]
|
||||
|
||||
@@ -84,7 +84,7 @@ func TestApplySeriesLimit(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
result := ApplySeriesLimit(series, orderBy, 2)
|
||||
result := ApplySeriesLimit(series, orderBy, 2, 0)
|
||||
|
||||
// Should return top 2 series by value: service-b (50.0), service-c (30.0)
|
||||
assert.Len(t, result, 2)
|
||||
@@ -209,7 +209,7 @@ func TestApplySeriesLimit(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
result := ApplySeriesLimit(series, orderBy, 3)
|
||||
result := ApplySeriesLimit(series, orderBy, 3, 0)
|
||||
|
||||
// Should return 3 series sorted by service (asc), then environment (desc)
|
||||
// Expected order:
|
||||
@@ -233,6 +233,25 @@ func TestApplySeriesLimit(t *testing.T) {
|
||||
assert.Equal(t, "prod", result[2].Labels[1].Value)
|
||||
assert.Equal(t, 40.0, result[2].Values[0].Value)
|
||||
})
|
||||
|
||||
t.Run("Offset skips the top series before the limit applies", func(t *testing.T) {
|
||||
hostKey := telemetrytypes.TelemetryFieldKey{Name: "host", FieldDataType: telemetrytypes.FieldDataTypeString}
|
||||
series := []*TimeSeries{
|
||||
{Labels: []*Label{{Key: hostKey, Value: "h1"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 10.0}}},
|
||||
{Labels: []*Label{{Key: hostKey, Value: "h2"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 50.0}}},
|
||||
{Labels: []*Label{{Key: hostKey, Value: "h3"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 30.0}}},
|
||||
{Labels: []*Label{{Key: hostKey, Value: "h4"}}, Values: []*TimeSeriesValue{{Timestamp: 1000, Value: 20.0}}},
|
||||
}
|
||||
|
||||
// Default order is value descending: h2, h3, h4, h1. Skipping one leaves h3, h4.
|
||||
result := ApplySeriesLimit(series, nil, 2, 1)
|
||||
|
||||
require.Len(t, result, 2)
|
||||
assert.Equal(t, "h3", result[0].Labels[0].Value)
|
||||
assert.Equal(t, "h4", result[1].Labels[0].Value)
|
||||
|
||||
assert.Empty(t, ApplySeriesLimit(series, nil, 2, 4))
|
||||
})
|
||||
}
|
||||
|
||||
func TestApplySeriesLimitRanksHeatmapSeriesByBucketTotals(t *testing.T) {
|
||||
@@ -271,7 +290,7 @@ func TestApplySeriesLimitRanksHeatmapSeriesByBucketTotals(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
result := ApplySeriesLimit(series, nil, 2)
|
||||
result := ApplySeriesLimit(series, nil, 2, 0)
|
||||
|
||||
require.Len(t, result, 2)
|
||||
assert.Equal(t, "busy", result[0].Labels[0].Value)
|
||||
|
||||
@@ -55,7 +55,7 @@ type QueryBuilderTraceOperator struct {
|
||||
Having *Having `json:"having,omitempty" description:"A condition on the aggregation results, for example count() > 10."`
|
||||
|
||||
Limit int `json:"limit,omitempty" description:"The maximum number of traces or groups to return. For raw requests 0 means 100 rows; for aggregations 0 means unlimited."`
|
||||
Offset int `json:"offset,omitempty" description:"The number of raw results to skip before returning. Aggregation requests ignore it."`
|
||||
Offset int `json:"offset,omitempty" description:"The number of raw or trace rows, or of ordered series or rows for time_series and scalar requests, to skip before the page starts."`
|
||||
Cursor string `json:"cursor,omitempty" description:"Accepted but currently ignored by the trace operator; page raw results with offset."`
|
||||
|
||||
Legend string `json:"legend" description:"The display name template for each series; {{field}} placeholders are replaced by group-by label values."`
|
||||
|
||||
19
tests/fixtures/querier.py
vendored
19
tests/fixtures/querier.py
vendored
@@ -300,6 +300,9 @@ def build_builder_query(
|
||||
functions: list[dict] | None = None,
|
||||
bucket_options: dict | None = None,
|
||||
disabled: bool = False,
|
||||
order: list[dict] | None = None,
|
||||
limit: int | None = None,
|
||||
offset: int | None = None,
|
||||
) -> dict:
|
||||
spec: dict[str, Any] = {
|
||||
"name": name,
|
||||
@@ -336,6 +339,15 @@ def build_builder_query(
|
||||
if functions:
|
||||
spec["functions"] = functions
|
||||
|
||||
if order:
|
||||
spec["order"] = order
|
||||
|
||||
if limit is not None:
|
||||
spec["limit"] = limit
|
||||
|
||||
if offset is not None:
|
||||
spec["offset"] = offset
|
||||
|
||||
return {"type": "builder_query", "spec": spec}
|
||||
|
||||
|
||||
@@ -348,6 +360,7 @@ def build_formula_query(
|
||||
disabled: bool = False,
|
||||
order: list[dict] | None = None,
|
||||
limit: int | None = None,
|
||||
offset: int | None = None,
|
||||
) -> dict:
|
||||
spec: dict[str, Any] = {
|
||||
"name": name,
|
||||
@@ -362,6 +375,8 @@ def build_formula_query(
|
||||
spec["order"] = order
|
||||
if limit is not None:
|
||||
spec["limit"] = limit
|
||||
if offset is not None:
|
||||
spec["offset"] = offset
|
||||
return {"type": "builder_formula", "spec": spec}
|
||||
|
||||
|
||||
@@ -668,6 +683,7 @@ def build_scalar_query(
|
||||
group_by: list[dict] | None = None,
|
||||
order: list[dict] | None = None,
|
||||
limit: int | None = None,
|
||||
offset: int | None = None,
|
||||
filter_expression: str | None = None,
|
||||
having_expression: str | None = None,
|
||||
step_interval: int = DEFAULT_STEP_INTERVAL,
|
||||
@@ -694,6 +710,9 @@ def build_scalar_query(
|
||||
if limit is not None:
|
||||
spec["limit"] = limit
|
||||
|
||||
if offset is not None:
|
||||
spec["offset"] = offset
|
||||
|
||||
if filter_expression:
|
||||
spec["filter"] = {"expression": filter_expression}
|
||||
|
||||
|
||||
81
tests/integration/tests/querierlogs/16_offset.py
Normal file
81
tests/integration/tests/querierlogs/16_offset.py
Normal file
@@ -0,0 +1,81 @@
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
|
||||
import pytest
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.logs import Logs
|
||||
from fixtures.querier import (
|
||||
Aggregation,
|
||||
BuilderQuery,
|
||||
RequestType,
|
||||
TelemetryFieldKey,
|
||||
get_all_series,
|
||||
make_query_request,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"limit, offset, expected_services",
|
||||
[
|
||||
# limit=2 selects the top two groups in SQL first; the skipped group must still count toward that page
|
||||
pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_limited_page"),
|
||||
pytest.param(None, 1, ["service-a", "service-b", "service-d"], id="offset_without_limit"),
|
||||
pytest.param(2, 4, [], id="offset_past_the_end"),
|
||||
],
|
||||
)
|
||||
def test_logs_time_series_offset_skips_series_in_order(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
limit: int | None,
|
||||
offset: int,
|
||||
expected_services: list[str],
|
||||
) -> None:
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
# Ordered by count, descending: service-c 7, service-a 5, service-b 3, service-d 1, all in the bucket before now.
|
||||
service_counts = {"service-a": 5, "service-b": 3, "service-c": 7, "service-d": 1}
|
||||
insert_logs(
|
||||
[
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=i + 1),
|
||||
resources={"service.name": service},
|
||||
body=f"{service} log {i}",
|
||||
)
|
||||
for service, count in service_counts.items()
|
||||
for i in range(count)
|
||||
]
|
||||
)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.TIME_SERIES,
|
||||
queries=[
|
||||
BuilderQuery(
|
||||
signal="logs",
|
||||
name="A",
|
||||
aggregations=[Aggregation("count()")],
|
||||
group_by=[TelemetryFieldKey("service.name", "string", "resource")],
|
||||
step_interval=60,
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
).to_dict()
|
||||
],
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
bucket_ms = int((now - timedelta(minutes=1)).timestamp() * 1000)
|
||||
assert [
|
||||
(
|
||||
next(label["value"] for label in series["labels"] if label["key"]["name"] == "service.name"),
|
||||
[(point["timestamp"], point["value"]) for point in series["values"]],
|
||||
)
|
||||
for series in get_all_series(response.json(), "A")
|
||||
] == [(service, [(bucket_ms, service_counts[service])]) for service in expected_services]
|
||||
75
tests/integration/tests/queriermetrics/19_offset.py
Normal file
75
tests/integration/tests/queriermetrics/19_offset.py
Normal file
@@ -0,0 +1,75 @@
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
|
||||
import pytest
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.metrics import Metrics
|
||||
from fixtures.querier import build_builder_query, get_all_series, make_query_request
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"limit, offset, expected_services",
|
||||
[
|
||||
pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_page"),
|
||||
pytest.param(None, 1, ["service-a", "service-b", "service-d"], id="offset_without_limit"),
|
||||
pytest.param(2, 4, [], id="offset_past_the_end"),
|
||||
],
|
||||
)
|
||||
def test_metrics_time_series_offset_skips_series_in_order(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_metrics: Callable[[list[Metrics]], None],
|
||||
limit: int | None,
|
||||
offset: int,
|
||||
expected_services: list[str],
|
||||
) -> None:
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
# Ordered by value, descending: service-c 70, service-a 50, service-b 30, service-d 10, one sample each in the bucket before now.
|
||||
service_values = {"service-a": 50.0, "service-b": 30.0, "service-c": 70.0, "service-d": 10.0}
|
||||
insert_metrics(
|
||||
[
|
||||
Metrics(
|
||||
metric_name="offset.gauge",
|
||||
labels={"service.name": service},
|
||||
timestamp=now - timedelta(seconds=1),
|
||||
temporality="Unspecified",
|
||||
type_="Gauge",
|
||||
is_monotonic=False,
|
||||
value=value,
|
||||
)
|
||||
for service, value in service_values.items()
|
||||
]
|
||||
)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
queries=[
|
||||
build_builder_query(
|
||||
"A",
|
||||
"offset.gauge",
|
||||
"avg",
|
||||
"sum",
|
||||
temporality="unspecified",
|
||||
group_by=["service.name"],
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert [
|
||||
(
|
||||
next(label["value"] for label in series["labels"] if label["key"]["name"] == "service.name"),
|
||||
[point["value"] for point in series["values"]],
|
||||
)
|
||||
for series in get_all_series(response.json(), "A")
|
||||
] == [(service, [service_values[service]]) for service in expected_services]
|
||||
249
tests/integration/tests/querierscalar/09_offset.py
Normal file
249
tests/integration/tests/querierscalar/09_offset.py
Normal file
@@ -0,0 +1,249 @@
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
|
||||
import pytest
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.logs import Logs
|
||||
from fixtures.metrics import Metrics
|
||||
from fixtures.querier import (
|
||||
RequestType,
|
||||
build_aggregation,
|
||||
build_formula_query,
|
||||
build_group_by_field,
|
||||
build_metrics_aggregation,
|
||||
build_order_by,
|
||||
build_scalar_query,
|
||||
find_named_result,
|
||||
get_scalar_table_data,
|
||||
make_query_request,
|
||||
make_scalar_query_request,
|
||||
)
|
||||
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
|
||||
|
||||
# Four services ordered by count, descending: service-c 7, service-a 5, service-b 3, service-d 1.
|
||||
service_counts = {"service-a": 5, "service-b": 3, "service-c": 7, "service-d": 1}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"limit, offset, expected_services",
|
||||
[
|
||||
pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_page"),
|
||||
pytest.param(None, 3, ["service-d"], id="offset_without_limit"),
|
||||
pytest.param(2, 4, [], id="offset_past_the_end"),
|
||||
],
|
||||
)
|
||||
def test_logs_scalar_offset_skips_rows_in_order(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
limit: int | None,
|
||||
offset: int,
|
||||
expected_services: list[str],
|
||||
) -> None:
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=i + 1),
|
||||
resources={"service.name": service},
|
||||
body=f"{service} log {i}",
|
||||
)
|
||||
for service, count in service_counts.items()
|
||||
for i in range(count)
|
||||
]
|
||||
)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_scalar_query_request(
|
||||
signoz,
|
||||
token,
|
||||
now,
|
||||
[
|
||||
build_scalar_query(
|
||||
name="A",
|
||||
signal="logs",
|
||||
aggregations=[build_aggregation("count()")],
|
||||
group_by=[build_group_by_field("service.name")],
|
||||
order=[build_order_by("count()", "desc")],
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert get_scalar_table_data(response.json()) == [[service, service_counts[service]] for service in expected_services]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"limit, offset, expected_services",
|
||||
[
|
||||
pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_page"),
|
||||
pytest.param(None, 3, ["service-d"], id="offset_without_limit"),
|
||||
pytest.param(2, 4, [], id="offset_past_the_end"),
|
||||
],
|
||||
)
|
||||
def test_traces_scalar_offset_skips_rows_in_order(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
limit: int | None,
|
||||
offset: int,
|
||||
expected_services: list[str],
|
||||
) -> None:
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
insert_traces(
|
||||
[
|
||||
Traces(
|
||||
timestamp=now - timedelta(seconds=i + 1),
|
||||
kind=TracesKind.SPAN_KIND_SERVER,
|
||||
status_code=TracesStatusCode.STATUS_CODE_OK,
|
||||
trace_id=TraceIdGenerator.trace_id(),
|
||||
span_id=TraceIdGenerator.span_id(),
|
||||
resources={"service.name": service},
|
||||
name=f"{service} span {i}",
|
||||
)
|
||||
for service, count in service_counts.items()
|
||||
for i in range(count)
|
||||
]
|
||||
)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_scalar_query_request(
|
||||
signoz,
|
||||
token,
|
||||
now,
|
||||
[
|
||||
build_scalar_query(
|
||||
name="A",
|
||||
signal="traces",
|
||||
aggregations=[build_aggregation("count()")],
|
||||
group_by=[build_group_by_field("service.name")],
|
||||
order=[build_order_by("count()", "desc")],
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert get_scalar_table_data(response.json()) == [[service, service_counts[service]] for service in expected_services]
|
||||
|
||||
|
||||
def test_metrics_scalar_offset_skips_rows_in_order(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_metrics: Callable[[list[Metrics]], None],
|
||||
) -> None:
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
# Ordered by value, descending: service-c 70, service-a 50, service-b 30, service-d 10.
|
||||
insert_metrics(
|
||||
[
|
||||
Metrics(
|
||||
metric_name="offset.gauge",
|
||||
labels={"service.name": service},
|
||||
timestamp=now - timedelta(seconds=1),
|
||||
temporality="Unspecified",
|
||||
type_="Gauge",
|
||||
is_monotonic=False,
|
||||
value=value,
|
||||
)
|
||||
for service, value in {"service-a": 50.0, "service-b": 30.0, "service-c": 70.0, "service-d": 10.0}.items()
|
||||
]
|
||||
)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_scalar_query_request(
|
||||
signoz,
|
||||
token,
|
||||
now,
|
||||
[
|
||||
build_scalar_query(
|
||||
name="A",
|
||||
signal="metrics",
|
||||
aggregations=[build_metrics_aggregation("offset.gauge", "latest", "sum", "unspecified", reduce_to="last")],
|
||||
group_by=[build_group_by_field("service.name", "string", "attribute")],
|
||||
order=[build_order_by("sum(offset.gauge)", "desc")],
|
||||
limit=2,
|
||||
offset=1,
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
assert get_scalar_table_data(response.json()) == [["service-a", 50.0], ["service-b", 30.0]]
|
||||
|
||||
|
||||
def test_formula_scalar_offset_skips_rows_in_order(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
insert_logs(
|
||||
[
|
||||
Logs(
|
||||
timestamp=now - timedelta(seconds=i + 1),
|
||||
resources={"service.name": service},
|
||||
body=f"{service} log {i}",
|
||||
)
|
||||
for service, count in service_counts.items()
|
||||
for i in range(count)
|
||||
]
|
||||
)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.SCALAR,
|
||||
queries=[
|
||||
build_scalar_query(
|
||||
name="A",
|
||||
signal="logs",
|
||||
aggregations=[build_aggregation("count()")],
|
||||
group_by=[build_group_by_field("service.name")],
|
||||
disabled=True,
|
||||
),
|
||||
build_formula_query(
|
||||
"F1",
|
||||
"A * 2",
|
||||
order=[build_order_by("__result", "desc")],
|
||||
limit=2,
|
||||
offset=1,
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
result = find_named_result(response.json()["data"]["data"]["results"], "F1")
|
||||
assert result is not None, "formula result F1 not found"
|
||||
service_column = next(i for i, column in enumerate(result["columns"]) if column["name"] == "service.name")
|
||||
value_column = next(i for i, column in enumerate(result["columns"]) if column["name"] == "__result")
|
||||
assert [[row[service_column], row[value_column]] for row in result["data"]] == [["service-a", 10.0], ["service-b", 6.0]]
|
||||
|
||||
|
||||
def test_negative_offset_is_rejected(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
) -> None:
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_scalar_query_request(
|
||||
signoz,
|
||||
token,
|
||||
datetime.now(tz=UTC).replace(second=0, microsecond=0),
|
||||
[build_scalar_query(name="A", signal="logs", aggregations=[build_aggregation("count()")], offset=-1)],
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.BAD_REQUEST, response.text
|
||||
assert response.json()["error"]["message"] == "invalid query 'A': offset must be non-negative, got -1"
|
||||
152
tests/integration/tests/queriertraces/15_offset.py
Normal file
152
tests/integration/tests/queriertraces/15_offset.py
Normal file
@@ -0,0 +1,152 @@
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
|
||||
import pytest
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.querier import (
|
||||
Aggregation,
|
||||
BuilderQuery,
|
||||
RequestType,
|
||||
TelemetryFieldKey,
|
||||
find_named_result,
|
||||
get_all_series,
|
||||
make_query_request,
|
||||
)
|
||||
from fixtures.traces import TraceIdGenerator, Traces, TracesKind, TracesStatusCode
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"limit, offset, expected_services",
|
||||
[
|
||||
# limit=2 selects the top two groups in SQL first; the skipped group must still count toward that page
|
||||
pytest.param(2, 1, ["service-a", "service-b"], id="offset_within_limited_page"),
|
||||
pytest.param(None, 1, ["service-a", "service-b", "service-d"], id="offset_without_limit"),
|
||||
pytest.param(2, 4, [], id="offset_past_the_end"),
|
||||
],
|
||||
)
|
||||
def test_traces_time_series_offset_skips_series_in_order(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
limit: int | None,
|
||||
offset: int,
|
||||
expected_services: list[str],
|
||||
) -> None:
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
# Ordered by count, descending: service-c 7, service-a 5, service-b 3, service-d 1, all in the bucket before now.
|
||||
service_counts = {"service-a": 5, "service-b": 3, "service-c": 7, "service-d": 1}
|
||||
insert_traces(
|
||||
[
|
||||
Traces(
|
||||
timestamp=now - timedelta(seconds=i + 1),
|
||||
kind=TracesKind.SPAN_KIND_SERVER,
|
||||
status_code=TracesStatusCode.STATUS_CODE_OK,
|
||||
trace_id=TraceIdGenerator.trace_id(),
|
||||
span_id=TraceIdGenerator.span_id(),
|
||||
resources={"service.name": service},
|
||||
name=f"{service} span {i}",
|
||||
)
|
||||
for service, count in service_counts.items()
|
||||
for i in range(count)
|
||||
]
|
||||
)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.TIME_SERIES,
|
||||
queries=[
|
||||
BuilderQuery(
|
||||
signal="traces",
|
||||
name="A",
|
||||
aggregations=[Aggregation("count()")],
|
||||
group_by=[TelemetryFieldKey("service.name", "string", "resource")],
|
||||
step_interval=60,
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
).to_dict()
|
||||
],
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
bucket_ms = int((now - timedelta(minutes=1)).timestamp() * 1000)
|
||||
assert [
|
||||
(
|
||||
next(label["value"] for label in series["labels"] if label["key"]["name"] == "service.name"),
|
||||
[(point["timestamp"], point["value"]) for point in series["values"]],
|
||||
)
|
||||
for series in get_all_series(response.json(), "A")
|
||||
] == [(service, [(bucket_ms, service_counts[service])]) for service in expected_services]
|
||||
|
||||
|
||||
def test_trace_operator_scalar_offset_skips_rows_in_order(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
|
||||
span_counts = {"offset-a": 4, "offset-b": 3, "offset-c": 2, "offset-d": 1}
|
||||
insert_traces(
|
||||
[
|
||||
Traces(
|
||||
timestamp=now - timedelta(seconds=10),
|
||||
duration=timedelta(seconds=1),
|
||||
trace_id=TraceIdGenerator.trace_id(),
|
||||
span_id=TraceIdGenerator.span_id(),
|
||||
parent_span_id="",
|
||||
name="scalar-offset",
|
||||
kind=TracesKind.SPAN_KIND_SERVER,
|
||||
status_code=TracesStatusCode.STATUS_CODE_OK,
|
||||
status_message="",
|
||||
resources={"service.name": service},
|
||||
attributes={"operation.type": "scalar-offset"},
|
||||
)
|
||||
for service, count in span_counts.items()
|
||||
for _ in range(count)
|
||||
]
|
||||
)
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((now - timedelta(minutes=5)).timestamp() * 1000),
|
||||
end_ms=int(now.timestamp() * 1000),
|
||||
request_type=RequestType.SCALAR,
|
||||
queries=[
|
||||
{
|
||||
"type": "builder_query",
|
||||
"spec": {
|
||||
"name": "A",
|
||||
"signal": "traces",
|
||||
"disabled": True,
|
||||
"filter": {"expression": "operation.type = 'scalar-offset'"},
|
||||
"aggregations": [{"expression": "count()"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "builder_trace_operator",
|
||||
"spec": {
|
||||
"name": "C",
|
||||
"expression": "A",
|
||||
"aggregations": [{"expression": "count()"}],
|
||||
"groupBy": [{"name": "service.name", "fieldContext": "resource"}],
|
||||
"limit": 2,
|
||||
"offset": 1,
|
||||
},
|
||||
},
|
||||
],
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
result = find_named_result(response.json()["data"]["data"]["results"], "C")
|
||||
assert result is not None, "trace_operator result C not found"
|
||||
assert result["data"] == [["offset-b", 3], ["offset-c", 2]]
|
||||
Reference in New Issue
Block a user