refactor existsTable

src/clickhouse/handleSinkRequest.ts

@@ -6,10 +6,11 @@ import { Clock, Manifest, PayloadBody } from "../schemas.js";
 import { client, config } from "../config.js";
 const { setTimeout } = require("timers/promises");
 
+// TO-DO: moves these to a separate file `src/clickhouse/stores.ts`
 const knownModuleHashes = new Set<string>();
 const knownBlockId = new Set<string>();
 const knownBlockIdFinal = new Set<string>();
-const existingTables = new Map<string, boolean>();
+const knownTables = new Map<string, boolean>();
 const queue = new PQueue({ concurrency: config.queueConcurrency });
 
 export async function handleSinkRequest({ data, ...metadata }: PayloadBody) {
@@ -51,7 +52,6 @@ function handleModuleHashes(queue: PQueue, manifest: Manifest) {
   }
 }
 
-
 // Final Block Index
 function handleFinalBlocks(queue: PQueue, manifest: Manifest, clock: Clock) {
   const block_id = clock.id;
@@ -99,8 +99,6 @@ async function handleEntityChange(
   change: EntityChange,
   metadata: { clock: Clock; manifest: Manifest }
 ) {
-  // TO-DO: existsTable needs to be refactored to use `client.query()`
-  // Or else should be removed entirely
   const tableExists = await existsTable(change.entity);
 
   let values = getValuesInEntityChange(change);
@@ -142,21 +140,25 @@ function insertEntityChange(
 
   return queue.add(() => client.insert({ values, table, format: "JSONStringsEachRow" }));
 }
-
-// TO-DO: this function won't work in a serverless function environment or running with multiple replicas
-// Cannot depend on memory to know if table exists or not
+// in memory TABLE name cache
+// if true => true
+// if false => false
+// if undefined => check EXISTS if true or false
+// TO-DO when schema TABLE is updated, update knownTables
 async function existsTable(table: string) {
-  if (!existingTables.has(table)) {
-    const response = await client.query({
-      query: "EXISTS " + table,
-      format: "JSONEachRow",
-    });
-    const data = await response.json<Array<{ result: 0 | 1 }>>();
-
-    const foundTable = data[0]?.result === 1;
-    existingTables.set(table, foundTable);
-
-    logger.info(`Found table '${table}': ${foundTable}. Saving data as json: ${!foundTable}`);
-  }
-  return existingTables.get(table)!;
+  // Return cached value if known (reduces number of EXISTS queries)
+  if ( knownTables.has(table) ) return knownTables.get(table);
+
+  // Check if table EXISTS
+  const response = await client.query({
+    query: "EXISTS " + table,
+    format: "JSONEachRow",
+  });
+
+  // handle EXISTS response
+  const data = await response.json<Array<{ result: 0 | 1 }>>();
+  const exists = data[0]?.result === 1;
+  knownTables.set(table, exists);
+  logger.info(`EXISTS [${table}=${exists}]`);
+  return exists;
 }