Updates to connector to fix concurrency

src/duckduckapi.ts

@@ -68,8 +68,9 @@ const connector: Connector<Configuration, State> = {
    * @param metrics
    */
   tryInitState(_: Configuration, __: unknown): Promise<State> {
-    const credentials: CredentialSchema = { url: DUCKDB_URL };
-    const client = new duckdb.Database(credentials.url, DUCKDB_CONFIG);
+    // const credentials: CredentialSchema = { url: DUCKDB_URL };
+    // const client = new duckdb.Database(credentials.url, DUCKDB_CONFIG);
+    const client = db;
     return Promise.resolve({ client: client });
   },
 
@@ -207,7 +208,7 @@ async function createDuckDBFile(schema: string): Promise<void> {
 
 export interface duckduckapi {
   dbSchema: string
-  loaderJob(): void;
+  loaderJob(db: duckdb.Database): void;
   getFunctions(): void;
 }
 
@@ -218,6 +219,6 @@ export async function makeConnector(dda: duckduckapi): Promise<Connector<Configu
   */
   await createDuckDBFile(dda.dbSchema);
   // spawn loaderjob execution on a cron
-  dda.loaderJob();
+  dda.loaderJob(db);
   return Promise.resolve(connector);
 }

src/handlers/query.ts

@@ -397,8 +397,7 @@ export async function plan_queries(
   if (configuration.config === null || configuration.config === undefined) {
     throw new Forbidden("Connector is not properly configured", {});
   }
-  let collection_alias: string =
-    configuration.config.collection_aliases[query.collection];
+  let collection_alias: string = configuration.config.collection_aliases[query.collection];
   let query_plan: SQLQuery[];
   if (query.variables) {
     let promises = query.variables.map((varSet) => {
@@ -443,6 +442,7 @@ export async function plan_queries(
 
 async function do_all(con: any, query: SQLQuery): Promise<any[]> {
   return new Promise((resolve, reject) => {
+
     con.all(query.sql, ...query.args, function (err: any, res: any) {
       if (err) {
         reject(err);
@@ -464,6 +464,7 @@ async function perform_query(
     const row_set = JSON.parse(res[0]["data"] as string) as RowSet;
     response.push(row_set);
   }
+  con.close();
   return response;
 }
 
@@ -483,8 +484,6 @@ export async function do_query(
   state: State,
   query: QueryRequest
 ): Promise<QueryResponse> {
-  console.log(query);
-
   if (is_query_function(query, configuration)) {
     return await perform_query_function(state, query);
   } else {

src/index.ts

@@ -1,9 +1,10 @@
 import {start} from "@hasura/ndc-sdk-typescript";
 import { makeConnector, duckduckapi, db} from "./duckduckapi";
-
+import * as duckdb from "duckdb";
 // get any api credentials from env vars
 
 const calendar: duckduckapi = {
+  // DROP TABLE IF EXISTS users;
   dbSchema: `
     CREATE TABLE IF NOT EXISTS users (id int, name string);
   `,
@@ -13,19 +14,21 @@ const calendar: duckduckapi = {
       // Implementation for retrieving functions
   },
 
-  loaderJob: () => {
+  loaderJob: (db: duckdb.Database) => {
     console.log("Running loader job...");
-
     async function insertLoop() {
         let id = 100;
         while (true) {
+          const con = db.connect();
           // Insert a row into the table
-          await db.all(`
+          con.exec('BEGIN TRANSACTION');
+          con.all(`
             INSERT INTO users (id, name) values (?, ?);
           `, id, 'name'+id.toString(), (err)=>{console.log(err)});
+          con.run('COMMIT');
           id++;
           console.log(`Inserted row ${id}`);
-
+          con.close();
           // Wait for 1 second before the next insertion
           await new Promise(resolve => setTimeout(resolve, 1000));
         }