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/querier.go

@@ -1020,7 +1020,7 @@ func (q *SingleTenantQuerier) DetectedFields(ctx context.Context, req *logproto.
 
 	// TODO(twhitney): converting from a step to a duration should be abstracted and reused,
 	// doing this in a few places now.
-	streams, err := streamsForFieldDetection(iters, req.LineLimit, time.Duration(req.Step*1e6))
+	streams, err := streamsForFieldDetection(iters, req.LineLimit, time.Duration(req.Step))
 	if err != nil {
 		return nil, err
 	}
@@ -1100,20 +1100,22 @@ func parseDetectedFields(ctx context.Context, limit uint32, streams logqlmodel.S
 	fieldCount := uint32(0)
 
 	for _, stream := range streams {
-
 		level.Debug(spanlogger.FromContext(ctx)).Log(
 			"detected_fields", "true",
 			"msg", fmt.Sprintf("looking for detected fields in stream %d with %d lines", stream.Hash, len(stream.Entries)))
 
 		for _, entry := range stream.Entries {
 			detected := parseLine(entry.Line)
 			for k, vals := range detected {
-				if fieldCount >= limit {
-					return detectedFields
+				df, ok := detectedFields[k]
+				if !ok && fieldCount < limit {
+					df = newParsedFields()
+					detectedFields[k] = df
+					fieldCount++
 				}
 
-				if _, ok := detectedFields[k]; !ok {
-					detectedFields[k] = newParsedFields()
+				if df == nil {
+					continue
 				}
 
 				for _, v := range vals {
@@ -1128,8 +1130,6 @@ func parseDetectedFields(ctx context.Context, limit uint32, streams logqlmodel.S
 				level.Debug(spanlogger.FromContext(ctx)).Log(
 					"detected_fields", "true",
 					"msg", fmt.Sprintf("detected field %s with %d values", k, len(vals)))
-
-				fieldCount++
 			}
 		}
 	}

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"
@@ -972,9 +973,12 @@ func (c Codec) EncodeRequest(ctx context.Context, r queryrangebase.Request) (*ht
 		return req.WithContext(ctx), nil
 	case *DetectedFieldsRequest:
 		params := url.Values{
-			"start": []string{fmt.Sprintf("%d", request.Start.UnixNano())},
-			"end":   []string{fmt.Sprintf("%d", request.End.UnixNano())},
-			"query": []string{request.GetQuery()},
+			"query":       []string{request.GetQuery()},
+			"start":       []string{fmt.Sprintf("%d", request.Start.UnixNano())},
+			"end":         []string{fmt.Sprintf("%d", request.End.UnixNano())},
+			"line_limit":  []string{fmt.Sprintf("%d", request.GetLineLimit())},
+			"field_limit": []string{fmt.Sprintf("%d", request.GetFieldLimit())},
+			"step":        []string{fmt.Sprintf("%d", request.GetStep())},
 		}
 
 		u := &url.URL{
@@ -1587,6 +1591,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])
 	}
@@ -1781,8 +1801,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)
 	}
 }
 
@@ -1950,6 +1974,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 {
@@ -2071,12 +2136,15 @@ type DetectedFieldsRequest struct {
 	path string
 }
 
-func NewDetectedFieldsRequest(start, end time.Time, query, path string) *DetectedFieldsRequest {
+func NewDetectedFieldsRequest(start, end time.Time, lineLimit, fieldLimit uint32, step int64, query, path string) *DetectedFieldsRequest {
 	return &DetectedFieldsRequest{
 		DetectedFieldsRequest: logproto.DetectedFieldsRequest{
-			Start: start,
-			End:   end,
-			Query: query,
+			Start:      start,
+			End:        end,
+			Query:      query,
+			LineLimit:  lineLimit,
+			FieldLimit: fieldLimit,
+			Step:       step,
 		},
 		path: path,
 	}
@@ -2103,7 +2171,15 @@ func (r *DetectedFieldsRequest) GetStartTs() time.Time {
 }
 
 func (r *DetectedFieldsRequest) GetStep() int64 {
-	return 0
+	return r.Step
+}
+
+func (r *DetectedFieldsRequest) GetLineLimit() uint32 {
+	return r.LineLimit
+}
+
+func (r *DetectedFieldsRequest) GetFieldLimit() uint32 {
+	return r.FieldLimit
 }
 
 func (r *DetectedFieldsRequest) Path() string {