Tempo: Fix exemplar data links when streaming TraceQL metrics (#108078)

* Fix exemplars when streaming metrics

* Fix type

public/app/plugins/datasource/tempo/datasource.ts

@@ -782,6 +782,9 @@ export class TempoDatasource extends DataSourceWithBackend<TempoQuery, TempoJson
         )
       )
     ).pipe(
+      map((response) => {
+        return enhanceTraceQlMetricsResponse(response, this.instanceSettings);
+      }),
       catchError((error) => {
         reportTempoQueryMetrics('grafana_traces_traceql_metrics_response', options, {
           success: false,

public/app/plugins/datasource/tempo/resultTransformer.ts

@@ -1,6 +1,7 @@
 import { SpanStatus } from '@opentelemetry/api';
 import { collectorTypes } from '@opentelemetry/exporter-collector';
 import { SemanticResourceAttributes } from '@opentelemetry/semantic-conventions';
+import { isEqual } from 'lodash';
 
 import {
   createDataFrame,
@@ -474,9 +475,20 @@ export function enhanceTraceQlMetricsResponse(
       const traceIDField = frame.fields.find((field: Field) => field.name === 'traceId');
       if (traceIDField) {
         const links = getDataLinks(instanceSettings);
-        traceIDField.config.links = traceIDField.config.links?.length
-          ? [...traceIDField.config.links, ...links]
-          : links;
+        const existingLinks = traceIDField.config.links || [];
+
+        // Filter out links that already exist
+        const newLinks = links.filter(
+          (link) =>
+            !existingLinks.some(
+              (existing: DataLink) =>
+                existing.title === link.title &&
+                existing.internal?.datasourceUid === link.internal?.datasourceUid &&
+                isEqual(existing.internal?.query, link.internal?.query)
+            )
+        );
+
+        traceIDField.config.links = existingLinks.length ? [...existingLinks, ...newLinks] : newLinks;
       }
       return frame;
     });