feat(webui): Create result metadata document in MongoDB to track search signal for Presto queries.

components/webui/client/public/settings.json

@@ -1,6 +1,6 @@
 {
     "ClpStorageEngine": "clp",
-    "ClpQueryEngine": "native",
+    "ClpQueryEngine": "clp",
     "MongoDbSearchResultsMetadataCollectionName": "results-metadata",
     "SqlDbClpArchivesTableName": "clp_archives",
     "SqlDbClpDatasetsTableName": "clp_datasets",

components/webui/client/src/config/index.ts

@@ -1,3 +1,4 @@
+import {CLP_QUERY_ENGINES} from "../../../common/index.js";
 import {settings} from "../settings";
 
 
@@ -9,14 +10,6 @@ enum CLP_STORAGE_ENGINES {
     CLP_S = "clp-s",
 }
 
-/**
- * Query engine options.
- */
-enum CLP_QUERY_ENGINES {
-    NATIVE = "native",
-    PRESTO = "presto",
-}
-
 const SETTINGS_STORAGE_ENGINE = settings.ClpStorageEngine as CLP_STORAGE_ENGINES;
 const SETTINGS_QUERY_ENGINE = settings.ClpQueryEngine as CLP_QUERY_ENGINES;
 

components/webui/client/src/pages/SearchPage/SearchControls/index.tsx

@@ -31,7 +31,7 @@ const SearchControls = () => {
     return (
         <form onSubmit={handleSubmit}>
             <div className={styles["searchControlsContainer"]}>
-                {SETTINGS_QUERY_ENGINE === CLP_QUERY_ENGINES.NATIVE ?
+                {SETTINGS_QUERY_ENGINE !== CLP_QUERY_ENGINES.PRESTO ?
                     (
                         <>
                             {CLP_STORAGE_ENGINES.CLP_S === SETTINGS_STORAGE_ENGINE && <Dataset/>}

components/webui/common/index.ts

@@ -88,6 +88,33 @@ enum SEARCH_SIGNAL {
     RESP_QUERYING = "resp-querying",
 }
 
+/**
+ * Presto search-related signals.
+ *
+ * Note: Using type instead of enum to match `presto-client-node` type definition.
+ */
+type PRESTO_SEARCH_SIGNAL =
+    | "WAITING_FOR_PREREQUISITES"
+    | "QUEUED"
+    | "WAITING_FOR_RESOURCES"
+    | "DISPATCHING"
+    | "PLANNING"
+    | "STARTING"
+    | "RUNNING"
+    | "FINISHING"
+    | "FINISHED"
+    | "CANCELED"
+    | "FAILED";
+
+/**
+ * CLP query engines.
+ */
+enum CLP_QUERY_ENGINES {
+    CLP = "clp",
+    CLP_S = "clp-s",
+    PRESTO = "presto",
+}
+
 /**
  * MongoDB document for search results metadata. `numTotalResults` is optional
  * since it is only set when the search job is completed.
@@ -98,13 +125,16 @@ interface SearchResultsMetadataDocument {
     // eslint-disable-next-line no-warning-comments
     // TODO: Replace with Nullable<string> when the `@common` directory refactoring is completed.
     errorMsg: string | null;
-    lastSignal: SEARCH_SIGNAL;
+    lastSignal: SEARCH_SIGNAL | PRESTO_SEARCH_SIGNAL;
     numTotalResults?: number;
+    queryEngine: CLP_QUERY_ENGINES;
 }
 export {
+    CLP_QUERY_ENGINES,
     SEARCH_SIGNAL,
 };
 export type {
+    PRESTO_SEARCH_SIGNAL,
     SearchResultsMetadataDocument,
     ClientToServerEvents,
     Err,

components/webui/server/settings.json

@@ -18,7 +18,7 @@
     "StreamFilesS3PathPrefix": null,
     "StreamFilesS3Profile": null,
 
-    "ClpQueryEngine": "native",
+    "ClpQueryEngine": "clp",
     "PrestoHost": "localhost",
     "PrestoPort": 8889
 }

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

@@ -1,6 +1,11 @@
 import {FastifyPluginAsyncTypebox} from "@fastify/type-provider-typebox";
 import {StatusCodes} from "http-status-codes";
 
+import {
+    CLP_QUERY_ENGINES,
+    type SearchResultsMetadataDocument,
+} from "../../../../../common/index.js";
+import settings from "../../../../settings.json" with {type: "json"};
 import {ErrorSchema} from "../../../schemas/error.js";
 import {
     PrestoQueryJobCreationSchema,
@@ -28,6 +33,10 @@ const plugin: FastifyPluginAsyncTypebox = async (fastify) => {
         throw new Error("MongoDB database not found");
     }
 
+    const searchResultsMetadataCollection = mongoDb.collection<SearchResultsMetadataDocument>(
+        settings.MongoDbSearchResultsMetadataCollectionName
+    );
+
     /**
      * Submits a search query.
      */
@@ -50,6 +59,7 @@ const plugin: FastifyPluginAsyncTypebox = async (fastify) => {
             let searchJobId: string;
 
             try {
+                // eslint-disable-next-line max-lines-per-function
                 searchJobId = await new Promise<string>((resolve, reject) => {
                     let isResolved = false;
                     Presto.client.execute({
@@ -100,14 +110,32 @@ const plugin: FastifyPluginAsyncTypebox = async (fastify) => {
                                 state: stats.state,
                             }, "Presto search state updated");
 
+                            // Insert metadata and resolve queryId on first call
                             if (false === isResolved) {
+                                searchResultsMetadataCollection.insertOne({
+                                    _id: queryId,
+                                    lastSignal: stats.state,
+                                    errorMsg: null,
+                                    queryEngine: CLP_QUERY_ENGINES.PRESTO,
+                                }).catch((err: unknown) => {
+                                    request.log.error(err, "Failed to insert Presto metadata");
+                                });
                                 isResolved = true;
                                 resolve(queryId);
+                            } else {
+                                // Update metadata on subsequent calls
+                                searchResultsMetadataCollection.updateOne(
+                                    {_id: queryId},
+                                    {$set: {lastSignal: stats.state}}
+                                ).catch((err: unknown) => {
+                                    request.log.error(err, "Failed to update Presto metadata");
+                                });
                             }
                         },
                         success: () => {
                             request.log.info("Presto search succeeded");
                         },
+                        timeout: null,
                     });
                 });
             } catch (error) {

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

@@ -5,6 +5,7 @@ import {
 import {StatusCodes} from "http-status-codes";
 
 import {
+    CLP_QUERY_ENGINES,
     SEARCH_SIGNAL,
     type SearchResultsMetadataDocument,
 } from "../../../../../common/index.js";
@@ -18,7 +19,6 @@ import {QUERY_JOB_TYPE} from "../../../typings/query.js";
 import {SEARCH_MAX_NUM_RESULTS} from "./typings.js";
 import {
     createMongoIndexes,
-    updateSearchResultsMeta,
     updateSearchSignalWhenJobsFinish,
 } from "./utils.js";
 
@@ -44,6 +44,8 @@ const plugin: FastifyPluginAsyncTypebox = async (fastify) => {
         settings.MongoDbSearchResultsMetadataCollectionName
     );
 
+    const queryEngine = settings.ClpQueryEngine as CLP_QUERY_ENGINES;
+
     /**
      * Submits a search query and initiates the search process.
      */
@@ -113,6 +115,7 @@ const plugin: FastifyPluginAsyncTypebox = async (fastify) => {
                 _id: searchJobId.toString(),
                 lastSignal: SEARCH_SIGNAL.RESP_QUERYING,
                 errorMsg: null,
+                queryEngine: queryEngine,
             });
 
             // Defer signal update until after response is sent
@@ -197,16 +200,18 @@ const plugin: FastifyPluginAsyncTypebox = async (fastify) => {
                 await QueryJobDbManager.cancelJob(searchJobId);
                 await QueryJobDbManager.cancelJob(aggregationJobId);
 
-                await updateSearchResultsMeta({
-                    fields: {
-                        lastSignal: SEARCH_SIGNAL.RESP_DONE,
-                        errorMsg: "Query cancelled before it could be completed.",
+                await searchResultsMetadataCollection.updateOne(
+                    {
+                        _id: searchJobId.toString(),
+                        lastSignal: SEARCH_SIGNAL.RESP_QUERYING,
                     },
-                    jobId: searchJobId,
-                    lastSignal: SEARCH_SIGNAL.RESP_QUERYING,
-                    logger: request.log,
-                    searchResultsMetadataCollection: searchResultsMetadataCollection,
-                });
+                    {
+                        $set: {
+                            lastSignal: SEARCH_SIGNAL.RESP_DONE,
+                            errorMsg: "Query cancelled before it could be completed.",
+                        },
+                    }
+                );
             } catch (err: unknown) {
                 const errMsg = "Failed to submit cancel request";
                 request.log.error(

components/webui/server/src/routes/api/search/typings.ts

@@ -7,25 +7,14 @@ import type {
     Db,
 } from "mongodb";
 
-import {
-    SEARCH_SIGNAL,
-    type SearchResultsMetadataDocument,
-} from "../../../../../common/index.js";
+import {type SearchResultsMetadataDocument} from "../../../../../common/index.js";
 
 
 /**
  * The maximum number of results to retrieve for a search.
  */
 const SEARCH_MAX_NUM_RESULTS = 1000;
 
-type UpdateSearchResultsMetaProps = {
-    fields: Partial<SearchResultsMetadataDocument>;
-    jobId: number;
-    lastSignal: SEARCH_SIGNAL;
-    logger: FastifyBaseLogger;
-    searchResultsMetadataCollection: Collection<SearchResultsMetadataDocument>;
-};
-
 type UpdateSearchSignalWhenJobsFinishProps = {
     aggregationJobId: number;
     logger: FastifyBaseLogger;
@@ -46,6 +35,5 @@ export {
     CreateMongoIndexesProps,
     SEARCH_MAX_NUM_RESULTS,
     SearchResultsMetadataDocument,
-    UpdateSearchResultsMetaProps,
     UpdateSearchSignalWhenJobsFinishProps,
 };

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

@@ -4,7 +4,6 @@ import {SEARCH_SIGNAL} from "../../../../../common/index.js";
 import {
     CreateMongoIndexesProps,
     SEARCH_MAX_NUM_RESULTS,
-    UpdateSearchResultsMetaProps,
     UpdateSearchSignalWhenJobsFinishProps,
 } from "./typings.js";
 
@@ -21,36 +20,6 @@ const hasCollection = async (mongoDb: Db, collectionName: string): Promise<boole
     return collections.some((collection: {name: string}) => collection.name === collectionName);
 };
 
-/**
- * Modifies the search results metadata for a given job ID.
- *
- * @param props
- * @param props.fields
- * @param props.jobId
- * @param props.lastSignal
- * @param props.logger
- * @param props.searchResultsMetadataCollection
- */
-const updateSearchResultsMeta = async ({
-    fields,
-    jobId,
-    lastSignal,
-    logger,
-    searchResultsMetadataCollection,
-}: UpdateSearchResultsMetaProps) => {
-    const filter = {
-        _id: jobId.toString(),
-        lastSignal: lastSignal,
-    };
-
-    const modifier = {
-        $set: fields,
-    };
-
-    logger.debug("SearchResultsMetadataCollection modifier = ", modifier);
-    await searchResultsMetadataCollection.updateOne(filter, modifier);
-};
-
 /**
  * Updates the search signal when the specified job finishes.
  *
@@ -103,20 +72,22 @@ const updateSearchSignalWhenJobsFinish = async ({
         return;
     }
 
-    await updateSearchResultsMeta({
-        fields: {
+    const filter = {
+        _id: searchJobId.toString(),
+        lastSignal: SEARCH_SIGNAL.RESP_QUERYING,
+    };
+    const modifier = {
+        $set: {
             lastSignal: SEARCH_SIGNAL.RESP_DONE,
             errorMsg: errorMsg,
             numTotalResults: Math.min(
                 numResultsInCollection,
                 SEARCH_MAX_NUM_RESULTS
             ),
         },
-        jobId: searchJobId,
-        lastSignal: SEARCH_SIGNAL.RESP_QUERYING,
-        logger: logger,
-        searchResultsMetadataCollection: searchResultsMetadataCollection,
-    });
+    };
+
+    await searchResultsMetadataCollection.updateOne(filter, modifier);
 };
 
 /**
@@ -158,6 +129,5 @@ const createMongoIndexes = async ({
 export {
     createMongoIndexes,
     hasCollection,
-    updateSearchResultsMeta,
     updateSearchSignalWhenJobsFinish,
 };