diff --git a/go.mod b/go.mod
index d0e62735554..0098c0eb73b 100644
--- a/go.mod
+++ b/go.mod
@@ -29,7 +29,7 @@ require (
 	github.com/andybalholm/brotli v1.1.1 // @grafana/partner-datasources
 	github.com/apache/arrow-go/v18 v18.0.1-0.20241212180703-82be143d7c30 // @grafana/plugins-platform-backend
 	github.com/armon/go-radix v1.0.0 // @grafana/grafana-app-platform-squad
-	github.com/aws/aws-sdk-go v1.55.5 // @grafana/aws-datasources
+	github.com/aws/aws-sdk-go v1.55.6 // @grafana/aws-datasources
 	github.com/beevik/etree v1.4.1 // @grafana/grafana-backend-group
 	github.com/benbjohnson/clock v1.3.5 // @grafana/alerting-backend
 	github.com/blang/semver/v4 v4.0.0 // indirect; @grafana/grafana-developer-enablement-squad
diff --git a/go.sum b/go.sum
index 96df6362cc9..b5f34629318 100644
--- a/go.sum
+++ b/go.sum
@@ -839,8 +839,8 @@ github.com/aws/aws-sdk-go v1.17.7/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN
 github.com/aws/aws-sdk-go v1.22.4/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/aws/aws-sdk-go v1.38.35/go.mod h1:hcU610XS61/+aQV88ixoOzUoG7v3b31pl2zKMmprdro=
 github.com/aws/aws-sdk-go v1.50.29/go.mod h1:LF8svs817+Nz+DmiMQKTO3ubZ/6IaTpq3TjupRn3Eqk=
-github.com/aws/aws-sdk-go v1.55.5 h1:KKUZBfBoyqy5d3swXyiC7Q76ic40rYcbqH7qjh59kzU=
-github.com/aws/aws-sdk-go v1.55.5/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
+github.com/aws/aws-sdk-go v1.55.6 h1:cSg4pvZ3m8dgYcgqB97MrcdjUmZ1BeMYKUxMMB89IPk=
+github.com/aws/aws-sdk-go v1.55.6/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
 github.com/aws/aws-sdk-go-v2 v1.30.3 h1:jUeBtG0Ih+ZIFH0F4UkmL9w3cSpaMv9tYYDbzILP8dY=
 github.com/aws/aws-sdk-go-v2 v1.30.3/go.mod h1:nIQjQVp5sfpQcTc9mPSr1B0PaWK5ByX9MOoDadSN4lc=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.3 h1:tW1/Rkad38LA15X4UQtjXZXNKsCgkshC3EbmcUmghTg=
diff --git a/pkg/build/go.mod b/pkg/build/go.mod
index 14352adba3e..ed6f08c804c 100644
--- a/pkg/build/go.mod
+++ b/pkg/build/go.mod
@@ -10,7 +10,7 @@ replace github.com/docker/docker => github.com/moby/moby v27.5.1+incompatible
 require (
 	cloud.google.com/go/storage v1.50.0 // @grafana/grafana-backend-group
 	github.com/Masterminds/semver/v3 v3.3.0 // @grafana/grafana-developer-enablement-squad
-	github.com/aws/aws-sdk-go v1.55.5 // @grafana/aws-datasources
+	github.com/aws/aws-sdk-go v1.55.6 // @grafana/aws-datasources
 	github.com/docker/docker v27.5.1+incompatible // @grafana/grafana-developer-enablement-squad
 	github.com/drone/drone-cli v1.8.0 // @grafana/grafana-developer-enablement-squad
 	github.com/gogo/protobuf v1.3.2 // indirect; @grafana/alerting-backend
diff --git a/pkg/build/go.sum b/pkg/build/go.sum
index 1d55ac2a88b..55b8c06217d 100644
--- a/pkg/build/go.sum
+++ b/pkg/build/go.sum
@@ -53,8 +53,8 @@ github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuy
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883 h1:bvNMNQO63//z+xNgfBlViaCIJKLlCJ6/fmUseuG0wVQ=
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883/go.mod h1:rCTlJbsFo29Kk6CurOXKm700vrz8f0KW0JNfpkRJY/8=
-github.com/aws/aws-sdk-go v1.55.5 h1:KKUZBfBoyqy5d3swXyiC7Q76ic40rYcbqH7qjh59kzU=
-github.com/aws/aws-sdk-go v1.55.5/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
+github.com/aws/aws-sdk-go v1.55.6 h1:cSg4pvZ3m8dgYcgqB97MrcdjUmZ1BeMYKUxMMB89IPk=
+github.com/aws/aws-sdk-go v1.55.6/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
 github.com/bmatcuk/doublestar v1.1.1/go.mod h1:UD6OnuiIn0yFxxA2le/rnRU1G4RaI4UvFv1sNto9p6w=
 github.com/bmatcuk/doublestar v1.3.4 h1:gPypJ5xD31uhX6Tf54sDPUOBXTqKH4c9aPY66CyQrS0=
 github.com/bmatcuk/doublestar v1.3.4/go.mod h1:wiQtGV+rzVYxB7WIlirSN++5HPtPlXEo9MEoZQC/PmE=
diff --git a/pkg/expr/convert_from_full_long.go b/pkg/expr/convert_from_full_long.go
new file mode 100644
index 00000000000..c4de09cbee3
--- /dev/null
+++ b/pkg/expr/convert_from_full_long.go
@@ -0,0 +1,128 @@
+package expr
+
+import (
+	"fmt"
+
+	"github.com/grafana/grafana-plugin-sdk-go/data"
+)
+
+func ConvertFromFullLongToNumericMulti(frames data.Frames) (data.Frames, error) {
+	if len(frames) != 1 {
+		return nil, fmt.Errorf("expected exactly one frame, got %d", len(frames))
+	}
+	frame := frames[0]
+	if frame.Meta == nil || frame.Meta.Type != numericFullLongType {
+		return nil, fmt.Errorf("expected frame of type %q", numericFullLongType)
+	}
+
+	var (
+		metricField  *data.Field
+		valueField   *data.Field
+		displayField *data.Field
+		labelFields  []*data.Field
+	)
+
+	// Identify key fields
+	for _, f := range frame.Fields {
+		switch f.Name {
+		case SQLMetricFieldName:
+			metricField = f
+		case SQLValueFieldName:
+			valueField = f
+		case SQLDisplayFieldName:
+			displayField = f
+		default:
+			if f.Type() == data.FieldTypeNullableString {
+				labelFields = append(labelFields, f)
+			}
+		}
+	}
+
+	if metricField == nil || valueField == nil {
+		return nil, fmt.Errorf("missing required fields: %q or %q", SQLMetricFieldName, SQLValueFieldName)
+	}
+
+	type seriesKey struct {
+		metric      string
+		labelFP     data.Fingerprint
+		displayName string
+	}
+
+	type seriesEntry struct {
+		indices     []int
+		labels      data.Labels
+		displayName *string
+	}
+
+	grouped := make(map[seriesKey]*seriesEntry)
+
+	for i := 0; i < frame.Rows(); i++ {
+		if valueField.NilAt(i) {
+			continue // skip null values
+		}
+
+		metric := metricField.At(i).(string)
+
+		// collect labels
+		labels := data.Labels{}
+		for _, f := range labelFields {
+			if f.NilAt(i) {
+				continue
+			}
+			val := f.At(i).(*string)
+			if val != nil {
+				labels[f.Name] = *val
+			}
+		}
+		fp := labels.Fingerprint()
+
+		// handle optional display name
+		var displayPtr *string
+		displayKey := ""
+		if displayField != nil && !displayField.NilAt(i) {
+			if raw := displayField.At(i).(*string); raw != nil {
+				displayPtr = raw
+				displayKey = *raw
+			}
+		}
+
+		key := seriesKey{
+			metric:      metric,
+			labelFP:     fp,
+			displayName: displayKey,
+		}
+
+		entry, ok := grouped[key]
+		if !ok {
+			entry = &seriesEntry{
+				labels:      labels,
+				displayName: displayPtr,
+			}
+			grouped[key] = entry
+		}
+		entry.indices = append(entry.indices, i)
+	}
+
+	var result data.Frames
+	for key, entry := range grouped {
+		values := make([]*float64, 0, len(entry.indices))
+		for _, i := range entry.indices {
+			v, err := valueField.FloatAt(i)
+			if err != nil {
+				return nil, fmt.Errorf("failed to convert value at index %d to float: %w", i, err)
+			}
+			values = append(values, &v)
+		}
+
+		field := data.NewField(key.metric, entry.labels, values)
+		if entry.displayName != nil {
+			field.Config = &data.FieldConfig{DisplayNameFromDS: *entry.displayName}
+		}
+
+		frame := data.NewFrame("", field)
+		frame.Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+		result = append(result, frame)
+	}
+
+	return result, nil
+}
diff --git a/pkg/expr/convert_from_full_long_test.go b/pkg/expr/convert_from_full_long_test.go
new file mode 100644
index 00000000000..2487fac661d
--- /dev/null
+++ b/pkg/expr/convert_from_full_long_test.go
@@ -0,0 +1,192 @@
+package expr
+
+import (
+	"sort"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConvertFromFullLongToNumericMulti(t *testing.T) {
+	t.Run("SingleRowNoLabels", func(t *testing.T) {
+		input := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(3.14)}),
+		)
+		input.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		out, err := ConvertFromFullLongToNumericMulti(data.Frames{input})
+		require.NoError(t, err)
+		require.Len(t, out, 1)
+
+		expected := data.NewFrame("",
+			data.NewField("cpu", nil, []*float64{fp(3.14)}),
+		)
+		expected.Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+
+		if diff := cmp.Diff(expected, out[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Mismatch (-want +got):\n%s", diff)
+		}
+	})
+
+	t.Run("TwoRowsWithLabelsAndDisplay", func(t *testing.T) {
+		input := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(2.0)}),
+			data.NewField(SQLDisplayFieldName, nil, []*string{sp("CPU A"), sp("CPU A")}),
+			data.NewField("host", nil, []*string{sp("a"), sp("a")}),
+		)
+		input.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		out, err := ConvertFromFullLongToNumericMulti(data.Frames{input})
+		require.NoError(t, err)
+		require.Len(t, out, 1)
+
+		expected := data.NewFrame("",
+			func() *data.Field {
+				f := data.NewField("cpu", data.Labels{"host": "a"}, []*float64{fp(1.0), fp(2.0)})
+				f.Config = &data.FieldConfig{DisplayNameFromDS: "CPU A"}
+				return f
+			}(),
+		)
+		expected.Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+
+		if diff := cmp.Diff(expected, out[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Mismatch (-want +got):\n%s", diff)
+		}
+	})
+
+	t.Run("SkipsNullValues", func(t *testing.T) {
+		input := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), nil}),
+		)
+		input.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		out, err := ConvertFromFullLongToNumericMulti(data.Frames{input})
+		require.NoError(t, err)
+		require.Len(t, out, 1)
+
+		expected := data.NewFrame("",
+			data.NewField("cpu", nil, []*float64{fp(1.0)}),
+		)
+		expected.Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+
+		if diff := cmp.Diff(expected, out[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Mismatch (-want +got):\n%s", diff)
+		}
+	})
+}
+
+func TestConvertNumericMultiRoundTripToFullLongAndBack(t *testing.T) {
+	t.Run("TwoFieldsWithSparseLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				data.NewField("cpu", data.Labels{"host": "a"}, []*float64{fp(1.0)}),
+			),
+			data.NewFrame("",
+				data.NewField("cpu", data.Labels{"host": "b", "env": "prod"}, []*float64{fp(2.0)}),
+			),
+		}
+		for _, f := range input {
+			f.Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+		}
+
+		fullLong, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, fullLong, 1)
+
+		roundTrip, err := ConvertFromFullLongToNumericMulti(fullLong)
+		require.NoError(t, err)
+
+		expected := data.Frames{
+			data.NewFrame("",
+				data.NewField("cpu", data.Labels{"host": "a"}, []*float64{fp(1.0)}),
+			),
+			data.NewFrame("",
+				data.NewField("cpu", data.Labels{"host": "b", "env": "prod"}, []*float64{fp(2.0)}),
+			),
+		}
+		for _, f := range expected {
+			f.Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+		}
+
+		sortFramesByMetricDisplayAndLabels(expected)
+		sortFramesByMetricDisplayAndLabels(roundTrip)
+
+		require.Len(t, roundTrip, len(expected))
+		for i := range expected {
+			if diff := cmp.Diff(expected[i], roundTrip[i], data.FrameTestCompareOptions()...); diff != "" {
+				t.Errorf("Mismatch on frame %d (-want +got):\n%s", i, diff)
+			}
+		}
+	})
+
+	t.Run("PreservesDisplayName", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				func() *data.Field {
+					f := data.NewField("cpu", data.Labels{"host": "a"}, []*float64{fp(1.0)})
+					f.Config = &data.FieldConfig{DisplayNameFromDS: "CPU A"}
+					return f
+				}(),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+
+		fullLong, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, fullLong, 1)
+
+		roundTrip, err := ConvertFromFullLongToNumericMulti(fullLong)
+		require.NoError(t, err)
+
+		expected := data.Frames{
+			data.NewFrame("",
+				func() *data.Field {
+					f := data.NewField("cpu", data.Labels{"host": "a"}, []*float64{fp(1.0)})
+					f.Config = &data.FieldConfig{DisplayNameFromDS: "CPU A"}
+					return f
+				}(),
+			),
+		}
+		expected[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+
+		sortFramesByMetricDisplayAndLabels(expected)
+		sortFramesByMetricDisplayAndLabels(roundTrip)
+
+		require.Len(t, roundTrip, 1)
+		if diff := cmp.Diff(expected[0], roundTrip[0], data.FrameTestCompareOptions()...); diff != "" {
+			t.Errorf("Mismatch (-want +got):\n%s", diff)
+		}
+	})
+}
+
+func sortFramesByMetricDisplayAndLabels(frames data.Frames) {
+	sort.Slice(frames, func(i, j int) bool {
+		fi := frames[i].Fields[0]
+		fj := frames[j].Fields[0]
+
+		// 1. Metric name
+		if fi.Name != fj.Name {
+			return fi.Name < fj.Name
+		}
+
+		// 2. Display name (if set)
+		var di, dj string
+		if fi.Config != nil {
+			di = fi.Config.DisplayNameFromDS
+		}
+		if fj.Config != nil {
+			dj = fj.Config.DisplayNameFromDS
+		}
+		if di != dj {
+			return di < dj
+		}
+
+		// 3. Labels fingerprint
+		return fi.Labels.Fingerprint() < fj.Labels.Fingerprint()
+	})
+}
diff --git a/pkg/expr/convert_to_full_long.go b/pkg/expr/convert_to_full_long.go
new file mode 100644
index 00000000000..94dc2648bd6
--- /dev/null
+++ b/pkg/expr/convert_to_full_long.go
@@ -0,0 +1,400 @@
+package expr
+
+import (
+	"fmt"
+	"sort"
+	"time"
+
+	"github.com/grafana/grafana-plugin-sdk-go/data"
+)
+
+const (
+	SQLMetricFieldName  = "__metric_name__"
+	SQLValueFieldName   = "__value__"
+	SQLDisplayFieldName = "__display_name__"
+
+	// These are not types in the SDK or dataplane contract yet.
+	numericFullLongType    = "numeric_full_long"
+	timeseriesFullLongType = "time_series_full_long"
+)
+
+func ConvertToFullLong(frames data.Frames) (data.Frames, error) {
+	if len(frames) == 0 {
+		return frames, nil
+	}
+
+	var inputType data.FrameType
+	if frames[0].Meta != nil && frames[0].Meta.Type != "" {
+		inputType = frames[0].Meta.Type
+	} else {
+		return nil, fmt.Errorf("input frame missing FrameMeta.Type")
+	}
+
+	if !supportedToLongConversion(inputType) {
+		return nil, fmt.Errorf("unsupported input dataframe type %s for full long conversion", inputType)
+	}
+
+	switch inputType {
+	case data.FrameTypeNumericMulti:
+		return convertNumericMultiToFullLong(frames)
+	case data.FrameTypeNumericWide:
+		return convertNumericWideToFullLong(frames)
+	case data.FrameTypeTimeSeriesMulti:
+		return convertTimeSeriesMultiToFullLong(frames)
+	case data.FrameTypeTimeSeriesWide:
+		return convertTimeSeriesWideToFullLong(frames)
+	default:
+		return nil, fmt.Errorf("unsupported input type %s for full long conversion", inputType)
+	}
+}
+
+func convertNumericMultiToFullLong(frames data.Frames) (data.Frames, error) {
+	wide := convertNumericMultiToNumericWide(frames)
+	return convertNumericWideToFullLong(wide)
+}
+
+func convertNumericWideToFullLong(frames data.Frames) (data.Frames, error) {
+	if len(frames) != 1 {
+		return nil, fmt.Errorf("expected exactly one frame for wide format, but got %d", len(frames))
+	}
+	inputFrame := frames[0]
+	if inputFrame.Rows() > 1 {
+		return nil, fmt.Errorf("expected no more than one row in the frame, but got %d", inputFrame.Rows())
+	}
+
+	var (
+		metricCol     = make([]string, 0, len(inputFrame.Fields))
+		valueCol      = make([]*float64, 0, len(inputFrame.Fields))
+		displayCol    = make([]*string, 0, len(inputFrame.Fields))
+		hasDisplayCol bool
+	)
+
+	labelKeySet := map[string]struct{}{}
+	for _, field := range inputFrame.Fields {
+		if !field.Type().Numeric() {
+			continue
+		}
+		val, err := field.FloatAt(0)
+		if err != nil {
+			continue
+		}
+		v := val
+		valueCol = append(valueCol, &v)
+		metricCol = append(metricCol, field.Name)
+
+		// Display name
+		var d *string
+		if field.Config != nil && field.Config.DisplayNameFromDS != "" {
+			s := field.Config.DisplayNameFromDS
+			d = &s
+			hasDisplayCol = true
+		}
+		displayCol = append(displayCol, d)
+
+		for k := range field.Labels {
+			labelKeySet[k] = struct{}{}
+		}
+	}
+
+	labelKeys := make([]string, 0, len(labelKeySet))
+
+	labelValues := make(map[string][]*string)
+	for k := range labelKeySet {
+		labelKeys = append(labelKeys, k)
+		labelValues[k] = make([]*string, 0, len(valueCol))
+	}
+	sort.Strings(labelKeys)
+
+	for _, field := range inputFrame.Fields {
+		if !field.Type().Numeric() {
+			continue
+		}
+		for _, k := range labelKeys {
+			var val *string
+			if field.Labels != nil {
+				if v, ok := field.Labels[k]; ok {
+					val = &v
+				}
+			}
+			labelValues[k] = append(labelValues[k], val)
+		}
+	}
+
+	fields := []*data.Field{
+		data.NewField(SQLMetricFieldName, nil, metricCol),
+		data.NewField(SQLValueFieldName, nil, valueCol),
+	}
+	if hasDisplayCol {
+		fields = append(fields, data.NewField(SQLDisplayFieldName, nil, displayCol))
+	}
+	for _, k := range labelKeys {
+		fields = append(fields, data.NewField(k, nil, labelValues[k]))
+	}
+
+	out := data.NewFrame("", fields...)
+	out.Meta = &data.FrameMeta{Type: numericFullLongType}
+	return data.Frames{out}, nil
+}
+
+func convertTimeSeriesMultiToFullLong(frames data.Frames) (data.Frames, error) {
+	type row struct {
+		t       time.Time
+		value   *float64
+		metric  string
+		display *string
+		labels  data.Labels
+	}
+
+	var rows []row
+	labelKeysSet := map[string]struct{}{}
+	hasDisplayCol := false
+
+	for _, frame := range frames {
+		var timeField *data.Field
+		for _, f := range frame.Fields {
+			if f.Type() == data.FieldTypeTime {
+				timeField = f
+				break
+			}
+		}
+		if timeField == nil {
+			return nil, fmt.Errorf("missing time field")
+		}
+		for _, f := range frame.Fields {
+			if !f.Type().Numeric() {
+				continue
+			}
+			var display *string
+			if f.Config != nil && f.Config.DisplayNameFromDS != "" {
+				s := f.Config.DisplayNameFromDS
+				display = &s
+				hasDisplayCol = true
+			}
+			for i := 0; i < f.Len(); i++ {
+				t := timeField.At(i).(time.Time)
+				v, err := f.FloatAt(i)
+				if err != nil {
+					continue
+				}
+				val := v
+				rows = append(rows, row{
+					t:       t,
+					value:   &val,
+					metric:  f.Name,
+					display: display,
+					labels:  f.Labels,
+				})
+				for k := range f.Labels {
+					labelKeysSet[k] = struct{}{}
+				}
+			}
+		}
+	}
+
+	labelKeys := make([]string, 0, len(labelKeysSet))
+	for k := range labelKeysSet {
+		labelKeys = append(labelKeys, k)
+	}
+	sort.Strings(labelKeys)
+	sort.SliceStable(rows, func(i, j int) bool {
+		if rows[i].t.Equal(rows[j].t) {
+			return rows[i].metric < rows[j].metric
+		}
+		return rows[i].t.Before(rows[j].t)
+	})
+
+	times := make([]time.Time, len(rows))
+	values := make([]*float64, len(rows))
+	metrics := make([]string, len(rows))
+	var displays []*string
+	if hasDisplayCol {
+		displays = make([]*string, len(rows))
+	}
+	labels := make(map[string][]*string)
+	for _, k := range labelKeys {
+		labels[k] = make([]*string, len(rows))
+	}
+
+	for i, r := range rows {
+		times[i] = r.t
+		values[i] = r.value
+		metrics[i] = r.metric
+		if hasDisplayCol {
+			displays[i] = r.display
+		}
+		for _, k := range labelKeys {
+			if v, ok := r.labels[k]; ok {
+				labels[k][i] = &v
+			}
+		}
+	}
+
+	fields := []*data.Field{
+		data.NewField("time", nil, times),
+		data.NewField(SQLValueFieldName, nil, values),
+		data.NewField(SQLMetricFieldName, nil, metrics),
+	}
+	if hasDisplayCol {
+		fields = append(fields, data.NewField(SQLDisplayFieldName, nil, displays))
+	}
+	for _, k := range labelKeys {
+		fields = append(fields, data.NewField(k, nil, labels[k]))
+	}
+
+	out := data.NewFrame("", fields...)
+	out.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+	return data.Frames{out}, nil
+}
+
+func convertTimeSeriesWideToFullLong(frames data.Frames) (data.Frames, error) {
+	if len(frames) != 1 {
+		return nil, fmt.Errorf("expected exactly one frame for wide format, but got %d", len(frames))
+	}
+	frame := frames[0]
+
+	var timeField *data.Field
+	for _, f := range frame.Fields {
+		if f.Type() == data.FieldTypeTime {
+			timeField = f
+			break
+		}
+	}
+	if timeField == nil {
+		return nil, fmt.Errorf("time field not found in TimeSeriesWide frame")
+	}
+
+	type row struct {
+		t       time.Time
+		value   *float64
+		metric  string
+		display *string
+		labels  data.Labels
+	}
+
+	var (
+		rows          []row
+		labelKeysSet  = map[string]struct{}{}
+		hasDisplayCol bool
+	)
+
+	// Collect all label keys
+	for _, f := range frame.Fields {
+		if !f.Type().Numeric() {
+			continue
+		}
+		for k := range f.Labels {
+			labelKeysSet[k] = struct{}{}
+		}
+	}
+
+	labelKeys := make([]string, 0, len(labelKeysSet))
+	for k := range labelKeysSet {
+		labelKeys = append(labelKeys, k)
+	}
+	sort.Strings(labelKeys)
+
+	timeLen := timeField.Len()
+	for _, f := range frame.Fields {
+		if !f.Type().Numeric() {
+			continue
+		}
+		var display *string
+		if f.Config != nil && f.Config.DisplayNameFromDS != "" {
+			s := f.Config.DisplayNameFromDS
+			display = &s
+			hasDisplayCol = true
+		}
+		for i := 0; i < timeLen; i++ {
+			t := timeField.At(i).(time.Time)
+			v, err := f.FloatAt(i)
+			if err != nil {
+				continue
+			}
+			val := v
+			rows = append(rows, row{
+				t:       t,
+				value:   &val,
+				metric:  f.Name,
+				display: display,
+				labels:  f.Labels,
+			})
+		}
+	}
+
+	sort.SliceStable(rows, func(i, j int) bool {
+		if rows[i].t.Equal(rows[j].t) {
+			return rows[i].metric < rows[j].metric
+		}
+		return rows[i].t.Before(rows[j].t)
+	})
+
+	times := make([]time.Time, len(rows))
+	values := make([]*float64, len(rows))
+	metrics := make([]string, len(rows))
+	var displays []*string
+	if hasDisplayCol {
+		displays = make([]*string, len(rows))
+	}
+	labels := make(map[string][]*string)
+	for _, k := range labelKeys {
+		labels[k] = make([]*string, len(rows))
+	}
+
+	for i, r := range rows {
+		times[i] = r.t
+		values[i] = r.value
+		metrics[i] = r.metric
+		if hasDisplayCol {
+			displays[i] = r.display
+		}
+		for _, k := range labelKeys {
+			if v, ok := r.labels[k]; ok {
+				labels[k][i] = &v
+			}
+		}
+	}
+
+	fields := []*data.Field{
+		data.NewField("time", nil, times),
+		data.NewField(SQLValueFieldName, nil, values),
+		data.NewField(SQLMetricFieldName, nil, metrics),
+	}
+	if hasDisplayCol {
+		fields = append(fields, data.NewField(SQLDisplayFieldName, nil, displays))
+	}
+	for _, k := range labelKeys {
+		fields = append(fields, data.NewField(k, nil, labels[k]))
+	}
+
+	out := data.NewFrame("", fields...)
+	out.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+	return data.Frames{out}, nil
+}
+
+func supportedToLongConversion(inputType data.FrameType) bool {
+	switch inputType {
+	case data.FrameTypeNumericMulti, data.FrameTypeNumericWide:
+		return true
+	case data.FrameTypeTimeSeriesMulti, data.FrameTypeTimeSeriesWide:
+		return true
+	default:
+		return false
+	}
+}
+
+func convertNumericMultiToNumericWide(frames data.Frames) data.Frames {
+	if len(frames) == 0 {
+		return nil
+	}
+
+	out := data.NewFrame("")
+	for _, frame := range frames {
+		for _, field := range frame.Fields {
+			if field.Type().Numeric() {
+				out.Fields = append(out.Fields, field)
+			}
+		}
+	}
+	out.Meta = &data.FrameMeta{Type: data.FrameTypeNumericWide}
+	return data.Frames{out}
+}
diff --git a/pkg/expr/convert_to_full_long_num_test.go b/pkg/expr/convert_to_full_long_num_test.go
new file mode 100644
index 00000000000..88a157f89aa
--- /dev/null
+++ b/pkg/expr/convert_to_full_long_num_test.go
@@ -0,0 +1,507 @@
+package expr
+
+import (
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConvertNumericWideToFullLong(t *testing.T) {
+	t.Run("SingleItemNoLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("numeric",
+				data.NewField("cpu", nil, []float64{3.14}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericWide}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(3.14)}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("MultiRowShouldError", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("numeric",
+				data.NewField("cpu", nil, []float64{1.0, 2.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericWide}
+
+		_, err := ConvertToFullLong(input)
+		require.Error(t, err)
+		require.Contains(t, err.Error(), "no more than one row")
+	})
+
+	t.Run("TwoItemsWithSingleLabel", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("numeric",
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0}),
+				data.NewField("cpu", data.Labels{"host": "b"}, []float64{2.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericWide}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(2.0)}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoItemsWithSparseLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("numeric",
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0}),
+				data.NewField("cpu", data.Labels{"host": "b", "env": "prod"}, []float64{2.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericWide}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(2.0)}),
+			data.NewField("env", nil, []*string{nil, sp("prod")}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoDifferentMetricsWithSharedLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("numeric",
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0}),
+				data.NewField("mem", data.Labels{"host": "a"}, []float64{4.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericWide}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "mem"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(4.0)}),
+			data.NewField("host", nil, []*string{sp("a"), sp("a")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoSparseMetricsAndLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("numeric",
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0}),
+				data.NewField("mem", data.Labels{"env": "prod"}, []float64{4.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericWide}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "mem"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(4.0)}),
+			data.NewField("env", nil, []*string{nil, sp("prod")}),
+			data.NewField("host", nil, []*string{sp("a"), nil}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("ExtraTimeFieldIsDropped", func(t *testing.T) {
+		// Note we may consider changing this behavior and looking into keeping
+		// remainder fields in the future.
+		input := data.Frames{
+			data.NewFrame("numeric",
+				data.NewField("timestamp", nil, []time.Time{time.Now()}), // extra time field
+				data.NewField("cpu", nil, []float64{1.23}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericWide}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.23)}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+}
+
+func TestConvertNumericWideToFullLongWithDisplayName(t *testing.T) {
+	t.Run("SingleFieldWithDisplayName", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("numeric",
+				func() *data.Field {
+					f := data.NewField("cpu", nil, []float64{3.14})
+					f.Config = &data.FieldConfig{DisplayNameFromDS: "CPU Display"}
+					return f
+				}(),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericWide}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(3.14)}),
+			data.NewField(SQLDisplayFieldName, nil, []*string{sp("CPU Display")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("MixedDisplayNames", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("numeric",
+				func() *data.Field {
+					f := data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0})
+					f.Config = &data.FieldConfig{DisplayNameFromDS: "CPU A"}
+					return f
+				}(),
+				data.NewField("cpu", data.Labels{"host": "b"}, []float64{2.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericWide}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(2.0)}),
+			data.NewField(SQLDisplayFieldName, nil, []*string{sp("CPU A"), nil}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+}
+
+func TestConvertNumericMultiToFullLong(t *testing.T) {
+	t.Run("SingleItemNoLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				data.NewField("cpu", nil, []float64{3.14}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(3.14)}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoItemsWithSingleLabel", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0}),
+			),
+			data.NewFrame("",
+				data.NewField("cpu", data.Labels{"host": "b"}, []float64{2.0}),
+			),
+		}
+		for _, f := range input {
+			f.Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+		}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(2.0)}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoItemsWithSparseLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0}),
+			),
+			data.NewFrame("",
+				data.NewField("cpu", data.Labels{"host": "b", "env": "prod"}, []float64{2.0}),
+			),
+		}
+		for _, f := range input {
+			f.Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+		}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(2.0)}),
+			data.NewField("env", nil, []*string{nil, sp("prod")}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoDifferentMetricsWithSharedLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0}),
+			),
+			data.NewFrame("",
+				data.NewField("mem", data.Labels{"host": "a"}, []float64{4.0}),
+			),
+		}
+		for _, f := range input {
+			f.Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+		}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "mem"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(4.0)}),
+			data.NewField("host", nil, []*string{sp("a"), sp("a")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoSparseMetricsAndLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0}),
+			),
+			data.NewFrame("",
+				data.NewField("mem", data.Labels{"env": "prod"}, []float64{4.0}),
+			),
+		}
+		for _, f := range input {
+			f.Meta = &data.FrameMeta{Type: data.FrameTypeNumericMulti}
+		}
+
+		expected := data.NewFrame("",
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "mem"}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(4.0)}),
+			data.NewField("env", nil, []*string{nil, sp("prod")}),
+			data.NewField("host", nil, []*string{sp("a"), nil}),
+		)
+		expected.Meta = &data.FrameMeta{Type: numericFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+}
+
+func TestConvertTimeSeriesWideToFullLong(t *testing.T) {
+	times := []time.Time{
+		time.Unix(0, 0),
+		time.Unix(10, 0),
+	}
+
+	t.Run("SingleSeriesNoLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", nil, []float64{1.0, 2.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesWide}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, times),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(2.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu"}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Mismatch (-want +got):\n%s", diff)
+		}
+	})
+
+	t.Run("TwoSeriesOneLabel", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0, 2.0}),
+				data.NewField("cpu", data.Labels{"host": "b"}, []float64{3.0, 4.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesWide}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, []time.Time{times[0], times[0], times[1], times[1]}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(3.0), fp(2.0), fp(4.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu", "cpu", "cpu"}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b"), sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Mismatch (-want +got):\n%s", diff)
+		}
+	})
+
+	t.Run("TwoMetricsWithSharedLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0, 2.0}),
+				data.NewField("mem", data.Labels{"host": "a"}, []float64{3.0, 4.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesWide}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, []time.Time{times[0], times[0], times[1], times[1]}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(3.0), fp(2.0), fp(4.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "mem", "cpu", "mem"}),
+			data.NewField("host", nil, []*string{sp("a"), sp("a"), sp("a"), sp("a")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Mismatch (-want +got):\n%s", diff)
+		}
+	})
+
+	t.Run("TwoSeriesSparseLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0, 2.0}),
+				data.NewField("cpu", data.Labels{"host": "b", "env": "prod"}, []float64{3.0, 4.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesWide}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, []time.Time{times[0], times[0], times[1], times[1]}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(3.0), fp(2.0), fp(4.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu", "cpu", "cpu"}),
+			data.NewField("env", nil, []*string{nil, sp("prod"), nil, sp("prod")}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b"), sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Mismatch (-want +got):\n%s", diff)
+		}
+	})
+
+	t.Run("TwoSeriesSparseMetricsAndLabels", func(t *testing.T) {
+		input := data.Frames{
+			data.NewFrame("",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0, 2.0}),
+				data.NewField("mem", data.Labels{"host": "b", "env": "prod"}, []float64{3.0, 4.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesWide}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, []time.Time{times[0], times[0], times[1], times[1]}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(3.0), fp(2.0), fp(4.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "mem", "cpu", "mem"}),
+			data.NewField("env", nil, []*string{nil, sp("prod"), nil, sp("prod")}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b"), sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Mismatch (-want +got):\n%s", diff)
+		}
+	})
+}
+
+func sp(s string) *string {
+	return &s
+}
diff --git a/pkg/expr/convert_to_full_long_ts_test.go b/pkg/expr/convert_to_full_long_ts_test.go
new file mode 100644
index 00000000000..eb43d9324fe
--- /dev/null
+++ b/pkg/expr/convert_to_full_long_ts_test.go
@@ -0,0 +1,373 @@
+package expr
+
+import (
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConvertTimeSeriesMultiToFullLong(t *testing.T) {
+	t.Run("SingleSeriesNoLabels", func(t *testing.T) {
+		times := []time.Time{
+			time.Unix(0, 0),
+			time.Unix(10, 0),
+			time.Unix(20, 0),
+		}
+		values := []float64{1.0, 2.0, 3.0}
+
+		input := data.Frames{
+			data.NewFrame("cpu",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", nil, values),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, times),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(2.0), fp(3.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu", "cpu"}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoSeriesOneLabel", func(t *testing.T) {
+		times := []time.Time{
+			time.Unix(0, 0),
+			time.Unix(10, 0),
+		}
+
+		input := data.Frames{
+			data.NewFrame("cpu",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0, 2.0}),
+			),
+			data.NewFrame("cpu",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "b"}, []float64{3.0, 4.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+		input[1].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, []time.Time{
+				time.Unix(0, 0), time.Unix(0, 0), time.Unix(10, 0), time.Unix(10, 0),
+			}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(3.0), fp(2.0), fp(4.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu", "cpu", "cpu"}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b"), sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoMetricsWithSharedLabels", func(t *testing.T) {
+		times := []time.Time{
+			time.Unix(0, 0),
+			time.Unix(10, 0),
+		}
+
+		input := data.Frames{
+			data.NewFrame("cpu",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0, 2.0}),
+			),
+			data.NewFrame("mem",
+				data.NewField("time", nil, times),
+				data.NewField("mem", data.Labels{"host": "a"}, []float64{3.0, 4.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+		input[1].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, []time.Time{
+				time.Unix(0, 0), time.Unix(0, 0), time.Unix(10, 0), time.Unix(10, 0),
+			}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(3.0), fp(2.0), fp(4.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "mem", "cpu", "mem"}),
+			data.NewField("host", nil, []*string{sp("a"), sp("a"), sp("a"), sp("a")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoSeriesSparseLabels", func(t *testing.T) {
+		times := []time.Time{
+			time.Unix(0, 0),
+			time.Unix(10, 0),
+		}
+
+		input := data.Frames{
+			data.NewFrame("cpu",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0, 2.0}),
+			),
+			data.NewFrame("cpu",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "b", "env": "prod"}, []float64{3.0, 4.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+		input[1].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, []time.Time{
+				time.Unix(0, 0), time.Unix(0, 0), time.Unix(10, 0), time.Unix(10, 0),
+			}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(3.0), fp(2.0), fp(4.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu", "cpu", "cpu"}),
+			data.NewField("env", nil, []*string{nil, sp("prod"), nil, sp("prod")}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b"), sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoSeriesSparseMetrics", func(t *testing.T) {
+		times := []time.Time{
+			time.Unix(0, 0),
+			time.Unix(10, 0),
+		}
+
+		input := data.Frames{
+			data.NewFrame("cpu",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0, 2.0}),
+			),
+			data.NewFrame("mem",
+				data.NewField("time", nil, times),
+				data.NewField("mem", data.Labels{"host": "b"}, []float64{3.0, 4.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+		input[1].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, []time.Time{
+				time.Unix(0, 0), time.Unix(0, 0), time.Unix(10, 0), time.Unix(10, 0),
+			}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(3.0), fp(2.0), fp(4.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "mem", "cpu", "mem"}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b"), sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("TwoSeriesSparseMetricsAndLabels", func(t *testing.T) {
+		times := []time.Time{
+			time.Unix(0, 0),
+			time.Unix(10, 0),
+		}
+
+		input := data.Frames{
+			data.NewFrame("cpu",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0, 2.0}),
+			),
+			data.NewFrame("mem",
+				data.NewField("time", nil, times),
+				data.NewField("mem", data.Labels{"host": "b", "env": "prod"}, []float64{3.0, 4.0}),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+		input[1].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, []time.Time{
+				time.Unix(0, 0), time.Unix(0, 0), time.Unix(10, 0), time.Unix(10, 0),
+			}),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(3.0), fp(2.0), fp(4.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "mem", "cpu", "mem"}),
+			data.NewField("env", nil, []*string{nil, sp("prod"), nil, sp("prod")}),
+			data.NewField("host", nil, []*string{sp("a"), sp("b"), sp("a"), sp("b")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+
+	t.Run("ThreeSeriesSparseTimeLabelsMetrics", func(t *testing.T) {
+		timesA := []time.Time{
+			time.Unix(0, 0),
+			time.Unix(10, 0),
+		}
+		timesB := []time.Time{
+			time.Unix(5, 0),
+			time.Unix(15, 0),
+		}
+		timesMem := []time.Time{
+			time.Unix(10, 0),
+			time.Unix(30, 0),
+		}
+
+		input := data.Frames{
+			data.NewFrame("cpu",
+				data.NewField("time", nil, timesA),
+				data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0, 2.0}),
+			),
+			data.NewFrame("cpu",
+				data.NewField("time", nil, timesB),
+				data.NewField("cpu", nil, []float64{9.0, 10.0}), // no labels
+			),
+			data.NewFrame("mem",
+				data.NewField("time", nil, timesMem),
+				data.NewField("mem", data.Labels{"host": "b", "env": "prod"}, []float64{3.0, 4.0}),
+			),
+		}
+		for _, f := range input {
+			f.Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+		}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, []time.Time{
+				time.Unix(0, 0),  // cpu a
+				time.Unix(5, 0),  // cpu no label
+				time.Unix(10, 0), // cpu a
+				time.Unix(10, 0), // mem
+				time.Unix(15, 0), // cpu no label
+				time.Unix(30, 0), // mem
+			}),
+			data.NewField(SQLValueFieldName, nil, []*float64{
+				fp(1.0), fp(9.0), fp(2.0), fp(3.0), fp(10.0), fp(4.0),
+			}),
+			data.NewField(SQLMetricFieldName, nil, []string{
+				"cpu", "cpu", "cpu", "mem", "cpu", "mem",
+			}),
+			data.NewField("env", nil, []*string{
+				nil, nil, nil, sp("prod"), nil, sp("prod"),
+			}),
+			data.NewField("host", nil, []*string{
+				sp("a"), nil, sp("a"), sp("b"), nil, sp("b"),
+			}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Result mismatch (-want +got):%s", diff)
+		}
+	})
+}
+
+func TestConvertTimeSeriesMultiToFullLongWithDisplayName(t *testing.T) {
+	t.Run("SingleSeriesWithDisplayName", func(t *testing.T) {
+		times := []time.Time{time.Unix(0, 0), time.Unix(10, 0)}
+
+		input := data.Frames{
+			data.NewFrame("cpu",
+				data.NewField("time", nil, times),
+				func() *data.Field {
+					f := data.NewField("cpu", nil, []float64{1.0, 2.0})
+					f.Config = &data.FieldConfig{DisplayNameFromDS: "CPU Display"}
+					return f
+				}(),
+			),
+		}
+		input[0].Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, times),
+			data.NewField(SQLValueFieldName, nil, []*float64{fp(1.0), fp(2.0)}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu"}),
+			data.NewField(SQLDisplayFieldName, nil, []*string{sp("CPU Display"), sp("CPU Display")}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Mismatch (-want +got):\n%s", diff)
+		}
+	})
+
+	t.Run("TwoSeriesMixedDisplayNames", func(t *testing.T) {
+		times := []time.Time{time.Unix(0, 0), time.Unix(10, 0)}
+
+		input := data.Frames{
+			data.NewFrame("cpu",
+				data.NewField("time", nil, times),
+				func() *data.Field {
+					f := data.NewField("cpu", data.Labels{"host": "a"}, []float64{1.0, 2.0})
+					f.Config = &data.FieldConfig{DisplayNameFromDS: "CPU A"}
+					return f
+				}(),
+			),
+			data.NewFrame("cpu",
+				data.NewField("time", nil, times),
+				data.NewField("cpu", data.Labels{"host": "b"}, []float64{3.0, 4.0}),
+			),
+		}
+		for _, f := range input {
+			f.Meta = &data.FrameMeta{Type: data.FrameTypeTimeSeriesMulti}
+		}
+
+		expected := data.NewFrame("",
+			data.NewField("time", nil, []time.Time{
+				times[0], times[0], times[1], times[1],
+			}),
+			data.NewField(SQLValueFieldName, nil, []*float64{
+				fp(1.0), fp(3.0), fp(2.0), fp(4.0),
+			}),
+			data.NewField(SQLMetricFieldName, nil, []string{"cpu", "cpu", "cpu", "cpu"}),
+			data.NewField(SQLDisplayFieldName, nil, []*string{
+				sp("CPU A"), nil, sp("CPU A"), nil,
+			}),
+			data.NewField("host", nil, []*string{
+				sp("a"), sp("b"), sp("a"), sp("b"),
+			}),
+		)
+		expected.Meta = &data.FrameMeta{Type: timeseriesFullLongType}
+
+		output, err := ConvertToFullLong(input)
+		require.NoError(t, err)
+		require.Len(t, output, 1)
+		if diff := cmp.Diff(expected, output[0], data.FrameTestCompareOptions()...); diff != "" {
+			require.FailNowf(t, "Mismatch (-want +got):\n%s", diff)
+		}
+	})
+}
diff --git a/pkg/expr/convert_to_long.go b/pkg/expr/convert_to_long.go
deleted file mode 100644
index 3ab3339b8b0..00000000000
--- a/pkg/expr/convert_to_long.go
+++ /dev/null
@@ -1,311 +0,0 @@
-package expr
-
-import (
-	"fmt"
-	"sort"
-	"time"
-
-	"github.com/grafana/grafana-plugin-sdk-go/data"
-)
-
-func ConvertToLong(frames data.Frames) (data.Frames, error) {
-	if len(frames) == 0 {
-		// general empty case for now
-		return frames, nil
-	}
-	// Four Conversion Possible Cases
-	// 1. NumericMulti -> NumericLong
-	// 2. NumericWide -> NumericLong
-	// 3. TimeSeriesMulti -> TimeSeriesLong
-	// 4. TimeSeriesWide -> TimeSeriesLong
-
-	// Detect if input type is declared
-	// First Check Frame Meta Type
-
-	var inputType data.FrameType
-	if frames[0].Meta != nil && frames[0].Meta.Type != "" {
-		inputType = frames[0].Meta.Type
-	}
-
-	// TODO: Add some guessing of Type if not declared
-	if inputType == "" {
-		return frames, fmt.Errorf("no input dataframe type set")
-	}
-
-	if !supportedToLongConversion(inputType) {
-		return frames, fmt.Errorf("unsupported input dataframe type %s for SQL expression", inputType)
-	}
-
-	toLong := getToLongConversionFunc(inputType)
-	if toLong == nil {
-		return frames, fmt.Errorf("could not get conversion function for input type %s", inputType)
-	}
-
-	return toLong(frames)
-}
-
-func convertNumericMultiToNumericLong(frames data.Frames) (data.Frames, error) {
-	// Apart from metadata, NumericMulti is basically NumericWide, except one frame per thing
-	// so we collapse into wide and call the wide conversion
-	wide := convertNumericMultiToNumericWide(frames)
-	return convertNumericWideToNumericLong(wide)
-}
-
-func convertNumericMultiToNumericWide(frames data.Frames) data.Frames {
-	newFrame := data.NewFrame("")
-	for _, frame := range frames {
-		for _, field := range frame.Fields {
-			if !field.Type().Numeric() {
-				continue
-			}
-			newField := data.NewFieldFromFieldType(field.Type(), field.Len())
-			newField.Name = field.Name
-			newField.Labels = field.Labels.Copy()
-			if field.Len() == 1 {
-				newField.Set(0, field.CopyAt(0))
-			}
-			newFrame.Fields = append(newFrame.Fields, newField)
-		}
-	}
-	return data.Frames{newFrame}
-}
-
-func convertNumericWideToNumericLong(frames data.Frames) (data.Frames, error) {
-	// Wide should only be one frame
-	if len(frames) != 1 {
-		return nil, fmt.Errorf("expected exactly one frame for wide format, but got %d", len(frames))
-	}
-	inputFrame := frames[0]
-
-	// The Frame should have no more than one row
-	if inputFrame.Rows() > 1 {
-		return nil, fmt.Errorf("expected no more than one row in the frame, but got %d", inputFrame.Rows())
-	}
-
-	// Gather:
-	// - unique numeric Field Names, and
-	// - unique Label Keys (from Numeric Fields only)
-	// each one maps to a field in the output long Frame.
-	uniqueNames := make([]string, 0)
-	uniqueKeys := make([]string, 0)
-
-	uniqueNamesMap := make(map[string]data.FieldType)
-	uniqueKeysMap := make(map[string]struct{})
-
-	prints := make(map[string]int)
-
-	registerPrint := func(labels data.Labels) {
-		fp := labels.Fingerprint().String()
-		if _, ok := prints[fp]; !ok {
-			prints[fp] = len(prints)
-		}
-	}
-
-	for _, field := range inputFrame.Fields {
-		if field.Type().Numeric() {
-			if _, ok := uniqueNamesMap[field.Name]; !ok {
-				uniqueNames = append(uniqueNames, field.Name)
-				uniqueNamesMap[field.Name] = field.Type()
-			}
-
-			if field.Labels != nil {
-				registerPrint(field.Labels)
-				for key := range field.Labels {
-					if _, ok := uniqueKeysMap[key]; !ok {
-						uniqueKeys = append(uniqueKeys, key)
-					}
-					uniqueKeysMap[key] = struct{}{}
-				}
-			}
-		}
-	}
-
-	// Create new fields for output Long frame
-	fields := make([]*data.Field, 0, len(uniqueNames)+len(uniqueKeys))
-
-	// Create the Numeric Fields, tracking the index of each field by name
-	// Note: May want to use FloatAt and and prepopulate with NaN so missing
-	// combinations of value can be NA instead of the zero value of 0.
-	var nameIndexMap = make(map[string]int, len(uniqueNames))
-	for i, name := range uniqueNames {
-		field := data.NewFieldFromFieldType(uniqueNamesMap[name], len(prints))
-		field.Name = name
-		fields = append(fields, field)
-		nameIndexMap[name] = i
-	}
-
-	// Create the String fields, tracking the index of each field by key
-	var keyIndexMap = make(map[string]int, len(uniqueKeys))
-	for i, k := range uniqueKeys {
-		fields = append(fields, data.NewField(k, nil, make([]string, len(prints))))
-		keyIndexMap[k] = len(nameIndexMap) + i
-	}
-
-	longFrame := data.NewFrame("", fields...)
-
-	if inputFrame.Rows() == 0 {
-		return data.Frames{longFrame}, nil
-	}
-
-	// Add Rows to the fields
-	for _, field := range inputFrame.Fields {
-		if !field.Type().Numeric() {
-			continue
-		}
-		fieldIdx := prints[field.Labels.Fingerprint().String()]
-		longFrame.Fields[nameIndexMap[field.Name]].Set(fieldIdx, field.CopyAt(0))
-		for key, value := range field.Labels {
-			longFrame.Fields[keyIndexMap[key]].Set(fieldIdx, value)
-		}
-	}
-
-	return data.Frames{longFrame}, nil
-}
-
-func convertTimeSeriesMultiToTimeSeriesLong(frames data.Frames) (data.Frames, error) {
-	// Collect all time values and ensure no duplicates
-	timeSet := make(map[time.Time]struct{})
-	labelKeys := make(map[string]struct{})     // Collect all unique label keys
-	numericFields := make(map[string]struct{}) // Collect unique numeric field names
-
-	for _, frame := range frames {
-		for _, field := range frame.Fields {
-			if field.Type() == data.FieldTypeTime {
-				for i := 0; i < field.Len(); i++ {
-					t := field.At(i).(time.Time)
-					timeSet[t] = struct{}{}
-				}
-			} else if field.Type().Numeric() {
-				numericFields[field.Name] = struct{}{}
-				if field.Labels != nil {
-					for key := range field.Labels {
-						labelKeys[key] = struct{}{}
-					}
-				}
-			}
-		}
-	}
-
-	// Create a sorted slice of unique time values
-	times := make([]time.Time, 0, len(timeSet))
-	for t := range timeSet {
-		times = append(times, t)
-	}
-	sort.Slice(times, func(i, j int) bool { return times[i].Before(times[j]) })
-
-	// Create output fields: Time, one numeric field per unique numeric name, and label fields
-	timeField := data.NewField("Time", nil, times)
-	outputNumericFields := make(map[string]*data.Field)
-	for name := range numericFields {
-		outputNumericFields[name] = data.NewField(name, nil, make([]float64, len(times)))
-	}
-	outputLabelFields := make(map[string]*data.Field)
-	for key := range labelKeys {
-		outputLabelFields[key] = data.NewField(key, nil, make([]string, len(times)))
-	}
-
-	// Map time to index for quick lookup
-	timeIndexMap := make(map[time.Time]int, len(times))
-	for i, t := range times {
-		timeIndexMap[t] = i
-	}
-
-	// Populate output fields
-	for _, frame := range frames {
-		var timeField *data.Field
-		for _, field := range frame.Fields {
-			if field.Type() == data.FieldTypeTime {
-				timeField = field
-				break
-			}
-		}
-
-		if timeField == nil {
-			return nil, fmt.Errorf("no time field found in frame")
-		}
-
-		for _, field := range frame.Fields {
-			if field.Type().Numeric() {
-				for i := 0; i < field.Len(); i++ {
-					t := timeField.At(i).(time.Time)
-					val, err := field.FloatAt(i)
-					if err != nil {
-						val = 0 // Default value for missing data
-					}
-					idx := timeIndexMap[t]
-					if outputField, exists := outputNumericFields[field.Name]; exists {
-						outputField.Set(idx, val)
-					}
-
-					// Add labels for the numeric field
-					for key, value := range field.Labels {
-						if outputField, exists := outputLabelFields[key]; exists {
-							outputField.Set(idx, value)
-						}
-					}
-				}
-			}
-		}
-	}
-
-	// Build the output frame
-	outputFields := []*data.Field{timeField}
-	for _, field := range outputNumericFields {
-		outputFields = append(outputFields, field)
-	}
-	for _, field := range outputLabelFields {
-		outputFields = append(outputFields, field)
-	}
-	outputFrame := data.NewFrame("time_series_long", outputFields...)
-
-	// Set metadata
-	if outputFrame.Meta == nil {
-		outputFrame.Meta = &data.FrameMeta{}
-	}
-	outputFrame.Meta.Type = data.FrameTypeTimeSeriesLong
-
-	return data.Frames{outputFrame}, nil
-}
-
-func convertTimeSeriesWideToTimeSeriesLong(frames data.Frames) (data.Frames, error) {
-	// Wide should only be one frame
-	if len(frames) != 1 {
-		return nil, fmt.Errorf("expected exactly one frame for wide format, but got %d", len(frames))
-	}
-	inputFrame := frames[0]
-	longFrame, err := data.WideToLong(inputFrame)
-	if err != nil {
-		return nil, fmt.Errorf("failed to convert wide time series to long timeseries for sql expression: %w", err)
-	}
-	return data.Frames{longFrame}, nil
-}
-
-func getToLongConversionFunc(inputType data.FrameType) func(data.Frames) (data.Frames, error) {
-	switch inputType {
-	case data.FrameTypeNumericMulti:
-		return convertNumericMultiToNumericLong
-	case data.FrameTypeNumericWide:
-		return convertNumericWideToNumericLong
-	case data.FrameTypeTimeSeriesMulti:
-		return convertTimeSeriesMultiToTimeSeriesLong
-	case data.FrameTypeTimeSeriesWide:
-		return convertTimeSeriesWideToTimeSeriesLong
-	default:
-		return convertErr
-	}
-}
-
-func convertErr(_ data.Frames) (data.Frames, error) {
-	return nil, fmt.Errorf("unsupported input type for SQL expression")
-}
-
-func supportedToLongConversion(inputType data.FrameType) bool {
-	switch inputType {
-	case data.FrameTypeNumericMulti, data.FrameTypeNumericWide:
-		return true
-	case data.FrameTypeTimeSeriesMulti, data.FrameTypeTimeSeriesWide:
-		return true
-	default:
-		return false
-	}
-}
diff --git a/pkg/expr/convert_to_long_test.go b/pkg/expr/convert_to_long_test.go
deleted file mode 100644
index 291fdb62f17..00000000000
--- a/pkg/expr/convert_to_long_test.go
+++ /dev/null
@@ -1,48 +0,0 @@
-package expr
-
-import (
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-	"github.com/grafana/grafana-plugin-sdk-go/data"
-	"github.com/stretchr/testify/require"
-)
-
-func TestConvertNumericMultiToLong(t *testing.T) {
-	input := data.Frames{
-		data.NewFrame("test",
-			data.NewField("Value", data.Labels{"city": "MIA"}, []int64{5})),
-		data.NewFrame("test",
-			data.NewField("Value", data.Labels{"city": "LGA"}, []int64{7}),
-		),
-	}
-	expectedFrame := data.NewFrame("",
-		data.NewField("Value", nil, []int64{5, 7}),
-		data.NewField("city", nil, []string{"MIA", "LGA"}),
-	)
-	output, err := convertNumericMultiToNumericLong(input)
-	require.NoError(t, err)
-
-	if diff := cmp.Diff(expectedFrame, output[0], data.FrameTestCompareOptions()...); diff != "" {
-		require.FailNowf(t, "Result mismatch (-want +got):%s\n", diff)
-	}
-}
-
-func TestConvertNumericWideToLong(t *testing.T) {
-	input := data.Frames{
-		data.NewFrame("test",
-			data.NewField("Value", data.Labels{"city": "MIA"}, []int64{5}),
-			data.NewField("Value", data.Labels{"city": "LGA"}, []int64{7}),
-		),
-	}
-	expectedFrame := data.NewFrame("",
-		data.NewField("Value", nil, []int64{5, 7}),
-		data.NewField("city", nil, []string{"MIA", "LGA"}),
-	)
-	output, err := convertNumericWideToNumericLong(input)
-	require.NoError(t, err)
-
-	if diff := cmp.Diff(expectedFrame, output[0], data.FrameTestCompareOptions()...); diff != "" {
-		require.FailNowf(t, "Result mismatch (-want +got):%s\n", diff)
-	}
-}
diff --git a/pkg/expr/nodes.go b/pkg/expr/nodes.go
index dea3b10e659..141ce36e3af 100644
--- a/pkg/expr/nodes.go
+++ b/pkg/expr/nodes.go
@@ -429,7 +429,7 @@ func (dn *DSNode) Execute(ctx context.Context, now time.Time, _ mathexp.Vars, s
 		}
 
 		if needsConversion {
-			convertedFrames, err := ConvertToLong(dataFrames)
+			convertedFrames, err := ConvertToFullLong(dataFrames)
 			if err != nil {
 				return result, fmt.Errorf("failed to convert data frames to long format for sql: %w", err)
 			}
diff --git a/pkg/storage/unified/apistore/go.mod b/pkg/storage/unified/apistore/go.mod
index a575dbab08b..2384349a25c 100644
--- a/pkg/storage/unified/apistore/go.mod
+++ b/pkg/storage/unified/apistore/go.mod
@@ -76,7 +76,7 @@ require (
 	github.com/armon/go-metrics v0.4.1 // indirect
 	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
 	github.com/at-wat/mqtt-go v0.19.4 // indirect
-	github.com/aws/aws-sdk-go v1.55.5 // indirect
+	github.com/aws/aws-sdk-go v1.55.6 // indirect
 	github.com/aws/aws-sdk-go-v2 v1.30.3 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.3 // indirect
 	github.com/aws/aws-sdk-go-v2/config v1.27.27 // indirect
diff --git a/pkg/storage/unified/apistore/go.sum b/pkg/storage/unified/apistore/go.sum
index 4d7afcb8d52..3f1f2a91fb6 100644
--- a/pkg/storage/unified/apistore/go.sum
+++ b/pkg/storage/unified/apistore/go.sum
@@ -735,8 +735,8 @@ github.com/at-wat/mqtt-go v0.19.4 h1:R2cbCU7O5PHQ38unbe1Y51ncG3KsFEJV6QeipDoqdLQ
 github.com/at-wat/mqtt-go v0.19.4/go.mod h1:AsiWc9kqVOhqq7LzUeWT/AkKUBfx3Sw5cEe8lc06fqA=
 github.com/aws/aws-sdk-go v1.17.7/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/aws/aws-sdk-go v1.38.35/go.mod h1:hcU610XS61/+aQV88ixoOzUoG7v3b31pl2zKMmprdro=
-github.com/aws/aws-sdk-go v1.55.5 h1:KKUZBfBoyqy5d3swXyiC7Q76ic40rYcbqH7qjh59kzU=
-github.com/aws/aws-sdk-go v1.55.5/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
+github.com/aws/aws-sdk-go v1.55.6 h1:cSg4pvZ3m8dgYcgqB97MrcdjUmZ1BeMYKUxMMB89IPk=
+github.com/aws/aws-sdk-go v1.55.6/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
 github.com/aws/aws-sdk-go-v2 v1.30.3 h1:jUeBtG0Ih+ZIFH0F4UkmL9w3cSpaMv9tYYDbzILP8dY=
 github.com/aws/aws-sdk-go-v2 v1.30.3/go.mod h1:nIQjQVp5sfpQcTc9mPSr1B0PaWK5ByX9MOoDadSN4lc=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.3 h1:tW1/Rkad38LA15X4UQtjXZXNKsCgkshC3EbmcUmghTg=
diff --git a/pkg/storage/unified/resource/go.mod b/pkg/storage/unified/resource/go.mod
index 689d9b2d97b..118beb80e1f 100644
--- a/pkg/storage/unified/resource/go.mod
+++ b/pkg/storage/unified/resource/go.mod
@@ -60,7 +60,7 @@ require (
 	github.com/VividCortex/mysqlerr v0.0.0-20170204212430-6c6b55f8796f // indirect
 	github.com/apache/arrow-go/v18 v18.0.1-0.20241212180703-82be143d7c30 // indirect
 	github.com/armon/go-metrics v0.4.1 // indirect
-	github.com/aws/aws-sdk-go v1.55.5 // indirect
+	github.com/aws/aws-sdk-go v1.55.6 // indirect
 	github.com/aws/aws-sdk-go-v2 v1.30.3 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.3 // indirect
 	github.com/aws/aws-sdk-go-v2/config v1.27.27 // indirect
diff --git a/pkg/storage/unified/resource/go.sum b/pkg/storage/unified/resource/go.sum
index 77e90d5ecf1..fad5380cff8 100644
--- a/pkg/storage/unified/resource/go.sum
+++ b/pkg/storage/unified/resource/go.sum
@@ -714,8 +714,8 @@ github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2/go.mod h1:W
 github.com/at-wat/mqtt-go v0.19.4 h1:R2cbCU7O5PHQ38unbe1Y51ncG3KsFEJV6QeipDoqdLQ=
 github.com/at-wat/mqtt-go v0.19.4/go.mod h1:AsiWc9kqVOhqq7LzUeWT/AkKUBfx3Sw5cEe8lc06fqA=
 github.com/aws/aws-sdk-go v1.17.7/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
-github.com/aws/aws-sdk-go v1.55.5 h1:KKUZBfBoyqy5d3swXyiC7Q76ic40rYcbqH7qjh59kzU=
-github.com/aws/aws-sdk-go v1.55.5/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
+github.com/aws/aws-sdk-go v1.55.6 h1:cSg4pvZ3m8dgYcgqB97MrcdjUmZ1BeMYKUxMMB89IPk=
+github.com/aws/aws-sdk-go v1.55.6/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
 github.com/aws/aws-sdk-go-v2 v1.30.3 h1:jUeBtG0Ih+ZIFH0F4UkmL9w3cSpaMv9tYYDbzILP8dY=
 github.com/aws/aws-sdk-go-v2 v1.30.3/go.mod h1:nIQjQVp5sfpQcTc9mPSr1B0PaWK5ByX9MOoDadSN4lc=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.3 h1:tW1/Rkad38LA15X4UQtjXZXNKsCgkshC3EbmcUmghTg=