SQL Expressions: Add endpoint to get Schemas (#108864)

Return the SQL schema for all DS queries in request (to provide information to AI / Autocomplete for SQL expressions).

All DS queries are treated as if they were inputs to SQL expressions in terms of conversion, regardless if they are selected in a query or not.

Requires feature toggle queryService = true

Endpoint is apis/query.grafana.app/v0alpha1/namespaces/default/sqlschemas

---------

Co-authored-by: Todd Treece <360020+toddtreece@users.noreply.github.com>

pkg/apis/query/v0alpha1/query.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	data "github.com/grafana/grafana-plugin-sdk-go/experimental/apis/data/v0alpha1"
+	"github.com/grafana/grafana/pkg/expr"
 )
 
 // Generic query request with shared time across all values
@@ -28,6 +29,14 @@ type QueryDataResponse struct {
 	backend.QueryDataResponse `json:",inline"`
 }
 
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+type SQLSchemas struct {
+	metav1.TypeMeta `json:",inline"`
+
+	// Backend wrapper (external dependency)
+	expr.SQLSchemas `json:"sqlSchemas,inline"`
+}
+
 // GetResponseCode return the right status code for the response by checking the responses.
 func GetResponseCode(rsp *backend.QueryDataResponse) int {
 	if rsp == nil {

pkg/apis/query/v0alpha1/zz_generated.deepcopy.go

@@ -262,3 +262,29 @@ func (in *QueryTypeDefinitionList) DeepCopyObject() runtime.Object {
 	}
 	return nil
 }
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *SQLSchemas) DeepCopyInto(out *SQLSchemas) {
+	*out = *in
+	out.TypeMeta = in.TypeMeta
+	out.SQLSchemas = in.SQLSchemas.DeepCopy()
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SQLSchemas.
+func (in *SQLSchemas) DeepCopy() *SQLSchemas {
+	if in == nil {
+		return nil
+	}
+	out := new(SQLSchemas)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object.
+func (in *SQLSchemas) DeepCopyObject() runtime.Object {
+	if c := in.DeepCopy(); c != nil {
+		return c
+	}
+	return nil
+}

pkg/apis/query/v0alpha1/zz_generated.openapi.go

@@ -23,6 +23,7 @@ func GetOpenAPIDefinitions(ref common.ReferenceCallback) map[string]common.OpenA
 		"github.com/grafana/grafana/pkg/apis/query/v0alpha1.QueryDataResponse":        schema_pkg_apis_query_v0alpha1_QueryDataResponse(ref),
 		"github.com/grafana/grafana/pkg/apis/query/v0alpha1.QueryTypeDefinition":      schema_pkg_apis_query_v0alpha1_QueryTypeDefinition(ref),
 		"github.com/grafana/grafana/pkg/apis/query/v0alpha1.QueryTypeDefinitionList":  schema_pkg_apis_query_v0alpha1_QueryTypeDefinitionList(ref),
+		"github.com/grafana/grafana/pkg/apis/query/v0alpha1.SQLSchemas":               schema_pkg_apis_query_v0alpha1_SQLSchemas(ref),
 	}
 }
 
@@ -482,3 +483,29 @@ func schema_pkg_apis_query_v0alpha1_QueryTypeDefinitionList(ref common.Reference
 			"github.com/grafana/grafana/pkg/apis/query/v0alpha1.QueryTypeDefinition", "k8s.io/apimachinery/pkg/apis/meta/v1.ListMeta"},
 	}
 }
+
+func schema_pkg_apis_query_v0alpha1_SQLSchemas(ref common.ReferenceCallback) common.OpenAPIDefinition {
+	return common.OpenAPIDefinition{
+		Schema: spec.Schema{
+			SchemaProps: spec.SchemaProps{
+				Type: []string{"object"},
+				Properties: map[string]spec.Schema{
+					"kind": {
+						SchemaProps: spec.SchemaProps{
+							Description: "Kind is a string value representing the REST resource this object represents. Servers may infer this from the endpoint the client submits requests to. Cannot be updated. In CamelCase. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds",
+							Type:        []string{"string"},
+							Format:      "",
+						},
+					},
+					"apiVersion": {
+						SchemaProps: spec.SchemaProps{
+							Description: "APIVersion defines the versioned schema of this representation of an object. Servers should convert recognized schemas to the latest internal value, and may reject unrecognized values. More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources",
+							Type:        []string{"string"},
+							Format:      "",
+						},
+					},
+				},
+			},
+		},
+	}
+}

