add formatting; minor fixes/adjustments

packages/analytics-engine/src/engine.ts

@@ -1,7 +1,11 @@
 import { TextDecoder } from "util";
 import type { SqliteDB } from "@miniflare/shared";
 import analytics from "./analytics";
-import buildSQLFunctions from "./functions";
+import buildSQLFunctions, { isDate } from "./functions";
+
+export type Format = "JSON" | "JSONEachRow" | "TabSeparated";
+
+export type MetaType = "DateTime" | "String" | "Float64";
 
 export interface AnalyticsEngineEvent {
  readonly doubles?: number[]; // up to 20
@@ -11,7 +15,7 @@ export interface AnalyticsEngineEvent {
 
 interface DataPoint {
  dataset: string;
- index1: string;
+ index1?: string;
  double1?: number;
  double2?: number;
  double3?: number;
@@ -54,7 +58,15 @@ interface DataPoint {
  blob20?: string | null;
 }
 
-export const kQuery = Symbol("kQuery");
+export interface ResponseData {
+ [key: string]: number | string;
+}
+
+export interface FormatJSON {
+ meta: { [key: string]: MetaType };
+ data: ResponseData[];
+ rows: number;
+}
 
 export class AnalyticsEngine {
  readonly #dataset: string;
@@ -126,7 +138,7 @@ export class AnalyticsEngine {
  blobsValues.push(`@${key}`);
  });
 
- const input = _prepare(
+ const [input] = _prepare(
  `INSERT INTO ${this.#dataset} (dataset, index1${
  doublesKeys.length > 0 ? `, ${doublesKeys}` : ""
  }${
@@ -139,16 +151,10 @@ export class AnalyticsEngine {
 
  insert.run(insertData);
  }
-
- async [kQuery](input: string): Promise<any> {
- const query = this.#db.prepare(_prepare(input));
-
- return query.get();
- }
 }
 
 /** @internal */
-export function _prepare(input: string): string {
+export function _prepare(input: string): [string, Format | undefined] {
  // split
  const pieces = input
  .replaceAll("\n", " ") // convert new lines to spaces
@@ -160,13 +166,16 @@ export function _prepare(input: string): string {
  // find all instances of "INTERVAL" and "QUANTILEWEIGHTED"
  const intervalIndexes = [];
  const quantileweigthedIndexes = [];
+ let formatIndex = -1;
  for (let i = 0, pl = pieces.length; i < pl; i++) {
- if (pieces[i].toLocaleLowerCase() === "interval") {
+ const piece = pieces[i].toLocaleLowerCase();
+ if (piece === "interval") {
  intervalIndexes.push(i);
  }
- if (pieces[i].toLocaleLowerCase().includes("quantileweighted")) {
+ if (piece.includes("quantileweighted")) {
  quantileweigthedIndexes.push(i);
  }
+ if (piece === "format") formatIndex = i;
  }
 
  // for each instance, convert "INTERVAL X Y" to "INTERVAL(X, Y)"
@@ -182,9 +191,84 @@ export function _prepare(input: string): string {
 
  // for each instance of quantileweighted, seperately aggregate columns;
  for (const qwIndex of quantileweigthedIndexes) {
+ // What if "quantileweighted (", the space between the two could cause an error
+ // adjust to "quantileweighted("
+ if (pieces[qwIndex + 1] === "(") {
+ pieces.splice(qwIndex + 1, 1);
+ pieces[qwIndex] = `${pieces[qwIndex]}(`;
+ }
  pieces[qwIndex + 3] = `__GET_QUANTILE_GROUP(${pieces[qwIndex + 3]})`;
  pieces[qwIndex + 5] = `__GET_QUANTILE_GROUP(${pieces[qwIndex + 5]})`;
  }
 
- return pieces.join(" ");
+ // if FORMAT exists, grab type and remove it
+ let formatType: Format = "JSON";
+ if (formatIndex >= 0) {
+ // change to new type. revert to JSON if unknown name
+ formatType = pieces[formatIndex + 1] as Format;
+ if (
+ formatType !== "JSON" &&
+ formatType !== "JSONEachRow" &&
+ formatType !== "TabSeparated"
+ ) {
+ formatType = "JSON";
+ }
+ // remove from string
+ pieces.splice(formatIndex, 1);
+ pieces.splice(formatIndex, 1);
+ }
+
+ return [pieces.join(" "), formatType];
+}
+
+/** @internal */
+export function _format(
+ data: ResponseData[] = [],
+ format: Format = "JSON"
+): string | FormatJSON {
+ if (format === "JSON") return _formatJSON(data);
+ else if (format === "JSONEachRow") return _formatJSONEachRow(data);
+ else return _formatTabSeparated(data);
+}
+
+function _formatJSON(data: ResponseData[]): FormatJSON {
+ const meta: { [key: string]: MetaType } = {};
+ // incase one of the data points might have a null value but another might not
+ for (const point of data) {
+ for (const [key, value] of Object.entries(point)) {
+ if (value !== null) meta[key] = _getType(value);
+ }
+ }
+
+ return {
+ meta,
+ data,
+ rows: data.length,
+ };
+}
+
+function _formatJSONEachRow(data: ResponseData[]): string {
+ let res = "";
+
+ for (const point of data) {
+ res += `${JSON.stringify(point)}\n`;
+ }
+
+ return res;
+}
+
+function _formatTabSeparated(data: ResponseData[]): string {
+ let res = "";
+
+ for (const point of data) {
+ res += `${Object.values(point).join("\t")}\n`;
+ }
+
+ return res;
+}
+
+function _getType(input: number | string): MetaType {
+ if (typeof input === "number") return "Float64";
+ if (isDate(input)) return "DateTime";
+ return "String";
 }

packages/analytics-engine/src/plugin.ts

@@ -9,7 +9,7 @@ import {
  resolveStoragePersist,
 } from "@miniflare/shared";
 import type { SqliteDB } from "@miniflare/shared";
-import { AnalyticsEngine } from "./engine";
+import { AnalyticsEngine, FormatJSON, _prepare, _format } from "./engine";
 
 export type ProcessedAnalyticsEngine = Record<string, string>; // { [name]: dataset }
 
@@ -88,6 +88,17 @@ export class AnalyticsEnginePlugin
  return this.#db;
  }
 
+ async query(
+ storageFactory: StorageFactory,
+ input: string
+ ): Promise<string | FormatJSON> {
+ await this.#setup(storageFactory);
+ const [query, format] = _prepare(input);
+ // @ts-expect-error: #setup already ensures #db exists.
+ const data = this.#db.prepare(query).all();
+ return _format(data, format);
+ }
+
  async #setup(storageFactory: StorageFactory): Promise<void> {
  if (this.#db === undefined) {
  // grab storage

packages/analytics-engine/test/engine.spec.ts

@@ -1,4 +1,8 @@
-import { AnalyticsEngine } from "@miniflare/analytics-engine";
+import {
+ AnalyticsEngine,
+ _prepare,
+ _format,
+} from "@miniflare/analytics-engine";
 import { Storage } from "@miniflare/shared";
 import { testClock } from "@miniflare/shared-test";
 import { MemoryStorage } from "@miniflare/storage-memory";
@@ -376,3 +380,98 @@ test("Analytics Engine: More than 50kB of blob data fails.", async (t) => {
  }
  );
 });
+
+test("Analytics Engine: Test FORMAT JSON.", async (t) => {
+ const { db, storage } = t.context;
+ // @ts-expect-error: protected but does exist
+ const { sqliteDB } = storage;
+
+ await db.writeDataPoint({
+ indexes: ["formatJSON"], // Sensor ID
+ blobs: ["input"],
+ doubles: [0, 3],
+ });
+
+ const [query, format] = _prepare(
+ "SELECT double2 AS answer FROM TEST_DATASET WHERE index1 = ? FORMAT JSON"
+ );
+
+ const stmt = sqliteDB.prepare(query);
+ const res = stmt.all("formatJSON");
+ t.deepEqual(res, [{ answer: 3 }]);
+ t.is(format, "JSON");
+
+ // JSON format
+ const formatedRes = _format(res, "JSON");
+ t.deepEqual(formatedRes, {
+ meta: { answer: "Float64" },
+ data: [{ answer: 3 }],
+ rows: 1,
+ });
+});
+
+test("Analytics Engine: Test FORMAT JSONEachRow.", async (t) => {
+ const { db, storage } = t.context;
+ // @ts-expect-error: protected but does exist
+ const { sqliteDB } = storage;
+
+ await db.writeDataPoint({
+ indexes: ["formatJSONEachRow"], // Sensor ID
+ blobs: ["input"],
+ doubles: [0, 1],
+ });
+ await db.writeDataPoint({
+ indexes: ["formatJSONEachRow"], // Sensor ID
+ blobs: ["input"],
+ doubles: [2, 3],
+ });
+
+ const [query, format] = _prepare(
+ "SELECT double1, double2 FROM TEST_DATASET WHERE index1 = ? FORMAT JSONEachRow"
+ );
+
+ const stmt = sqliteDB.prepare(query);
+ const res = stmt.all("formatJSONEachRow");
+ t.deepEqual(res, [
+ { double1: 0, double2: 1 },
+ { double1: 2, double2: 3 },
+ ]);
+ t.is(format, "JSONEachRow");
+
+ // JSON format
+ const formatedRes = _format(res, "JSONEachRow");
+ t.is(formatedRes, `{"double1":0,"double2":1}\n{"double1":2,"double2":3}\n`);
+});
+
+test("Analytics Engine: Test FORMAT TabSeparated.", async (t) => {
+ const { db, storage } = t.context;
+ // @ts-expect-error: protected but does exist
+ const { sqliteDB } = storage;
+
+ await db.writeDataPoint({
+ indexes: ["formatTabSeparated"], // Sensor ID
+ blobs: ["input"],
+ doubles: [0, 1],
+ });
+ await db.writeDataPoint({
+ indexes: ["formatTabSeparated"], // Sensor ID
+ blobs: ["input"],
+ doubles: [2, 3],
+ });
+
+ const [query, format] = _prepare(
+ "SELECT double1, double2 FROM TEST_DATASET WHERE index1 = ? FORMAT TabSeparated"
+ );
+
+ const stmt = sqliteDB.prepare(query);
+ const res = stmt.all("formatTabSeparated");
+ t.deepEqual(res, [
+ { double1: 0, double2: 1 },
+ { double1: 2, double2: 3 },
+ ]);
+ t.is(format, "TabSeparated");
+
+ // JSON format
+ const formatedRes = _format(res, "TabSeparated");
+ t.is(formatedRes, `0\t1\n2\t3\n`);
+});