feat: split detected fields queries

pkg/logproto/logproto.proto

@@ -452,6 +452,7 @@ message DetectedFieldsRequest {
 
 message DetectedFieldsResponse {
   repeated DetectedField fields = 1;
+  uint32 limit = 2;
 }
 
 message DetectedField {

pkg/querier/queryrange/codec.go

@@ -17,6 +17,7 @@ import (
 	"golang.org/x/exp/maps"
 
 	"github.com/grafana/loki/v3/pkg/storage/chunk/cache/resultscache"
+	"github.com/grafana/loki/v3/pkg/storage/detected"
 	"github.com/grafana/loki/v3/pkg/storage/stores/index/seriesvolume"
 
 	"github.com/grafana/dskit/httpgrpc"
@@ -1541,6 +1542,22 @@ func (Codec) MergeResponse(responses ...queryrangebase.Response) (queryrangebase
 			Response: seriesvolume.Merge(resps, resp0.Response.Limit),
 			Headers:  headers,
 		}, nil
+	case *DetectedFieldsResponse:
+		resp0 := responses[0].(*DetectedFieldsResponse)
+		headers := resp0.Headers
+		limit := resp0.Response.Limit
+
+		fields := []*logproto.DetectedField{}
+		for _, r := range responses {
+			fields = append(fields, r.(*DetectedFieldsResponse).Response.Fields...)
+		}
+
+		return &DetectedFieldsResponse{
+			Response: &logproto.DetectedFieldsResponse{
+				Fields: detected.MergeFields(fields, limit),
+			},
+			Headers: headers,
+		}, nil
 	default:
 		return nil, fmt.Errorf("unknown response type (%T) in merging responses", responses[0])
 	}
@@ -1735,8 +1752,12 @@ func ParamsFromRequest(req queryrangebase.Request) (logql.Params, error) {
 		return &paramsStatsWrapper{
 			IndexStatsRequest: r,
 		}, nil
+	case *DetectedFieldsRequest:
+		return &paramsDetectedFieldsWrapper{
+			DetectedFieldsRequest: r,
+		}, nil
 	default:
-		return nil, fmt.Errorf("expected one of the *LokiRequest, *LokiInstantRequest, *LokiSeriesRequest, *LokiLabelNamesRequest, got (%T)", r)
+		return nil, fmt.Errorf("expected one of the *LokiRequest, *LokiInstantRequest, *LokiSeriesRequest, *LokiLabelNamesRequest, *DetectedFieldsRequest, got (%T)", r)
 	}
 }
 
@@ -1904,6 +1925,47 @@ func (p paramsStatsWrapper) Shards() []string {
 	return make([]string, 0)
 }
 
+type paramsDetectedFieldsWrapper struct {
+	*DetectedFieldsRequest
+}
+
+func (p paramsDetectedFieldsWrapper) QueryString() string {
+	return p.GetQuery()
+}
+
+func (p paramsDetectedFieldsWrapper) GetExpression() syntax.Expr {
+	expr, err := syntax.ParseExpr(p.GetQuery())
+	if err != nil {
+		return nil
+	}
+
+	return expr
+}
+
+func (p paramsDetectedFieldsWrapper) Start() time.Time {
+	return p.GetStartTs()
+}
+
+func (p paramsDetectedFieldsWrapper) End() time.Time {
+	return p.GetEndTs()
+}
+
+func (p paramsDetectedFieldsWrapper) Step() time.Duration {
+	return time.Duration(p.GetStep() * 1e6)
+}
+
+func (p paramsDetectedFieldsWrapper) Interval() time.Duration {
+	return 0
+}
+
+func (p paramsDetectedFieldsWrapper) Direction() logproto.Direction {
+	return logproto.BACKWARD
+}
+func (p paramsDetectedFieldsWrapper) Limit() uint32 { return p.DetectedFieldsRequest.LineLimit }
+func (p paramsDetectedFieldsWrapper) Shards() []string {
+	return make([]string, 0)
+}
+
 func httpResponseHeadersToPromResponseHeaders(httpHeaders http.Header) []queryrangebase.PrometheusResponseHeader {
 	var promHeaders []queryrangebase.PrometheusResponseHeader
 	for h, hv := range httpHeaders {

pkg/querier/queryrange/codec_test.go

@@ -1013,7 +1013,7 @@ func Test_codec_EncodeResponse(t *testing.T) {
 			},
 		},
 		{
-			"loki legacy", "/api/promt/query",
+			"loki legacy", "/api/prompt/query",
 			&LokiResponse{
 				Status:    loghttp.QueryStatusSuccess,
 				Direction: logproto.FORWARD,
@@ -1622,6 +1622,104 @@ func Test_codec_MergeResponse(t *testing.T) {
 	}
 }
 
+func Test_codec_MergeResponse_DetectedFieldsResponse(t *testing.T) {
+	t.Run("merges the responses, taking the highest cardinality", func(t *testing.T) {
+		responses := []queryrangebase.Response{
+			&DetectedFieldsResponse{
+				Response: &logproto.DetectedFieldsResponse{
+					Fields: []*logproto.DetectedField{{
+						Label:       "foo",
+						Type:        logproto.DetectedFieldString,
+						Cardinality: 1,
+					}},
+					Limit: 2,
+				},
+			},
+			&DetectedFieldsResponse{
+				Response: &logproto.DetectedFieldsResponse{
+					Fields: []*logproto.DetectedField{{
+						Label:       "foo",
+						Type:        logproto.DetectedFieldString,
+						Cardinality: 3,
+					}},
+					Limit: 2,
+				},
+			},
+		}
+
+		got, err := DefaultCodec.MergeResponse(responses...)
+		require.Nil(t, err)
+		response := got.(*DetectedFieldsResponse).Response
+		require.Equal(t, 1, len(response.Fields))
+
+		foo := response.Fields[0]
+		require.Equal(t, foo.Label, "foo")
+		require.Equal(t, foo.Type, logproto.DetectedFieldString)
+		require.Equal(t, foo.Cardinality, uint64(3))
+	})
+
+	t.Run("merges the responses, enforcing the limit", func(t *testing.T) {
+		responses := []queryrangebase.Response{
+			&DetectedFieldsResponse{
+				Response: &logproto.DetectedFieldsResponse{
+					Fields: []*logproto.DetectedField{
+						{
+							Label:       "foo",
+							Type:        logproto.DetectedFieldString,
+							Cardinality: 1,
+						},
+						{
+							Label:       "bar",
+							Type:        logproto.DetectedFieldInt,
+							Cardinality: 42,
+						},
+					},
+					Limit: 2,
+				},
+			},
+			&DetectedFieldsResponse{
+				Response: &logproto.DetectedFieldsResponse{
+					Fields: []*logproto.DetectedField{
+						{
+							Label:       "foo",
+							Type:        logproto.DetectedFieldString,
+							Cardinality: 27,
+						}, {
+							Label:       "baz",
+							Type:        logproto.DetectedFieldBoolean,
+							Cardinality: 3,
+						},
+					},
+					Limit: 2,
+				},
+			},
+		}
+
+		got, err := DefaultCodec.MergeResponse(responses...)
+		require.Nil(t, err)
+		response := got.(*DetectedFieldsResponse).Response
+		require.Equal(t, 2, len(response.Fields))
+
+		var foo *logproto.DetectedField
+		var baz *logproto.DetectedField
+		for _, f := range response.Fields {
+			if f.Label == "foo" {
+				foo = f
+			}
+			if f.Label == "baz" {
+				baz = f
+			}
+		}
+
+		require.Equal(t, foo.Label, "foo")
+		require.Equal(t, foo.Type, logproto.DetectedFieldString)
+		require.Equal(t, 27, int(foo.Cardinality))
+
+		require.Nil(t, baz)
+	})
+
+}
+
 type badResponse struct{}
 
 func (badResponse) Reset()                                                 {}