pkg/expr/sql/frame_table.go

@@ -30,7 +30,7 @@ func (ft *FrameTable) String() string {
 	return ft.Name()
 }
 
-func schemaFromFrame(frame *data.Frame) mysql.Schema {
+func SchemaFromFrame(frame *data.Frame) mysql.Schema {
 	schema := make(mysql.Schema, len(frame.Fields))
 
 	for i, field := range frame.Fields {
@@ -48,7 +48,7 @@ func schemaFromFrame(frame *data.Frame) mysql.Schema {
 // Schema implements the mysql.Table interface
 func (ft *FrameTable) Schema() mysql.Schema {
 	if ft.schema == nil {
-		ft.schema = schemaFromFrame(ft.Frame)
+		ft.schema = SchemaFromFrame(ft.Frame)
 	}
 	return ft.schema
 }

pkg/expr/sql_schema.go

@@ -0,0 +1,231 @@
+package expr
+
+import (
+	"context"
+	"reflect"
+	"time"
+
+	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"github.com/grafana/grafana/pkg/expr/mathexp"
+	"github.com/grafana/grafana/pkg/expr/sql"
+)
+
+// BasicColumn represents the column type for data that is input to a SQL expression.
+type BasicColumn struct {
+	Name               string         `json:"name"`
+	MySQLType          string         `json:"mysqlType"`
+	Nullable           bool           `json:"nullable"`
+	DataFrameFieldType data.FieldType `json:"dataFrameFieldType"`
+}
+
+// SchemaInfo provides information and some sample data for data that could be an input
+// to a SQL expression.
+type SchemaInfo struct {
+	Columns    []BasicColumn `json:"columns"`
+	SampleRows [][]any       `json:"sampleRows"`
+	Error      string        `json:"error,omitempty"`
+}
+
+// SQLSchemas returns info about what the Schema for a DS query will be like if the
+// query were to be used an input to SQL expressions. So effectively post SQL expressions input
+// conversion.
+// There is a a manual DeepCopy at the end of this file that will need to be updated when this our the
+// underlying structs are change. The hack script will also need to be run to update the Query service API
+// generated types.
+type SQLSchemas map[string]SchemaInfo
+
+// GetSQLSchemas returns what the schemas are for SQL expressions for all DS queries
+// in the request. It executes the queries to get the schemas.
+// Intended use is for autocomplete and AI, so used during the authoring/editing experience only.
+func (s *Service) GetSQLSchemas(ctx context.Context, req Request) (SQLSchemas, error) {
+	// Extract DS Nodes and Execute Them
+	// Building the pipeline is maybe not best, as it can have more errors.
+	filtered := make([]Query, 0, len(req.Queries))
+	for _, q := range req.Queries {
+		if NodeTypeFromDatasourceUID(q.DataSource.UID) == TypeDatasourceNode {
+			filtered = append(filtered, q)
+		}
+	}
+	req.Queries = filtered
+	pipeline, err := s.buildPipeline(ctx, &req)
+	if err != nil {
+		return nil, err
+	}
+
+	var schemas = make(SQLSchemas)
+
+	for _, node := range pipeline {
+		// For now, execute calls convert at the end, so we are being lazy and running the full conversion. Longer run we want to run without
+		// full conversion and just get the schema. Maybe conversion should be
+		dsNode := node.(*DSNode)
+		// Make all input to SQL
+		dsNode.isInputToSQLExpr = true
+
+		// TODO: check where time is coming from, don't recall
+		res, err := dsNode.Execute(ctx, time.Now(), mathexp.Vars{}, s)
+		if err != nil {
+			schemas[dsNode.RefID()] = SchemaInfo{Error: err.Error()}
+			continue
+			// we want to continue and get the schemas we can
+		}
+		if res.Error != nil {
+			schemas[dsNode.RefID()] = SchemaInfo{Error: res.Error.Error()}
+			continue
+			// we want to continue and get the schemas we can
+		}
+
+		frames := res.Values.AsDataFrames(dsNode.RefID())
+		if len(frames) == 0 {
+			schemas[dsNode.RefID()] = SchemaInfo{Error: "no data"}
+		}
+		frame := frames[0]
+
+		schema := sql.SchemaFromFrame(frame)
+		columns := make([]BasicColumn, 0, len(schema))
+		for _, col := range schema {
+			fT, _ := sql.MySQLColToFieldType(col)
+			columns = append(columns, BasicColumn{
+				Name:               col.Name,
+				MySQLType:          col.Type.String(),
+				Nullable:           col.Nullable,
+				DataFrameFieldType: fT,
+			})
+		}
+
+		// Cap at 3 rows.
+		const maxRows = 3
+		n := frame.Rows()
+		if n > maxRows {
+			n = maxRows
+		}
+		sampleRows := make([][]any, 0, n)
+		for i := 0; i < n; i++ {
+			sampleRows = append(sampleRows, frame.RowCopy(i))
+		}
+
+		schemas[dsNode.RefID()] = SchemaInfo{Columns: columns, SampleRows: sampleRows}
+	}
+
+	return schemas, nil
+}
+
+// DeepCopy returns a deep copy of the schema.
+// Used AI to make it, the kubernetes one doesn't like any or interface{}
+func (s SQLSchemas) DeepCopy() SQLSchemas {
+	if s == nil {
+		return nil
+	}
+	out := make(SQLSchemas, len(s))
+	for k, v := range s {
+		out[k] = SchemaInfo{
+			Columns:    copyColumns(v.Columns),
+			SampleRows: deepCopySampleRows2D(v.SampleRows),
+			Error:      v.Error,
+		}
+	}
+	return out
+}
+
+func copyColumns(in []BasicColumn) []BasicColumn {
+	if in == nil {
+		return nil
+	}
+	out := make([]BasicColumn, len(in))
+	copy(out, in) // BasicColumn is value-only, so this suffices
+	return out
+}
+
+// Deep-copy [][]any preserving nil vs empty slices and cloning elements.
+func deepCopySampleRows2D(in [][]any) [][]any {
+	if in == nil {
+		return nil
+	}
+	out := make([][]any, len(in))
+	for i, row := range in {
+		if row == nil {
+			// preserve nil inner slice
+			continue
+		}
+		newRow := make([]any, len(row))
+		for j, v := range row {
+			newRow[j] = deepCopyAny(v)
+		}
+		out[i] = newRow
+	}
+	return out
+}
+
+// Recursively clone pointers, maps, slices, arrays, and interfaces.
+// Structs are copied by value (shallow for their internals).
+func deepCopyAny(v any) any {
+	if v == nil {
+		return nil
+	}
+	return deepCopyRV(reflect.ValueOf(v)).Interface()
+}
+
+func deepCopyRV(rv reflect.Value) reflect.Value {
+	if !rv.IsValid() {
+		return rv
+	}
+
+	switch rv.Kind() {
+	case reflect.Ptr:
+		if rv.IsNil() {
+			return rv
+		}
+		elemCopy := deepCopyRV(rv.Elem())
+		newPtr := reflect.New(rv.Type().Elem())
+		if elemCopy.Type().AssignableTo(newPtr.Elem().Type()) {
+			newPtr.Elem().Set(elemCopy)
+		} else if elemCopy.Type().ConvertibleTo(newPtr.Elem().Type()) {
+			newPtr.Elem().Set(elemCopy.Convert(newPtr.Elem().Type()))
+		} else {
+			newPtr.Elem().Set(rv.Elem()) // fallback: shallow
+		}
+		return newPtr
+
+	case reflect.Interface:
+		if rv.IsNil() {
+			return rv
+		}
+		return deepCopyRV(rv.Elem())
+
+	case reflect.Map:
+		if rv.IsNil() {
+			return reflect.Zero(rv.Type())
+		}
+		newMap := reflect.MakeMapWithSize(rv.Type(), rv.Len())
+		for _, k := range rv.MapKeys() {
+			newMap.SetMapIndex(deepCopyRV(k), deepCopyRV(rv.MapIndex(k)))
+		}
+		return newMap
+
+	case reflect.Slice:
+		if rv.IsNil() {
+			return reflect.Zero(rv.Type())
+		}
+		n := rv.Len()
+		newSlice := reflect.MakeSlice(rv.Type(), n, n)
+		for i := 0; i < n; i++ {
+			newSlice.Index(i).Set(deepCopyRV(rv.Index(i)))
+		}
+		return newSlice
+
+	case reflect.Array:
+		n := rv.Len()
+		newArr := reflect.New(rv.Type()).Elem()
+		for i := 0; i < n; i++ {
+			newArr.Index(i).Set(deepCopyRV(rv.Index(i)))
+		}
+		return newArr
+
+	case reflect.Struct:
+		// Value copy (OK unless the struct contains references you also want deep-copied).
+		return rv
+
+	default:
+		// Scalars (string, bool, numbers), etc.
+		return rv
+	}
+}

pkg/registry/apis/query/query.go

@@ -241,25 +241,40 @@ func (r *queryREST) Connect(connectCtx context.Context, name string, _ runtime.O
 	}), nil
 }
 
-func handleQuery(ctx context.Context, raw query.QueryDataRequest, b QueryAPIBuilder, httpreq *http.Request, responder responderWrapper, connectLogger log.Logger) (*backend.QueryDataResponse, error) {
-	var jsonQueries = make([]*simplejson.Json, 0, len(raw.Queries))
-	for _, query := range raw.Queries {
-		dsRef, err := getValidDataSourceRef(ctx, query.Datasource, query.DatasourceID, b.legacyDatasourceLookup)
-		if err != nil {
-			connectLogger.Error("error getting valid datasource ref", err)
-		}
-		if dsRef != nil {
-			query.Datasource = dsRef
+type preparedQuery struct {
+	mReq          dtos.MetricRequest
+	cache         datasources.CacheService
+	headers       map[string]string
+	logger        log.Logger
+	builder       dsquerierclient.QSDatasourceClientBuilder
+	exprSvc       *expr.Service
+	reportMetrics func()
+}
+
+func prepareQuery(
+	ctx context.Context,
+	raw query.QueryDataRequest,
+	b QueryAPIBuilder,
+	httpreq *http.Request,
+	connectLogger log.Logger,
+) (*preparedQuery, error) {
+	// Normalize DS refs and build []*simplejson.Json
+	jsonQueries := make([]*simplejson.Json, 0, len(raw.Queries))
+	for _, q := range raw.Queries {
+		if dsRef, derr := getValidDataSourceRef(ctx, q.Datasource, q.DatasourceID, b.legacyDatasourceLookup); derr != nil {
+			connectLogger.Error("error getting valid datasource ref", "err", derr)
+		} else if dsRef != nil {
+			q.Datasource = dsRef
 		}
 
-		jsonBytes, err := json.Marshal(query)
+		jsonBytes, err := json.Marshal(q)
 		if err != nil {
-			connectLogger.Error("error marshalling", err)
+			connectLogger.Error("error marshalling query", "err", err)
 		}
 
-		sjQuery, _ := simplejson.NewJson(jsonBytes)
+		sjQuery, err := simplejson.NewJson(jsonBytes)
 		if err != nil {
-			connectLogger.Error("error unmarshalling", err)
+			connectLogger.Error("error creating simplejson for query", "err", err)
 		}
 
 		jsonQueries = append(jsonQueries, sjQuery)
@@ -274,13 +289,11 @@ func handleQuery(ctx context.Context, raw query.QueryDataRequest, b QueryAPIBuil
 	cache := &MyCacheService{
 		legacy: b.legacyDatasourceLookup,
 	}
-
 	headers := ExtractKnownHeaders(httpreq.Header)
 
 	instance, err := b.instanceProvider.GetInstance(ctx, connectLogger, headers)
 	if err != nil {
 		connectLogger.Error("failed to get instance configuration settings", "err", err)
-		responder.Error(err)
 		return nil, err
 	}
 
@@ -288,12 +301,14 @@ func handleQuery(ctx context.Context, raw query.QueryDataRequest, b QueryAPIBuil
 
 	dsQuerierLoggerWithSlug := instance.GetLogger()
 
+	// Datasource client qsDsClientBuilder
 	qsDsClientBuilder := dsquerierclient.NewQsDatasourceClientBuilderWithInstance(
 		instance,
 		ctx,
 		dsQuerierLoggerWithSlug,
 	)
 
+	// Expressions service
 	exprService := expr.ProvideService(
 		&setting.Cfg{
 			ExpressionsEnabled:            instanceConfig.ExpressionsEnabled,
@@ -310,17 +325,37 @@ func handleQuery(ctx context.Context, raw query.QueryDataRequest, b QueryAPIBuil
 		qsDsClientBuilder,
 	)
 
-	qdr, err := service.QueryData(ctx, dsQuerierLoggerWithSlug, cache, exprService, mReq, qsDsClientBuilder, headers)
+	return &preparedQuery{
+		mReq:          mReq,
+		cache:         cache,
+		headers:       headers,
+		logger:        dsQuerierLoggerWithSlug,
+		builder:       qsDsClientBuilder,
+		exprSvc:       exprService,
+		reportMetrics: func() { instance.ReportMetrics() },
+	}, nil
+}
 
-	// tell the `instance` structure that it can now report
-	// metrics that are only reported once during a request
-	instance.ReportMetrics()
+func handlePreparedQuery(ctx context.Context, pq *preparedQuery) (*backend.QueryDataResponse, error) {
+	resp, err := service.QueryData(ctx, pq.logger, pq.cache, pq.exprSvc, pq.mReq, pq.builder, pq.headers)
+	pq.reportMetrics()
+	return resp, err
+}
 
+func handleQuery(
+	ctx context.Context,
+	raw query.QueryDataRequest,
+	b QueryAPIBuilder,
+	httpreq *http.Request,
+	responder responderWrapper,
+	connectLogger log.Logger,
+) (*backend.QueryDataResponse, error) {
+	pq, err := prepareQuery(ctx, raw, b, httpreq, connectLogger)
 	if err != nil {
-		return qdr, err
+		responder.Error(err)
+		return nil, err
 	}
-
-	return qdr, nil
+	return handlePreparedQuery(ctx, pq)
 }
 
 type responderWrapper struct {

pkg/registry/apis/query/register.go

@@ -161,6 +161,7 @@ func addKnownTypes(scheme *runtime.Scheme, gv schema.GroupVersion) {
 		&query.QueryDataResponse{},
 		&query.QueryTypeDefinition{},
 		&query.QueryTypeDefinitionList{},
+		&query.SQLSchemas{},
 	)
 }
 
@@ -201,6 +202,8 @@ func (b *QueryAPIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserver.APIG
 	// The query endpoint -- NOTE, this uses a rewrite hack to allow requests without a name parameter
 	storage["query"] = newQueryREST(b)
 
+	storage["sqlschemas"] = newSQLSchemasREST(b)
+
 	// Register the expressions query schemas
 	err := queryschema.RegisterQueryTypes(b.queryTypes, storage)
 

pkg/registry/apis/query/sql_schema.go

@@ -0,0 +1,170 @@
+package query
+
+import (
+	"context"
+	"net/http"
+	"strconv"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana/pkg/expr"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	errorsK8s "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apiserver/pkg/endpoints/request"
+	"k8s.io/apiserver/pkg/registry/rest"
+
+	query "github.com/grafana/grafana/pkg/apis/query/v0alpha1"
+	"github.com/grafana/grafana/pkg/infra/log"
+	service "github.com/grafana/grafana/pkg/services/query"
+	"github.com/grafana/grafana/pkg/web"
+)
+
+type sqlSchemaREST struct {
+	logger  log.Logger
+	builder *QueryAPIBuilder
+}
+
+var (
+	_ rest.Storage              = (*sqlSchemaREST)(nil)
+	_ rest.SingularNameProvider = (*sqlSchemaREST)(nil)
+	_ rest.Connecter            = (*sqlSchemaREST)(nil)
+	_ rest.Scoper               = (*sqlSchemaREST)(nil)
+	_ rest.StorageMetadata      = (*sqlSchemaREST)(nil)
+)
+
+func newSQLSchemasREST(builder *QueryAPIBuilder) *sqlSchemaREST {
+	return &sqlSchemaREST{
+		logger:  log.New("query.sqlschemas"),
+		builder: builder,
+	}
+}
+
+func (r *sqlSchemaREST) New() runtime.Object {
+	// This is added as the "ResponseType" regardless what ProducesObject() says :)
+	return &query.SQLSchemas{}
+}
+
+func (r *sqlSchemaREST) Destroy() {}
+
+func (r *sqlSchemaREST) NamespaceScoped() bool {
+	return true
+}
+
+func (r *sqlSchemaREST) GetSingularName() string {
+	return "SQLSchema" // Used for the
+}
+
+func (r *sqlSchemaREST) ProducesMIMETypes(verb string) []string {
+	return []string{"application/json"} // and parquet!
+}
+
+func (r *sqlSchemaREST) ProducesObject(verb string) interface{} {
+	return &query.SQLSchemas{}
+}
+
+func (r *sqlSchemaREST) ConnectMethods() []string {
+	return []string{"POST"}
+}
+
+func (r *sqlSchemaREST) NewConnectOptions() (runtime.Object, bool, string) {
+	return nil, false, "" // true means you can use the trailing path as a variable
+}
+
+// called by mt query service and also when queryServiceFromUI is enabled, can be both mt and st
+func (r *sqlSchemaREST) Connect(connectCtx context.Context, name string, _ runtime.Object, incomingResponder rest.Responder) (http.Handler, error) {
+	// See: /pkg/services/apiserver/builder/helper.go#L34
+	// The name is set with a rewriter hack
+	if name != "name" {
+		r.logger.Debug("Connect name is not name")
+		return nil, errorsK8s.NewNotFound(schema.GroupResource{}, name)
+	}
+	b := r.builder
+
+	return http.HandlerFunc(func(w http.ResponseWriter, httpreq *http.Request) {
+		ctx, span := b.tracer.Start(httpreq.Context(), "QueryService.GetSQLSchemas")
+		defer span.End()
+		ctx = request.WithNamespace(ctx, request.NamespaceValue(connectCtx))
+		traceId := span.SpanContext().TraceID()
+		connectLogger := b.log.New("traceId", traceId.String(), "rule_uid", httpreq.Header.Get("X-Rule-Uid"))
+		responder := newResponderWrapper(incomingResponder,
+			func(statusCode *int, obj runtime.Object) {
+				if *statusCode/100 == 4 {
+					span.SetStatus(codes.Error, strconv.Itoa(*statusCode))
+				}
+
+				if *statusCode >= 500 {
+					o, ok := obj.(*query.QueryDataResponse)
+					if ok && o.Responses != nil {
+						for refId, response := range o.Responses {
+							if response.ErrorSource == backend.ErrorSourceDownstream {
+								*statusCode = http.StatusBadRequest //force this to be a 400 since it's downstream
+								span.SetStatus(codes.Error, strconv.Itoa(*statusCode))
+								span.SetAttributes(attribute.String("error.source", "downstream"))
+								break
+							} else if response.Error != nil {
+								connectLogger.Debug("500 error without downstream error source", "error", response.Error, "errorSource", response.ErrorSource, "refId", refId)
+								span.SetStatus(codes.Error, "500 error without downstream error source")
+							} else {
+								span.SetStatus(codes.Error, "500 error without downstream error source and no Error message")
+								span.SetAttributes(attribute.String("error.ref_id", refId))
+							}
+						}
+					}
+				}
+			},
+
+			func(err error) {
+				connectLogger.Error("error caught in handler", "err", err)
+				span.SetStatus(codes.Error, "query error")
+
+				if err == nil {
+					return
+				}
+
+				span.RecordError(err)
+			})
+
+		raw := &query.QueryDataRequest{}
+		err := web.Bind(httpreq, raw)
+		if err != nil {
+			connectLogger.Error("Hit unexpected error when reading query", "err", err)
+			err = errorsK8s.NewBadRequest("error reading query")
+			responder.Error(err)
+			return
+		}
+
+		qdr, err := handleSQLSchemaQuery(ctx, *raw, *b, httpreq, *responder, connectLogger)
+		if err != nil {
+			responder.Error(err)
+			return
+		}
+
+		responder.Object(200, &query.SQLSchemas{
+			SQLSchemas: qdr,
+		})
+	}), nil
+}
+
+func handlePreparedSQLSchema(ctx context.Context, pq *preparedQuery) (expr.SQLSchemas, error) {
+	resp, err := service.GetSQLSchemas(ctx, pq.logger, pq.cache, pq.exprSvc, pq.mReq, pq.builder, pq.headers)
+	pq.reportMetrics()
+	return resp, err
+}
+
+func handleSQLSchemaQuery(
+	ctx context.Context,
+	raw query.QueryDataRequest,
+	b QueryAPIBuilder,
+	httpreq *http.Request,
+	responder responderWrapper,
+	connectLogger log.Logger,
+) (expr.SQLSchemas, error) {
+	pq, err := prepareQuery(ctx, raw, b, httpreq, connectLogger)
+	if err != nil {
+		responder.Error(err)
+		return nil, err
+	}
+	return handlePreparedSQLSchema(ctx, pq)
+}

pkg/services/apiserver/builder/helper.go

@@ -58,6 +58,12 @@ var PathRewriters = []filters.PathRewriter{
 			return matches[1] + "/name" // connector requires a name
 		},
 	},
+	{
+		Pattern: regexp.MustCompile(`(/apis/query.grafana.app/v0alpha1/namespaces/.*/sqlschemas$)`),
+		ReplaceFunc: func(matches []string) string {
+			return matches[1] + "/name" // connector requires a name
+		},
+	},
 	{
 		Pattern: regexp.MustCompile(`(/apis/.*/v0alpha1/namespaces/.*/queryconvert$)`),
 		ReplaceFunc: func(matches []string) string {

pkg/services/publicdashboards/public_dashboard_service_mock.go

@@ -4,9 +4,12 @@ package publicdashboards
 
 import (
 	context "context"
+	"fmt"
 
 	backend "github.com/grafana/grafana-plugin-sdk-go/backend"
 
+	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/expr"
 	dashboards "github.com/grafana/grafana/pkg/services/dashboards"
 
 	dtos "github.com/grafana/grafana/pkg/api/dtos"
@@ -587,6 +590,10 @@ func (_m *FakePublicDashboardService) Update(ctx context.Context, u *user.Signed
 	return r0, r1
 }
 
+func (_m *FakePublicDashboardService) GetSQLSchemas(ctx context.Context, user identity.Requester, reqDTO dtos.MetricRequest) (expr.SQLSchemas, error) {
+	return nil, fmt.Errorf("not implemented in public dashboards")
+}
+
 // NewFakePublicDashboardService creates a new instance of FakePublicDashboardService. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
 // The first argument is typically a *testing.T value.
 func NewFakePublicDashboardService(t interface {

pkg/services/publicdashboards/publicdashboard.go

@@ -5,6 +5,8 @@ import (
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana/pkg/api/dtos"
+	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/expr"
 	contextmodel "github.com/grafana/grafana/pkg/services/contexthandler/model"
 	"github.com/grafana/grafana/pkg/services/dashboards"
 	. "github.com/grafana/grafana/pkg/services/publicdashboards/models"
@@ -37,6 +39,8 @@ type Service interface {
 
 	ExistsEnabledByAccessToken(ctx context.Context, accessToken string) (bool, error)
 	ExistsEnabledByDashboardUid(ctx context.Context, dashboardUid string) (bool, error)
+
+	GetSQLSchemas(ctx context.Context, user identity.Requester, reqDTO dtos.MetricRequest) (expr.SQLSchemas, error)
 }
 
 // ServiceWrapper these methods have different behavior between OSS and Enterprise. The latter would call the OSS service first

pkg/services/publicdashboards/service/service.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/grafana/grafana/pkg/api/dtos"
 	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/expr"
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/infra/metrics"
 	"github.com/grafana/grafana/pkg/services/accesscontrol"
@@ -512,6 +513,10 @@ func (pd *PublicDashboardServiceImpl) logIsEnabledChanged(existingPubdash *Publi
 	}
 }
 
+func (pd *PublicDashboardServiceImpl) GetSQLSchemas(ctx context.Context, user identity.Requester, reqDTO dtos.MetricRequest) (expr.SQLSchemas, error) {
+	return nil, fmt.Errorf("sql schema endpoint not supported with public dashboards")
+}
+
 // Checks to see if PublicDashboard.ExistsEnabledByDashboardUid is true on create or changed on update
 func publicDashboardIsEnabledChanged(existingPubdash *PublicDashboard, newPubdash *PublicDashboard) bool {
 	// creating dashboard, enabled true

pkg/services/query/expr_sql_schema.go

@@ -0,0 +1,75 @@
+package query
+
+import (
+	"context"
+
+	"github.com/grafana/grafana/pkg/api/dtos"
+	"github.com/grafana/grafana/pkg/apimachinery/errutil"
+	"github.com/grafana/grafana/pkg/apimachinery/identity"
+	"github.com/grafana/grafana/pkg/expr"
+	"github.com/grafana/grafana/pkg/infra/log"
+	"github.com/grafana/grafana/pkg/services/datasources"
+	"github.com/grafana/grafana/pkg/services/dsquerierclient"
+	"github.com/grafana/grafana/pkg/services/validations"
+)
+
+func (s *ServiceImpl) GetSQLSchemas(ctx context.Context, user identity.Requester, reqDTO dtos.MetricRequest) (expr.SQLSchemas, error) {
+	//TODO DEdupe code
+
+	parsedReq, err := s.parseMetricRequest(ctx, user, false, reqDTO, false)
+	if err != nil {
+		return expr.SQLSchemas{}, err
+	}
+	exprReq := expr.Request{
+		Queries: []expr.Query{},
+	}
+
+	if user != nil { // for passthrough authentication, SSE does not authenticate
+		exprReq.User = user
+		exprReq.OrgId = user.GetOrgID()
+	}
+
+	for _, pq := range parsedReq.getFlattenedQueries() {
+		if pq.datasource == nil {
+			return nil, ErrMissingDataSourceInfo.Build(errutil.TemplateData{
+				Public: map[string]any{
+					"RefId": pq.query.RefID,
+				},
+			})
+		}
+
+		exprReq.Queries = append(exprReq.Queries, expr.Query{
+			JSON:          pq.query.JSON,
+			Interval:      pq.query.Interval,
+			RefID:         pq.query.RefID,
+			MaxDataPoints: pq.query.MaxDataPoints,
+			QueryType:     pq.query.QueryType,
+			DataSource:    pq.datasource,
+			TimeRange: expr.AbsoluteTimeRange{
+				From: pq.query.TimeRange.From,
+				To:   pq.query.TimeRange.To,
+			},
+		})
+	}
+
+	return s.expressionService.GetSQLSchemas(ctx, exprReq)
+}
+
+func GetSQLSchemas(ctx context.Context, log log.Logger, dscache datasources.CacheService, exprService *expr.Service, reqDTO dtos.MetricRequest, qsDatasourceClientBuilder dsquerierclient.QSDatasourceClientBuilder, headers map[string]string) (expr.SQLSchemas, error) {
+	s := &ServiceImpl{
+		log:                        log,
+		dataSourceCache:            dscache,
+		expressionService:          exprService,
+		dataSourceRequestValidator: validations.ProvideValidator(),
+		qsDatasourceClientBuilder:  qsDatasourceClientBuilder,
+		headers:                    headers,
+		concurrentQueryLimit:       16, // TODO: make it configurable
+	}
+
+	user, err := identity.GetRequester(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return s.GetSQLSchemas(ctx, user, reqDTO)
+}

pkg/services/query/query.go

@@ -73,6 +73,8 @@ type Service interface {
 
 	// this is more "forward compatible", for example supports per-query time ranges
 	QueryDataNew(ctx context.Context, user identity.Requester, skipDSCache bool, reqDTO dtos.MetricRequest) (*backend.QueryDataResponse, error)
+
+	GetSQLSchemas(ctx context.Context, user identity.Requester, reqDTO dtos.MetricRequest) (expr.SQLSchemas, error)
 }
 
 // Gives us compile time error if the service does not adhere to the contract of the interface

pkg/services/query/query_service_mock.go

@@ -4,10 +4,12 @@ package query
 
 import (
 	context "context"
+	"fmt"
 
 	backend "github.com/grafana/grafana-plugin-sdk-go/backend"
 
 	dtos "github.com/grafana/grafana/pkg/api/dtos"
+	"github.com/grafana/grafana/pkg/expr"
 
 	identity "github.com/grafana/grafana/pkg/apimachinery/identity"
 
@@ -97,6 +99,10 @@ func (_m *FakeQueryService) Run(ctx context.Context) error {
 	return r0
 }
 
+func (_m *FakeQueryService) GetSQLSchemas(ctx context.Context, user identity.Requester, reqDTO dtos.MetricRequest) (expr.SQLSchemas, error) {
+	return nil, fmt.Errorf("sql schema endpoint not supported with public dashboards")
+}
+
 // NewFakeQueryService creates a new instance of FakeQueryService. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
 // The first argument is typically a *testing.T value.
 func NewFakeQueryService(t interface {