Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: support for window based pagination in new trace v4 #6440

Open
wants to merge 7 commits into
base: develop
Choose a base branch
from
+673 −53
Open

feat: support for window based pagination in new trace v4 #6440

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
5bf01a7
feat: support for window based pagination in new trace v4
nityanandagohain Nov 14, 2024
5077726
fix: update pagination logic
nityanandagohain Nov 14, 2024
373397c
fix: update comment
nityanandagohain Nov 14, 2024
7366913
fix: substract correct length
nityanandagohain Nov 14, 2024
b8a803f
fix: revert changes
nityanandagohain Nov 14, 2024
dee206c
fix: add tests for querier
nityanandagohain Nov 14, 2024
62c6c9b
fix: rename matcher
nityanandagohain Nov 14, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 61 additions & 23 deletions pkg/query-service/app/querier/querier.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,8 @@ type querier struct {
returnedSeries []*v3.Series
returnedErr error

UseLogsNewSchema bool
UseLogsNewSchema bool
UseTraceNewSchema bool
}

type QuerierOptions struct {
Expand Down Expand Up @@ -308,15 +309,17 @@ func (q *querier) runClickHouseQueries(ctx context.Context, params *v3.QueryRang
return results, errQueriesByName, err
}

func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangeParamsV3, tsRanges []utils.LogsListTsRange) ([]*v3.Result, map[string]error, error) {
func (q *querier) runWindowBasedListQuery(ctx context.Context, params *v3.QueryRangeParamsV3, tsRanges []utils.LogsListTsRange) ([]*v3.Result, map[string]error, error) {
nityanandagohain marked this conversation as resolved.
Show resolved Hide resolved
res := make([]*v3.Result, 0)
qName := ""
pageSize := uint64(0)
limit := uint64(0)

// se we are considering only one query
for name, v := range params.CompositeQuery.BuilderQueries {
qName = name
pageSize = v.PageSize
limit = v.Limit
}
data := []*v3.Row{}

Expand All @@ -343,21 +346,46 @@ func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangePar
data = append(data, rowList...)
}

// append a filter to the params
if len(data) > 0 {
params.CompositeQuery.BuilderQueries[qName].Filters.Items = append(params.CompositeQuery.BuilderQueries[qName].Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{
Key: "id",
IsColumn: true,
DataType: "string",
},
Operator: v3.FilterOperatorLessThan,
Value: data[len(data)-1].Data["id"],
})
}
// appending the filter to get the next set of data
if params.CompositeQuery.BuilderQueries[qName].DataSource == v3.DataSourceLogs {
if len(data) > 0 {
params.CompositeQuery.BuilderQueries[qName].Filters.Items = append(params.CompositeQuery.BuilderQueries[qName].Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{
Key: "id",
IsColumn: true,
DataType: "string",
},
Operator: v3.FilterOperatorLessThan,
Value: data[len(data)-1].Data["id"],
})
}

