feat($locf): Implement last observation carry forward operator. Closes …

…#271

src/operators/window/index.ts

@@ -3,5 +3,6 @@ export * from "./derivative";
 export * from "./documentNumber";
 export * from "./expMovingAvg";
 export * from "./integral";
+export * from "./locf";
 export * from "./rank";
 export * from "./shift";

src/operators/window/locf.ts

@@ -0,0 +1,32 @@
+import { ComputeOptions, computeValue, Options } from "../../core";
+import { AnyVal, RawObject } from "../../types";
+import { WindowOperatorInput } from "../pipeline/_internal";
+
+/**
+ * Last observation carried forward. Sets values for null and missing fields in a window to the last non-null value for the field.
+ */
+export function $locf(
+  obj: RawObject,
+  collection: RawObject[],
+  expr: WindowOperatorInput,
+  options?: Options
+): AnyVal {
+  let lastObserved = computeValue(
+    obj,
+    expr.inputExpr,
+    null,
+    ComputeOptions.init(options, obj)
+  );
+  if (lastObserved === undefined && expr.documentNumber > 1) {
+    const previous = collection[expr.documentNumber - 2];
+    lastObserved = computeValue(
+      previous,
+      expr.inputExpr,
+      null,
+      ComputeOptions.init(options, previous)
+    );
+  }
+  // TODO: consider using a temporary ramdom field to store observations.
+  obj[expr.field] = lastObserved;
+  return lastObserved;
+}

test/operators/window/locf.test.ts

@@ -0,0 +1,76 @@
+import "../../../src/init/system";
+
+import { aggregate } from "../../../src";
+import { ProcessingMode } from "../../../src/core";
+
+const options = { processingMode: ProcessingMode.CLONE_INPUT };
+
+describe("operators/window/locf", () => {
+  describe("$locf", () => {
+    it("Fill Missing Values with the Last Observed Value", () => {
+      const result = aggregate(
+        [
+          {
+            time: new Date("2021-03-08T09:00:00.000Z"),
+            price: 500,
+          },
+          {
+            time: new Date("2021-03-08T10:00:00.000Z"),
+          },
+          {
+            time: new Date("2021-03-08T11:00:00.000Z"),
+            price: 515,
+          },
+          {
+            time: new Date("2021-03-08T12:00:00.000Z"),
+          },
+          {
+            time: new Date("2021-03-08T13:00:00.000Z"),
+          },
+          {
+            time: new Date("2021-03-08T14:00:00.000Z"),
+            price: 485,
+          },
+        ],
+        [
+          {
+            $setWindowFields: {
+              sortBy: { time: 1 },
+              output: {
+                price: { $locf: "$price" },
+              },
+            },
+          },
+        ],
+        options
+      );
+
+      expect(result).toEqual([
+        {
+          time: new Date("2021-03-08T09:00:00.000Z"),
+          price: 500,
+        },
+        {
+          time: new Date("2021-03-08T10:00:00.000Z"),
+          price: 500,
+        },
+        {
+          time: new Date("2021-03-08T11:00:00.000Z"),
+          price: 515,
+        },
+        {
+          time: new Date("2021-03-08T12:00:00.000Z"),
+          price: 515,
+        },
+        {
+          time: new Date("2021-03-08T13:00:00.000Z"),
+          price: 515,
+        },
+        {
+          time: new Date("2021-03-08T14:00:00.000Z"),
+          price: 485,
+        },
+      ]);
+    });
+  });
+});