chore(engine): Fix logical plan for metric queries (#18471)

The logical plan for metric queries previously omitted the SORT instruction, because if would have translated into a SortMerge with a topK.

However, due to the recent change, the sort merge does not do the topK any more, but is still used to merge multiple DataObjScan inputs in DESC timestamp order.
This is required for the SELECT instruction (Filter operation), since it only accept a single input, which is ensured by the SORT.

Fixes the error `filter expects exactly one input, got N`.

Signed-off-by: Christian Haudum <christian.haudum@gmail.com>

pkg/engine/planner/logical/planner.go

@@ -99,17 +99,16 @@ func buildPlanForLogQuery(expr syntax.LogSelectorExpr, params logql.Params, isMe
 		},
 	)
 
-	// Metric queries currently do not expect the logs to be sorted by timestamp.
-	if !isMetricQuery {
-		// SORT -> SortMerge
-		direction := params.Direction()
-		if direction == logproto.FORWARD {
-			return nil, fmt.Errorf("forward search log queries are not supported: %w", errUnimplemented)
-		}
-		ascending := direction == logproto.FORWARD
-		builder = builder.Sort(*timestampColumnRef(), ascending, false)
+	direction := params.Direction()
+	if !isMetricQuery && direction == logproto.FORWARD {
+		return nil, fmt.Errorf("forward search log queries are not supported: %w", errUnimplemented)
 	}
 
+	// SORT -> SortMerge
+	// We always sort DESC. ASC timestamp sorting is not supported for logs queries,
+	// and metric queries do not care about the direction.
+	builder = builder.Sort(*timestampColumnRef(), false, false)
+
 	// SELECT -> Filter
 	start := params.Start()
 	end := params.End()

pkg/engine/planner/logical/planner_test.go

@@ -137,15 +137,16 @@ func TestConvertAST_MetricQuery_Success(t *testing.T) {
 %2 = MATCH_RE label.namespace "loki-.*"
 %3 = AND %1 %2
 %4 = MAKETABLE [selector=%3, shard=0_of_1]
-%5 = GTE builtin.timestamp 1970-01-01T00:55:00Z
-%6 = SELECT %4 [predicate=%5]
-%7 = LT builtin.timestamp 1970-01-01T02:00:00Z
-%8 = SELECT %6 [predicate=%7]
-%9 = MATCH_STR builtin.message "metric.go"
-%10 = SELECT %8 [predicate=%9]
-%11 = RANGE_AGGREGATION %10 [operation=count, start_ts=1970-01-01T01:00:00Z, end_ts=1970-01-01T02:00:00Z, step=0s, range=5m0s]
-%12 = VECTOR_AGGREGATION %11 [operation=sum, group_by=(ambiguous.level)]
-RETURN %12
+%5 = SORT %4 [column=builtin.timestamp, asc=false, nulls_first=false]
+%6 = GTE builtin.timestamp 1970-01-01T00:55:00Z
+%7 = SELECT %5 [predicate=%6]
+%8 = LT builtin.timestamp 1970-01-01T02:00:00Z
+%9 = SELECT %7 [predicate=%8]
+%10 = MATCH_STR builtin.message "metric.go"
+%11 = SELECT %9 [predicate=%10]
+%12 = RANGE_AGGREGATION %11 [operation=count, start_ts=1970-01-01T01:00:00Z, end_ts=1970-01-01T02:00:00Z, step=0s, range=5m0s]
+%13 = VECTOR_AGGREGATION %12 [operation=sum, group_by=(ambiguous.level)]
+RETURN %13
 `
 
 	require.Equal(t, expected, logicalPlan.String())