Send query plan to querier. (#11246)

**What this PR does / why we need it**:
Following #11123 and in order to
enable #10417 the query frontend
should send the serialized LogQL AST instead of the query string to the
queriers. This enables the frontend to change the AST and inject
expressions that are not expressible in LogQL.

**Checklist**
- [ ] Reviewed the
[`CONTRIBUTING.md`](https://github.com/grafana/loki/blob/main/CONTRIBUTING.md)
guide (**required**)
- [ ] Documentation added
- [x] Tests updated
- [ ] `CHANGELOG.md` updated
- [ ] If the change is worth mentioning in the release notes, add
`add-to-release-notes` label
- [ ] Changes that require user attention or interaction to upgrade are
documented in `docs/sources/setup/upgrade/_index.md`
- [ ] For Helm chart changes bump the Helm chart version in
`production/helm/loki/Chart.yaml` and update
`production/helm/loki/CHANGELOG.md` and
`production/helm/loki/README.md`. [Example
PR](d10549e)
- [ ] If the change is deprecating or removing a configuration option,
update the `deprecated-config.yaml` and `deleted-config.yaml` files
respectively in the `tools/deprecated-config-checker` directory.
[Example
PR](0d4416a)

---------

Signed-off-by: Callum Styan <[email protected]>
Co-authored-by: Callum Styan <[email protected]>

pkg/logcli/client/file.go

@@ -69,7 +69,7 @@ func (f *FileClient) Query(q string, limit int, t time.Time, direction logproto.
 
 	ctx = user.InjectOrgID(ctx, f.orgID)
 
-	params := logql.NewLiteralParams(
+	params, err := logql.NewLiteralParams(
 		q,
 		t, t,
 		0,
@@ -78,6 +78,9 @@ func (f *FileClient) Query(q string, limit int, t time.Time, direction logproto.
 		uint32(limit),
 		nil,
 	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse query: %w", err)
+	}
 
 	query := f.engine.Query(params)
 
@@ -106,7 +109,7 @@ func (f *FileClient) QueryRange(queryStr string, limit int, start, end time.Time
 
 	ctx = user.InjectOrgID(ctx, f.orgID)
 
-	params := logql.NewLiteralParams(
+	params, err := logql.NewLiteralParams(
 		queryStr,
 		start,
 		end,
@@ -116,6 +119,9 @@ func (f *FileClient) QueryRange(queryStr string, limit int, start, end time.Time
 		uint32(limit),
 		nil,
 	)
+	if err != nil {
+		return nil, err
+	}
 
 	query := f.engine.Query(params)
 

pkg/logcli/query/query.go

@@ -451,7 +451,7 @@ func (q *Query) DoLocalQuery(out output.LogOutput, statistics bool, orgID string
 	var query logql.Query
 
 	if q.isInstant() {
-		query = eng.Query(logql.NewLiteralParams(
+		params, err := logql.NewLiteralParams(
 			q.QueryString,
 			q.Start,
 			q.Start,
@@ -460,9 +460,14 @@ func (q *Query) DoLocalQuery(out output.LogOutput, statistics bool, orgID string
 			q.resultsDirection(),
 			uint32(q.Limit),
 			nil,
-		))
+		)
+		if err != nil {
+			return err
+		}
+
+		query = eng.Query(params)
 	} else {
-		query = eng.Query(logql.NewLiteralParams(
+		params, err := logql.NewLiteralParams(
 			q.QueryString,
 			q.Start,
 			q.End,
@@ -471,7 +476,16 @@ func (q *Query) DoLocalQuery(out output.LogOutput, statistics bool, orgID string
 			q.resultsDirection(),
 			uint32(q.Limit),
 			nil,
-		))
+		)
+		if err != nil {
+			return err
+		}
+
+		query = eng.Query(params)
+	}
+
+	if err != nil {
+		return err
 	}
 
 	// execute the query

pkg/logcli/query/query_test.go

@@ -425,7 +425,10 @@ func (t *testQueryClient) Query(_ string, _ int, _ time.Time, _ logproto.Directi
 func (t *testQueryClient) QueryRange(queryStr string, limit int, from, through time.Time, direction logproto.Direction, step, interval time.Duration, _ bool) (*loghttp.QueryResponse, error) {
 	ctx := user.InjectOrgID(context.Background(), "fake")
 
-	params := logql.NewLiteralParams(queryStr, from, through, step, interval, direction, uint32(limit), nil)
+	params, err := logql.NewLiteralParams(queryStr, from, through, step, interval, direction, uint32(limit), nil)
+	if err != nil {
+		return nil, err
+	}
 
 	v, err := t.engine.Query(params).Exec(ctx)
 	if err != nil {

pkg/logproto/indexgateway.proto

@@ -2,7 +2,6 @@ syntax = "proto3";
 
 package indexgatewaypb;
 
-import "gogoproto/gogo.proto";
 import "pkg/logproto/logproto.proto";
 
 option go_package = "github.com/grafana/loki/pkg/logproto";

pkg/logproto/sketch.proto

@@ -2,7 +2,6 @@ syntax = "proto3";
 
 package logproto;
 
-import "gogoproto/gogo.proto";
 import "pkg/logproto/logproto.proto";
 
 option go_package = "github.com/grafana/loki/pkg/logproto";

pkg/logql/blocker.go

@@ -33,8 +33,8 @@ func (qb *queryBlocker) isBlocked(ctx context.Context, tenant string) bool {
 		return false
 	}
 
-	query := qb.q.params.Query()
-	typ, err := QueryType(query)
+	query := qb.q.params.QueryString()
+	typ, err := QueryType(qb.q.params.GetExpression())
 	if err != nil {
 		typ = "unknown"
 	}

pkg/logql/blocker_test.go

@@ -145,15 +145,10 @@ func TestEngine_ExecWithBlockedQueries(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			limits.blockedQueries = test.blocked
 
-			q := eng.Query(LiteralParams{
-				qs:        test.q,
-				start:     time.Unix(0, 0),
-				end:       time.Unix(100000, 0),
-				step:      60 * time.Second,
-				direction: logproto.FORWARD,
-				limit:     1000,
-			})
-			_, err := q.Exec(user.InjectOrgID(context.Background(), "fake"))
+			params, err := NewLiteralParams(test.q, time.Unix(0, 0), time.Unix(100000, 0), 60*time.Second, 0, logproto.FORWARD, 1000, nil)
+			require.NoError(t, err)
+			q := eng.Query(params)
+			_, err = q.Exec(user.InjectOrgID(context.Background(), "fake"))
 
 			if test.expectedErr == nil {
 				require.NoError(t, err)