DataFrame: insert null values along interval (#44622) (#44792)

(cherry picked from commit 3504844ad7)

Co-authored-by: Leon Sorokin <leeoniya@gmail.com>

packages/grafana-data/src/types/dataFrame.ts

@@ -67,6 +67,12 @@ export interface FieldConfig<TOptions = any> {
   min?: number | null;
   max?: number | null;
 
+  // Interval indicates the expected regular step between values in the series.
+  // When an interval exists, consumers can identify "missing" values when the expected value is not present.
+  // The grafana timeseries visualization will render disconnected values when missing values are found it the time field.
+  // The interval uses the same units as the values.  For time.Time, this is defined in milliseconds.
+  interval?: number | null;
+
   // Convert input values into a display string
   mappings?: ValueMapping[];
 

packages/grafana-ui/src/components/GraphNG/nullInsertThreshold.test.ts

@@ -0,0 +1,215 @@
+import { ArrayVector, FieldType, MutableDataFrame } from '@grafana/data';
+import { applyNullInsertThreshold } from './nullInsertThreshold';
+
+function randInt(min: number, max: number) {
+  return Math.floor(Math.random() * (max - min + 1) + min);
+}
+
+function genFrame() {
+  let fieldCount = 10;
+  let valueCount = 3000;
+  let step = 1000;
+  let skipProb = 0.5;
+  let skipSteps = [1, 5]; // min, max
+
+  let allValues = Array(fieldCount);
+
+  allValues[0] = Array(valueCount);
+
+  for (let i = 0, curStep = Date.now(); i < valueCount; i++) {
+    curStep = allValues[0][i] = curStep + step * (Math.random() < skipProb ? randInt(skipSteps[0], skipSteps[1]) : 1);
+  }
+
+  for (let fi = 1; fi < fieldCount; fi++) {
+    let values = Array(valueCount);
+
+    for (let i = 0; i < valueCount; i++) {
+      values[i] = Math.random() * 100;
+    }
+
+    allValues[fi] = values;
+  }
+
+  return {
+    length: valueCount,
+    fields: allValues.map((values, i) => {
+      return {
+        name: 'A-' + i,
+        type: i === 0 ? FieldType.time : FieldType.number,
+        config: {
+          interval: i === 0 ? step : null,
+        },
+        values: new ArrayVector(values),
+      };
+    }),
+  };
+}
+
+describe('nullInsertThreshold Transformer', () => {
+  test('should insert nulls at +threshold between adjacent > threshold: 1', () => {
+    const df = new MutableDataFrame({
+      refId: 'A',
+      fields: [
+        { name: 'Time', type: FieldType.time, values: [1, 3, 10] },
+        { name: 'One', type: FieldType.number, config: { custom: { insertNulls: 1 } }, values: [4, 6, 8] },
+        { name: 'Two', type: FieldType.string, config: { custom: { insertNulls: 1 } }, values: ['a', 'b', 'c'] },
+      ],
+    });
+
+    const result = applyNullInsertThreshold(df);
+
+    expect(result.fields[0].values.toArray()).toStrictEqual([1, 2, 3, 4, 10]);
+    expect(result.fields[1].values.toArray()).toStrictEqual([4, null, 6, null, 8]);
+    expect(result.fields[2].values.toArray()).toStrictEqual(['a', null, 'b', null, 'c']);
+  });
+
+  test('should insert nulls at +threshold between adjacent > threshold: 2', () => {
+    const df = new MutableDataFrame({
+      refId: 'A',
+      fields: [
+        { name: 'Time', type: FieldType.time, values: [5, 7, 11] },
+        { name: 'One', type: FieldType.number, config: { custom: { insertNulls: 2 } }, values: [4, 6, 8] },
+        { name: 'Two', type: FieldType.string, config: { custom: { insertNulls: 2 } }, values: ['a', 'b', 'c'] },
+      ],
+    });
+
+    const result = applyNullInsertThreshold(df);
+
+    expect(result.fields[0].values.toArray()).toStrictEqual([5, 7, 9, 11]);
+    expect(result.fields[1].values.toArray()).toStrictEqual([4, 6, null, 8]);
+    expect(result.fields[2].values.toArray()).toStrictEqual(['a', 'b', null, 'c']);
+  });
+
+  test('should insert nulls at +interval between adjacent > interval: 1', () => {
+    const df = new MutableDataFrame({
+      refId: 'A',
+      fields: [
+        { name: 'Time', type: FieldType.time, config: { interval: 1 }, values: [1, 3, 10] },
+        { name: 'One', type: FieldType.number, values: [4, 6, 8] },
+        { name: 'Two', type: FieldType.string, values: ['a', 'b', 'c'] },
+      ],
+    });
+
+    const result = applyNullInsertThreshold(df);
+
+    expect(result.fields[0].values.toArray()).toStrictEqual([1, 2, 3, 4, 10]);
+    expect(result.fields[1].values.toArray()).toStrictEqual([4, null, 6, null, 8]);
+    expect(result.fields[2].values.toArray()).toStrictEqual(['a', null, 'b', null, 'c']);
+  });
+
+  // TODO: make this work
+  test.skip('should insert nulls at +threshold (when defined) instead of +interval', () => {
+    const df = new MutableDataFrame({
+      refId: 'A',
+      fields: [
+        { name: 'Time', type: FieldType.time, config: { interval: 2 }, values: [5, 7, 11] },
+        { name: 'One', type: FieldType.number, config: { custom: { insertNulls: 1 } }, values: [4, 6, 8] },
+        { name: 'Two', type: FieldType.string, config: { custom: { insertNulls: 1 } }, values: ['a', 'b', 'c'] },
+      ],
+    });
+
+    const result = applyNullInsertThreshold(df);
+
+    expect(result.fields[0].values.toArray()).toStrictEqual([5, 6, 7, 8, 11]);
+    expect(result.fields[1].values.toArray()).toStrictEqual([4, null, 6, null, 8]);
+    expect(result.fields[2].values.toArray()).toStrictEqual(['a', null, 'b', null, 'c']);
+  });
+
+  test('should insert nulls at midpoints between adjacent > interval: 2', () => {
+    const df = new MutableDataFrame({
+      refId: 'A',
+      fields: [
+        { name: 'Time', type: FieldType.time, config: { interval: 2 }, values: [5, 7, 11] },
+        { name: 'One', type: FieldType.number, values: [4, 6, 8] },
+        { name: 'Two', type: FieldType.string, values: ['a', 'b', 'c'] },
+      ],
+    });
+
+    const result = applyNullInsertThreshold(df);
+
+    expect(result.fields[0].values.toArray()).toStrictEqual([5, 7, 9, 11]);
+    expect(result.fields[1].values.toArray()).toStrictEqual([4, 6, null, 8]);
+    expect(result.fields[2].values.toArray()).toStrictEqual(['a', 'b', null, 'c']);
+  });
+
+  test('should noop on fewer than two values', () => {
+    const df = new MutableDataFrame({
+      refId: 'A',
+      fields: [
+        { name: 'Time', type: FieldType.time, config: { interval: 1 }, values: [1] },
+        { name: 'Value', type: FieldType.number, values: [1] },
+      ],
+    });
+
+    const result = applyNullInsertThreshold(df);
+
+    expect(result).toBe(df);
+  });
+
+  test('should noop on invalid threshold', () => {
+    const df = new MutableDataFrame({
+      refId: 'A',
+      fields: [
+        { name: 'Time', type: FieldType.time, values: [1, 2, 4] },
+        { name: 'Value', type: FieldType.number, config: { custom: { insertNulls: -1 } }, values: [1, 1, 1] },
+      ],
+    });
+
+    const result = applyNullInsertThreshold(df);
+
+    expect(result).toBe(df);
+  });
+
+  test('should noop on invalid interval', () => {
+    const df = new MutableDataFrame({
+      refId: 'A',
+      fields: [
+        { name: 'Time', type: FieldType.time, config: { interval: -1 }, values: [1, 2, 4] },
+        { name: 'Value', type: FieldType.number, values: [1, 1, 1] },
+      ],
+    });
+
+    const result = applyNullInsertThreshold(df);
+
+    expect(result).toBe(df);
+  });
+
+  test('should noop when no missing steps', () => {
+    const df = new MutableDataFrame({
+      refId: 'A',
+      fields: [
+        { name: 'Time', type: FieldType.time, config: { interval: 1 }, values: [1, 2, 3] },
+        { name: 'Value', type: FieldType.number, values: [1, 1, 1] },
+      ],
+    });
+
+    const result = applyNullInsertThreshold(df);
+
+    expect(result).toBe(df);
+  });
+
+  test('should noop when refFieldName not found', () => {
+    const df = new MutableDataFrame({
+      refId: 'A',
+      fields: [
+        { name: 'Time', type: FieldType.time, config: { interval: 1 }, values: [1, 2, 5] },
+        { name: 'Value', type: FieldType.number, values: [1, 1, 1] },
+      ],
+    });
+
+    const result = applyNullInsertThreshold(df, 'Time2');
+
+    expect(result).toBe(df);
+  });
+
+  test('perf stress test should be <= 10ms', () => {
+    // 10 fields x 3,000 values with 50% skip (output = 10 fields x 6,000 values)
+    let bigFrameA = genFrame();
+
+    // eslint-disable-next-line no-console
+    console.time('insertValues-10x3k');
+    applyNullInsertThreshold(bigFrameA);
+    // eslint-disable-next-line no-console
+    console.timeEnd('insertValues-10x3k');
+  });
+});

packages/grafana-ui/src/components/GraphNG/nullInsertThreshold.ts

@@ -0,0 +1,115 @@
+import { ArrayVector, DataFrame, FieldType } from '@grafana/data';
+
+type InsertMode = (prev: number, next: number, threshold: number) => number;
+
+const INSERT_MODES = {
+  threshold: (prev: number, next: number, threshold: number) => prev + threshold,
+  midpoint: (prev: number, next: number, threshold: number) => (prev + next) / 2,
+  // previous time + 1ms to prevent StateTimeline from forward-interpolating prior state
+  plusone: (prev: number, next: number, threshold: number) => prev + 1,
+};
+
+export function applyNullInsertThreshold(
+  frame: DataFrame,
+  refFieldName?: string | null,
+  insertMode: InsertMode = INSERT_MODES.threshold
+): DataFrame {
+  if (frame.length < 2) {
+    return frame;
+  }
+
+  const refField = frame.fields.find((field) => {
+    // note: getFieldDisplayName() would require full DF[]
+    return refFieldName != null ? field.name === refFieldName : field.type === FieldType.time;
+  });
+
+  if (refField == null) {
+    return frame;
+  }
+
+  const thresholds = frame.fields.map((field) => field.config.custom?.insertNulls ?? refField.config.interval ?? null);
+
+  const uniqueThresholds = new Set<number>(thresholds);
+
+  uniqueThresholds.delete(null as any);
+
+  if (uniqueThresholds.size === 0) {
+    return frame;
+  }
+
+  if (uniqueThresholds.size === 1) {
+    const threshold = uniqueThresholds.values().next().value;
+
+    if (threshold <= 0) {
+      return frame;
+    }
+
+    const refValues = refField.values.toArray();
+
+    const frameValues = frame.fields.map((field) => field.values.toArray());
+
+    const filledFieldValues = nullInsertThreshold(refValues, frameValues, threshold, insertMode);
+
+    if (filledFieldValues === frameValues) {
+      return frame;
+    }
+
+    return {
+      ...frame,
+      length: filledFieldValues[0].length,
+      fields: frame.fields.map((field, i) => ({
+        ...field,
+        values: new ArrayVector(filledFieldValues[i]),
+      })),
+    };
+  }
+
+  // TODO: unique threshold-per-field (via overrides) is unimplemented
+  // should be done by processing each (refField + thresholdA-field1 + thresholdA-field2...)
+  // as a separate nullInsertThreshold() dataset, then re-join into single dataset via join()
+  return frame;
+}
+
+function nullInsertThreshold(refValues: number[], frameValues: any[][], threshold: number, getInsertValue: InsertMode) {
+  const len = refValues.length;
+  let prevValue: number = refValues[0];
+  const refValuesNew: number[] = [prevValue];
+
+  for (let i = 1; i < len; i++) {
+    const curValue = refValues[i];
+
+    if (curValue - prevValue > threshold) {
+      refValuesNew.push(getInsertValue(prevValue, curValue, threshold));
+    }
+
+    refValuesNew.push(curValue);
+
+    prevValue = curValue;
+  }
+
+  const filledLen = refValuesNew.length;
+
+  if (filledLen === len) {
+    return frameValues;
+  }
+
+  const filledFieldValues: any[][] = [];
+
+  for (let fieldValues of frameValues) {
+    let filledValues;
+
+    if (fieldValues !== refValues) {
+      filledValues = Array(filledLen);
+
+      for (let i = 0, j = 0; i < filledLen; i++) {
+        filledValues[i] = refValues[j] === refValuesNew[i] ? fieldValues[j++] : null;
+      }
+    } else {
+      filledValues = refValuesNew;
+    }
+
+    filledFieldValues.push(filledValues);
+  }
+
+  return filledFieldValues;
+}

packages/grafana-ui/src/components/GraphNG/utils.ts

@@ -1,6 +1,7 @@
 import { XYFieldMatchers } from './types';
 import { ArrayVector, DataFrame, FieldConfig, FieldType, outerJoinDataFrames } from '@grafana/data';
 import { nullToUndefThreshold } from './nullToUndefThreshold';
+import { applyNullInsertThreshold } from './nullInsertThreshold';
 import { AxisPlacement, GraphFieldConfig, ScaleDistribution, ScaleDistributionConfig } from '@grafana/schema';
 import { FIXED_UNIT } from './GraphNG';
 
@@ -32,7 +33,7 @@ function applySpanNullsThresholds(frame: DataFrame) {
 
 export function preparePlotFrame(frames: DataFrame[], dimFields: XYFieldMatchers) {
   let alignedFrame = outerJoinDataFrames({
-    frames: frames,
+    frames: frames.map((frame) => applyNullInsertThreshold(frame)),
     joinBy: dimFields.x,
     keep: dimFields.y,
     keepOriginIndices: true,

packages/grafana-ui/src/components/Sparkline/utils.ts

@@ -1,5 +1,6 @@
 import { DataFrame, FieldConfig, FieldSparkline, IndexVector } from '@grafana/data';
 import { GraphFieldConfig } from '@grafana/schema';
+import { applyNullInsertThreshold } from '../GraphNG/nullInsertThreshold';
 
 /** @internal
  * Given a sparkline config returns a DataFrame ready to be turned into Plot data set
@@ -11,7 +12,7 @@ export function preparePlotFrame(sparkline: FieldSparkline, config?: FieldConfig
     ...config,
   };
 
-  return {
+  return applyNullInsertThreshold({
     refId: 'sparkline',
     fields: [
       sparkline.x ?? IndexVector.newField(length),
@@ -21,5 +22,5 @@ export function preparePlotFrame(sparkline: FieldSparkline, config?: FieldConfig
       },
     ],
     length,
-  };
+  });
 }