if uint64(len(data)) >= pageSize {
break
}
} else {
// we are updating the offset and limit based on the number of traces we have found in the current timerange
// eg -
// 1)offset = 0, limit = 100, tsRanges = [t1, t10], [t10, 20], [t20, t30]
//
// if 100 traces are there in [t1, t10] then 100 will return immediately.
// if 10 traces are there in [t1, t10] then we get 10, set offset to 0 and limit to 90, search in the next timerange of [t10, 20]
// if we don't find any trace in [t1, t10], then we search in [t10, 20] with offset=0, limit=100

//
// 2) offset = 50, limit = 100, tsRanges = [t1, t10], [t10, 20], [t20, t30]
//
// If we find 100 traces in [t1, t10] then we return immediately
// If we find 50 in [t1, t10] then it will set offset = 0 and limit = 50 and search in the next timerange of [t10, 20]
// if we don't find any trace in [t1, t10], then we search in [t10, 20] with offset=50, limit=100
if len(data) > 0 {
params.CompositeQuery.BuilderQueries[qName].Offset = 0
params.CompositeQuery.BuilderQueries[qName].Limit = limit - uint64(len(data))
}

if uint64(len(data)) >= pageSize {
break
if uint64(len(data)) >= limit {
break
}
}
}
res = append(res, &v3.Result{
Expand All @@ -368,15 +396,25 @@ func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangePar
}

func (q *querier) runBuilderListQueries(ctx context.Context, params *v3.QueryRangeParamsV3) ([]*v3.Result, map[string]error, error) {
// List query has support for only one query.
if q.UseLogsNewSchema && params.CompositeQuery != nil && len(params.CompositeQuery.BuilderQueries) == 1 {
// List query has support for only one query
// we are skipping for PanelTypeTrace as it has a custom order by regardless of what's in the payload
if params.CompositeQuery != nil &&
len(params.CompositeQuery.BuilderQueries) == 1 &&
params.CompositeQuery.PanelType != v3.PanelTypeTrace {
for _, v := range params.CompositeQuery.BuilderQueries {
if (v.DataSource == v3.DataSourceLogs && !q.UseLogsNewSchema) ||
(v.DataSource == v3.DataSourceTraces && !q.UseTraceNewSchema) {
break
}

// only allow of logs queries with timestamp ordering desc
if v.DataSource == v3.DataSourceLogs && len(v.OrderBy) == 1 && v.OrderBy[0].ColumnName == "timestamp" && v.OrderBy[0].Order == "desc" {
startEndArr := utils.GetLogsListTsRanges(params.Start, params.End)
if len(startEndArr) > 0 {
return q.runLogsListQuery(ctx, params, startEndArr)
}
// TODO(nitya): allow for timestamp asc
if (v.DataSource == v3.DataSourceLogs || v.DataSource == v3.DataSourceTraces) &&
len(v.OrderBy) == 1 &&
v.OrderBy[0].ColumnName == "timestamp" &&
v.OrderBy[0].Order == "desc" {
startEndArr := utils.GetListTsRanges(params.Start, params.End)
return q.runWindowBasedListQuery(ctx, params, startEndArr)
}
}
}
Expand Down
86 changes: 62 additions & 24 deletions pkg/query-service/app/querier/v2/querier.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,10 +48,11 @@ type querier struct {
testingMode bool
queriesExecuted []string
// tuple of start and end time in milliseconds
timeRanges [][]int
returnedSeries []*v3.Series
returnedErr error
UseLogsNewSchema bool
timeRanges [][]int
returnedSeries []*v3.Series
returnedErr error
UseLogsNewSchema bool
UseTraceNewSchema bool
}

type QuerierOptions struct {
Expand Down Expand Up @@ -308,15 +309,17 @@ func (q *querier) runClickHouseQueries(ctx context.Context, params *v3.QueryRang
return results, errQueriesByName, err
}

func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangeParamsV3, tsRanges []utils.LogsListTsRange) ([]*v3.Result, map[string]error, error) {
func (q *querier) runWindowBasedListQuery(ctx context.Context, params *v3.QueryRangeParamsV3, tsRanges []utils.LogsListTsRange) ([]*v3.Result, map[string]error, error) {
res := make([]*v3.Result, 0)
qName := ""
pageSize := uint64(0)
limit := uint64(0)

// se we are considering only one query
for name, v := range params.CompositeQuery.BuilderQueries {
qName = name
pageSize = v.PageSize
limit = v.Limit
}
data := []*v3.Row{}

Expand Down Expand Up @@ -344,20 +347,45 @@ func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangePar
}

// append a filter to the params
if len(data) > 0 {
params.CompositeQuery.BuilderQueries[qName].Filters.Items = append(params.CompositeQuery.BuilderQueries[qName].Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{
Key: "id",
IsColumn: true,
DataType: "string",
},
Operator: v3.FilterOperatorLessThan,
Value: data[len(data)-1].Data["id"],
})
}
if params.CompositeQuery.BuilderQueries[qName].DataSource == v3.DataSourceLogs {
if len(data) > 0 {
params.CompositeQuery.BuilderQueries[qName].Filters.Items = append(params.CompositeQuery.BuilderQueries[qName].Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{
Key: "id",
IsColumn: true,
DataType: "string",
},
Operator: v3.FilterOperatorLessThan,
Value: data[len(data)-1].Data["id"],
})
}

if uint64(len(data)) >= pageSize {
break
if uint64(len(data)) >= pageSize {
break
}
} else {
// we are updating the offset and limit based on the number of traces we have found in the current timerange
// eg -
// 1)offset = 0, limit = 100, tsRanges = [t1, t10], [t10, 20], [t20, t30]
//
// if 100 traces are there in [t1, t10] then 100 will return immediately.
// if 10 traces are there in [t1, t10] then we get 10, set offset to 0 and limit to 90, search in the next timerange of [t10, 20]
// if we don't find any trace in [t1, t10], then we search in [t10, 20] with offset=0, limit=100

//
// 2) offset = 50, limit = 100, tsRanges = [t1, t10], [t10, 20], [t20, t30]
//
// If we find 100 traces in [t1, t10] then we return immediately
// If we find 50 in [t1, t10] then it will set offset = 0 and limit = 50 and search in the next timerange of [t10, 20]
// if we don't find any trace in [t1, t10], then we search in [t10, 20] with offset=50, limit=100
if len(data) > 0 {
params.CompositeQuery.BuilderQueries[qName].Offset = 0
nityanandagohain marked this conversation as resolved.
Show resolved Hide resolved
params.CompositeQuery.BuilderQueries[qName].Limit = limit - uint64(len(data))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are over-subtracting by using the len of data? It should be subtracting the data fetched in the last recent request, not the whole data? Example, [t1-t2]:20, [t2-t3]:10, [t3-t4]:0, [t4:t5]:0, [t5:t6]:10, with offset = 0, limit = 40. after first iteration, the limit = 20, second iteration, the limit = 20 - (30) = -10.

Please add tests (use the clickhouse mocker to return rows).

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

disregard, I see limit is a global value that's not changed on each iteration. I would still encourage adding tests.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For testing I will have to mock for reader, since I have to mock the function reader.GetListResultV3 . Any reference on how do I add that ?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Have added testes, please have a look again @srikanthccv

}

if uint64(len(data)) >= limit {
break
}
}
}
res = append(res, &v3.Result{
Expand All @@ -369,14 +397,24 @@ func (q *querier) runLogsListQuery(ctx context.Context, params *v3.QueryRangePar

func (q *querier) runBuilderListQueries(ctx context.Context, params *v3.QueryRangeParamsV3) ([]*v3.Result, map[string]error, error) {
// List query has support for only one query.
if q.UseLogsNewSchema && params.CompositeQuery != nil && len(params.CompositeQuery.BuilderQueries) == 1 {
// we are skipping for PanelTypeTrace as it has a custom order by regardless of what's in the payload
if params.CompositeQuery != nil &&
len(params.CompositeQuery.BuilderQueries) == 1 &&
params.CompositeQuery.PanelType != v3.PanelTypeTrace {
for _, v := range params.CompositeQuery.BuilderQueries {
if (v.DataSource == v3.DataSourceLogs && !q.UseLogsNewSchema) ||
(v.DataSource == v3.DataSourceTraces && !q.UseTraceNewSchema) {
break
}

// only allow of logs queries with timestamp ordering desc
if v.DataSource == v3.DataSourceLogs && len(v.OrderBy) == 1 && v.OrderBy[0].ColumnName == "timestamp" && v.OrderBy[0].Order == "desc" {
startEndArr := utils.GetLogsListTsRanges(params.Start, params.End)
if len(startEndArr) > 0 {
return q.runLogsListQuery(ctx, params, startEndArr)
}
// TODO(nitya): allow for timestamp asc
if (v.DataSource == v3.DataSourceLogs || v.DataSource == v3.DataSourceTraces) &&
len(v.OrderBy) == 1 &&
v.OrderBy[0].ColumnName == "timestamp" &&
v.OrderBy[0].Order == "desc" {
startEndArr := utils.GetListTsRanges(params.Start, params.End)
return q.runWindowBasedListQuery(ctx, params, startEndArr)
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion pkg/query-service/utils/logs.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ type LogsListTsRange struct {
End int64
}

func GetLogsListTsRanges(start, end int64) []LogsListTsRange {
func GetListTsRanges(start, end int64) []LogsListTsRange {
startNano := GetEpochNanoSecs(start)
endNano := GetEpochNanoSecs(end)
result := []LogsListTsRange{}
Expand Down
4 changes: 2 additions & 2 deletions pkg/query-service/utils/logs_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ import (
v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
)

func TestLogsListTsRange(t *testing.T) {
func TestListTsRange(t *testing.T) {
startEndData := []struct {
name string
start int64
Expand Down Expand Up @@ -44,7 +44,7 @@ func TestLogsListTsRange(t *testing.T) {
}

for _, test := range startEndData {
res := GetLogsListTsRanges(test.start, test.end)
res := GetListTsRanges(test.start, test.end)
for i, v := range res {
if test.res[i].Start != v.Start || test.res[i].End != v.End {
t.Errorf("expected range was %v - %v, got %v - %v", v.Start, v.End, test.res[i].Start, test.res[i].End)
Expand Down
Loading