feat(webui): Dump query results from Presto into MongoDB. (y-scope#1150)

Author: davemarco

components/webui/server/src/routes/api/presto-search/index.ts

@@ -6,21 +6,28 @@ import {
     PrestoQueryJobCreationSchema,
     PrestoQueryJobSchema,
 } from "../../../schemas/presto-search.js";
+import {insertPrestoRowsToMongo} from "./utils.js";
 
 
 /**
  * Presto search API routes.
  *
  * @param fastify
  */
+// eslint-disable-next-line max-lines-per-function
 const plugin: FastifyPluginAsyncTypebox = async (fastify) => {
-    const {Presto} = fastify;
+    const {Presto, mongo} = fastify;
+    const mongoDb = mongo.db;
 
     if ("undefined" === typeof Presto) {
         // If Presto client is not available, skip the plugin registration.
         return;
     }
 
+    if ("undefined" === typeof mongoDb) {
+        throw new Error("MongoDB database not found");
+    }
+
     /**
      * Submits a search query.
      */
@@ -36,7 +43,7 @@ const plugin: FastifyPluginAsyncTypebox = async (fastify) => {
                 tags: ["Presto Search"],
             },
         },
-
+        // eslint-disable-next-line max-lines-per-function
         async (request, reply) => {
             const {queryString} = request.body;
 
@@ -45,13 +52,39 @@ const plugin: FastifyPluginAsyncTypebox = async (fastify) => {
             try {
                 searchJobId = await new Promise<string>((resolve, reject) => {
                     let isResolved = false;
-
                     Presto.client.execute({
                         // eslint-disable-next-line no-warning-comments
-                        // TODO: Data, error, and success handlers are dummy implementations
+                        // TODO: Error, and success handlers are dummy implementations
                         // and will be replaced with proper implementations.
                         data: (_, data, columns) => {
-                            request.log.info({columns, data}, "Presto data");
+                            request.log.info(
+                                `Received ${data.length} rows from Presto query`
+                            );
+
+                            if (false === isResolved) {
+                                request.log.error(
+                                    "Presto data received before searchJobId was resolved; " +
+                                    "skipping insert."
+                                );
+
+                                return;
+                            }
+
+                            if (0 === data.length) {
+                                return;
+                            }
+
+                            insertPrestoRowsToMongo(
+                                data,
+                                columns,
+                                searchJobId,
+                                mongoDb
+                            ).catch((err: unknown) => {
+                                request.log.error(
+                                    err,
+                                    "Failed to insert Presto results into MongoDB"
+                                );
+                            });
                         },
                         error: (error) => {
                             request.log.info(error, "Presto search failed");

components/webui/server/src/routes/api/presto-search/utils.ts

@@ -0,0 +1,47 @@
+import type {
+    Db,
+    InsertManyResult,
+} from "mongodb";
+
+
+/**
+ * Converts a Presto result row (array of values) into an object, using the provided column
+ * definitions to assign property names.
+ *
+ * @param row Array of values representing a single Presto result row.
+ * @param columns Array of column definitions, each containing a `name` property.
+ * @return An object mapping each column name to its corresponding value from the row.
+ */
+const prestoRowToObject = (
+    row: unknown[],
+    columns: {name: string}[]
+): Record<string, unknown> => {
+    const obj: Record<string, unknown> = {};
+    columns.forEach((col, idx) => {
+        obj[col.name] = row[idx];
+    });
+
+    return obj;
+};
+
+/**
+ * Inserts Presto rows into a MongoDB collection for a given search job.
+ *
+ * @param data Array of Presto result rows
+ * @param columns Array of column definitions
+ * @param searchJobId
+ * @param mongoDb
+ * @return Promise that resolves when the insertion is complete
+ */
+const insertPrestoRowsToMongo = (
+    data: unknown[][],
+    columns: {name: string}[],
+    searchJobId: string,
+    mongoDb: Db
+): Promise<InsertManyResult<Document>> => {
+    const collection = mongoDb.collection(searchJobId);
+    const resultDocs = data.map((row) => prestoRowToObject(row, columns));
+    return collection.insertMany(resultDocs);
+};
+
+export {insertPrestoRowsToMongo};