(q string)
| 286 | } |
| 287 | |
| 288 | func sortPlanForQuery(q string) (sortPlan, error) { |
| 289 | expr, err := cortexparser.ParseExpr(q) |
| 290 | if err != nil { |
| 291 | return 0, err |
| 292 | } |
| 293 | // Check if the root expression is topk, bottomk, limitk, or limit_ratio |
| 294 | if aggr, ok := expr.(*promqlparser.AggregateExpr); ok { |
| 295 | if aggr.Op == promqlparser.TOPK || aggr.Op == promqlparser.BOTTOMK || aggr.Op == promqlparser.LIMITK || aggr.Op == promqlparser.LIMIT_RATIO { |
| 296 | return mergeOnly, nil |
| 297 | } |
| 298 | } |
| 299 | checkForSort := func(expr promqlparser.Expr) (sortAsc, sortDesc bool) { |
| 300 | if n, ok := expr.(*promqlparser.Call); ok { |
| 301 | if n.Func != nil { |
| 302 | if n.Func.Name == "sort" { |
| 303 | sortAsc = true |
| 304 | } |
| 305 | if n.Func.Name == "sort_desc" { |
| 306 | sortDesc = true |
| 307 | } |
| 308 | if n.Func.Name == "sort_by_label" { |
| 309 | sortAsc = true |
| 310 | } |
| 311 | if n.Func.Name == "sort_by_label_desc" { |
| 312 | sortDesc = true |
| 313 | } |
| 314 | } |
| 315 | } |
| 316 | return sortAsc, sortDesc |
| 317 | } |
| 318 | // Check the root expression for sort |
| 319 | if sortAsc, sortDesc := checkForSort(expr); sortAsc || sortDesc { |
| 320 | if sortAsc { |
| 321 | return sortByValuesAsc, nil |
| 322 | } |
| 323 | return sortByValuesDesc, nil |
| 324 | } |
| 325 | |
| 326 | // If the root expression is a binary expression, check the LHS and RHS for sort |
| 327 | if bin, ok := expr.(*promqlparser.BinaryExpr); ok { |
| 328 | if sortAsc, sortDesc := checkForSort(bin.LHS); sortAsc || sortDesc { |
| 329 | if sortAsc { |
| 330 | return sortByValuesAsc, nil |
| 331 | } |
| 332 | return sortByValuesDesc, nil |
| 333 | } |
| 334 | if sortAsc, sortDesc := checkForSort(bin.RHS); sortAsc || sortDesc { |
| 335 | if sortAsc { |
| 336 | return sortByValuesAsc, nil |
| 337 | } |
| 338 | return sortByValuesDesc, nil |
| 339 | } |
| 340 | } |
| 341 | return sortByLabels, nil |
| 342 | } |
| 343 | |
| 344 | // mergeSampleStreams deduplicates sample streams using a map. |
| 345 | func mergeSampleStreams(output map[string]SampleStream, sampleStreams []SampleStream) { |
no outgoing calls