PanelQueryRunner: Use rxjs forkJoin (like Scenes), not lodash merge (#109220)

packages/grafana-data/src/internal/index.ts

@@ -26,6 +26,7 @@ export {
   type ConvertFieldTypeTransformerOptions,
   convertFieldType,
 } from '../transformations/transformers/convertFieldType';
+export { type ConvertFrameTypeTransformerOptions, FrameType } from '../transformations/transformers/convertFrameType';
 export { type FilterFieldsByNameTransformerOptions } from '../transformations/transformers/filterByName';
 export { type FilterFramesByRefIdTransformerOptions } from '../transformations/transformers/filterByRefId';
 export { FormatStringOutput, type FormatStringTransformerOptions } from '../transformations/transformers/formatString';

packages/grafana-data/src/transformations/transformers.ts

@@ -1,6 +1,7 @@
 import { calculateFieldTransformer } from './transformers/calculateField';
 import { concatenateTransformer } from './transformers/concat';
 import { convertFieldTypeTransformer } from './transformers/convertFieldType';
+import { convertFrameTypeTransformer } from './transformers/convertFrameType';
 import { ensureColumnsTransformer } from './transformers/ensureColumns';
 import { filterFieldsTransformer, filterFramesTransformer } from './transformers/filter';
 import { filterFieldsByNameTransformer } from './transformers/filterByName';
@@ -53,6 +54,7 @@ export const standardTransformers = {
   renameByRegexTransformer,
   histogramTransformer,
   convertFieldTypeTransformer,
+  convertFrameTypeTransformer,
   groupingToMatrixTransformer,
   limitTransformer,
   groupToNestedTable,

packages/grafana-data/src/transformations/transformers/convertFrameType.test.ts

@@ -0,0 +1,45 @@
+import { DataTransformerConfig } from '@grafana/schema';
+
+import { toDataFrame } from '../../dataframe/processDataFrame';
+import { mockTransformationsRegistry } from '../../internal';
+import { FieldType } from '../../types/dataFrame';
+import { transformDataFrame } from '../transformDataFrame';
+
+import { convertFrameTypeTransformer, ConvertFrameTypeTransformerOptions, FrameType } from './convertFrameType';
+import { DataTransformerID } from './ids';
+
+describe('convert frame type', () => {
+  beforeAll(() => {
+    mockTransformationsRegistry([convertFrameTypeTransformer]);
+  });
+
+  it('will convert a series frame into an exemplar frame', async () => {
+    const seriesFrame = toDataFrame({
+      fields: [
+        { name: 'Time', type: FieldType.time, values: [1000, 2000] },
+        { name: 'Value', type: FieldType.number, values: [1, 100] },
+      ],
+    });
+
+    const cfg: DataTransformerConfig<ConvertFrameTypeTransformerOptions> = {
+      id: DataTransformerID.convertFrameType,
+      options: {
+        targetType: FrameType.Exemplar,
+      },
+    };
+
+    await expect(transformDataFrame([cfg], [seriesFrame])).toEmitValuesWith((received) => {
+      const processed = received[0];
+
+      expect(processed[0]).toEqual({
+        name: 'exemplar',
+        meta: { custom: { resultType: 'exemplar' }, dataTopic: 'annotations' },
+        length: 2,
+        fields: [
+          { config: {}, name: 'Time', type: 'time', values: [1000, 2000] },
+          { config: {}, name: 'Value', type: 'number', values: [1, 100] },
+        ],
+      });
+    });
+  });
+});

packages/grafana-data/src/transformations/transformers/convertFrameType.ts

@@ -0,0 +1,66 @@
+import { map } from 'rxjs/operators';
+
+import { DataTopic } from '@grafana/schema';
+
+import { DataFrame } from '../../types/dataFrame';
+import { DataTransformerInfo } from '../../types/transformations';
+
+import { DataTransformerID } from './ids';
+
+/*
+"schema": {
+  "meta": {
+    "custom": {
+      "resultType": "exemplar"
+    }
+  },
+*/
+
+// ResultType?
+
+export enum FrameType {
+  Exemplar = 'exemplar',
+  TimeRegion = 'timeRegion',
+  Annotation = 'annotation',
+}
+
+export interface ConvertFrameTypeTransformerOptions {
+  targetType?: FrameType;
+}
+
+/** @alpha */
+export const convertFrameTypeTransformer: DataTransformerInfo<ConvertFrameTypeTransformerOptions> = {
+  id: DataTransformerID.convertFrameType,
+  name: 'Convert frame type',
+  description: 'Convert data frame(s) to another type.',
+
+  operator: (options) => (source) =>
+    source.pipe(
+      map((data) => {
+        return convertFrameType(options, data);
+      })
+    ),
+};
+
+function convertFrameType(options: ConvertFrameTypeTransformerOptions, frames: DataFrame[]): DataFrame[] {
+  const { targetType } = options;
+  return targetType === FrameType.Exemplar ? frames.map(convertSeriesToExemplar) : frames;
+}
+
+function convertSeriesToExemplar(frame: DataFrame): DataFrame {
+  // TODO: ensure time field
+  // TODO: ensure value field
+
+  return {
+    ...frame,
+    name: 'exemplar',
+    meta: {
+      ...frame.meta,
+      dataTopic: DataTopic.Annotations,
+      custom: {
+        ...frame.meta?.custom,
+        resultType: 'exemplar',
+      },
+    },
+  };
+}

packages/grafana-data/src/transformations/transformers/ids.ts

@@ -27,6 +27,7 @@ export enum DataTransformerID {
   rowsToFields = 'rowsToFields',
   prepareTimeSeries = 'prepareTimeSeries',
   convertFieldType = 'convertFieldType',
+  convertFrameType = 'convertFrameType',
   fieldLookup = 'fieldLookup',
   heatmap = 'heatmap',
   spatial = 'spatial',

public/app/features/query/state/PanelQueryRunner.test.ts

@@ -5,7 +5,8 @@ import { Subject } from 'rxjs';
 // Importing this way to be able to spy on grafana/data
 
 import * as grafanaData from '@grafana/data';
-import { DataSourceApi, dateTime, TypedVariableModel } from '@grafana/data';
+import { DataSourceApi, DataTransformerID, dateTime, TypedVariableModel } from '@grafana/data';
+import { FrameType, mockTransformationsRegistry } from '@grafana/data/internal';
 import { DataSourceSrv, setDataSourceSrv, setEchoSrv } from '@grafana/runtime';
 import { TemplateSrvMock } from 'app/features/templating/template_srv.mock';
 
@@ -181,6 +182,11 @@ function describeQueryRunnerScenario(
 }
 
 describe('PanelQueryRunner', () => {
+  beforeAll(() => {
+    const { convertFrameTypeTransformer } = grafanaData.standardTransformers;
+    mockTransformationsRegistry([convertFrameTypeTransformer]);
+  });
+
   beforeEach(() => {
     jest.clearAllMocks();
   });
@@ -319,6 +325,48 @@ describe('PanelQueryRunner', () => {
     }
   );
 
+  describeQueryRunnerScenario(
+    'transformations',
+    (ctx) => {
+      it('should re-categorize any anno frames returned by series transformations', async () => {
+        ctx.runner.getData({ withTransforms: true, withFieldConfig: false }).subscribe({
+          next: (data: grafanaData.PanelData) => {
+            try {
+              expect(data.series).toEqual([]);
+              expect(data.annotations).toEqual([
+                {
+                  name: 'exemplar',
+                  meta: { custom: { resultType: 'exemplar' }, dataTopic: 'annotations' },
+                  length: 2,
+                  fields: [
+                    { config: {}, name: 'Time', state: null, type: 'time', values: [1000, 2000] },
+                    { config: {}, name: 'Value', state: null, type: 'number', values: [1, 2] },
+                  ],
+                },
+              ]);
+              return data;
+            } catch (e) {
+              return Promise.reject(e instanceof Error ? e.message : e);
+            }
+          },
+        });
+      });
+    },
+    {
+      getFieldOverrideOptions: () => undefined,
+      getTransformations: () => [
+        {
+          id: DataTransformerID.convertFrameType,
+          topic: grafanaData.DataTopic.Series,
+          options: {
+            targetType: FrameType.Exemplar,
+          },
+        },
+      ],
+      getDataSupport: () => ({ annotations: true, alertStates: false }),
+    }
+  );
+
   describeQueryRunnerScenario(
     'getData',
     (ctx) => {

public/app/features/query/state/PanelQueryRunner.ts

@@ -1,5 +1,5 @@
-import { cloneDeep, merge, isEqual } from 'lodash';
-import { Observable, of, ReplaySubject, Unsubscribable } from 'rxjs';
+import { cloneDeep, isEqual } from 'lodash';
+import { forkJoin, Observable, of, ReplaySubject, Unsubscribable } from 'rxjs';
 import { map, mergeMap, catchError } from 'rxjs/operators';
 
 import {
@@ -237,11 +237,24 @@ export class PanelQueryRunner {
     let seriesStream = transformDataFrame(seriesTransformations, data.series, ctx);
     let annotationsStream = transformDataFrame(annotationsTransformations, data.annotations ?? [], ctx);
 
-    return merge(seriesStream, annotationsStream).pipe(
-      map((frames) => {
-        let isAnnotations = frames.some((f) => f.meta?.dataTopic === DataTopic.Annotations);
-        let transformed = isAnnotations ? { annotations: frames } : { series: frames };
-        return { ...data, ...transformed };
+    let series: DataFrame[] = [];
+    let annotations: DataFrame[] = [];
+
+    return forkJoin([seriesStream, annotationsStream]).pipe(
+      map((results) => {
+        // this strategy allows transformations to take in series frames and produce anno frames
+        // we look at each transformation's result and put it in the correct place
+        results.forEach((frames) => {
+          for (const frame of frames) {
+            if (frame.meta?.dataTopic === DataTopic.Annotations) {
+              annotations.push(frame);
+            } else {
+              series.push(frame);
+            }
+          }
+        });
+
+        return { ...data, series, annotations };
       }),
       catchError((err) => {
         console.warn('Error running transformation:', err);