Merge remote-tracking branch 'duckdb/main' into v1.2-histrionicus

Author: carlopi

lib/include/duckdb/web/webdb.h

@@ -33,6 +33,8 @@ class WebDB {
    public:
     /// A connection
     class Connection {
+        friend WebDB;
+
        protected:
         /// The webdb
         WebDB& webdb_;

lib/src/webdb.cc

@@ -2,6 +2,8 @@
 
 #include "duckdb/web/webdb.h"
 
+#include <emscripten/val.h>
+
 #include <chrono>
 #include <cstddef>
 #include <cstdio>
@@ -783,11 +785,53 @@ std::string WebDB::Tokenize(std::string_view text) {
 /// Get the version
 std::string_view WebDB::GetVersion() { return database_->LibraryVersion(); }
 
+class ProgressBarCustom : public ProgressBarDisplay {
+    double value{0.0};
+    double times{0.0};
+    double to_send{1.0};
+
+   public:
+    ProgressBarCustom() {
+        value = 0.0;
+        times = 0.0;
+        to_send = 1.0;
+    }
+    ~ProgressBarCustom() {}
+    static void SendMessage(double end, double percentage, double times) {
+        emscripten::val::global("DUCKDB_RUNTIME").call<void>("progressUpdate", end ? 1.0 : 0.0, percentage, times);
+    }
+
+   public:
+    void Update(double percentage) {
+        if (percentage >= value + 1.0) {
+            value = percentage;
+            times = 1.0;
+            SendMessage(false, percentage, times);
+            to_send = 10.0;
+        } else {
+            times += 1.0;
+            if (times >= to_send) {
+                SendMessage(false, percentage, times);
+                to_send *= 10.0;
+            }
+        }
+    }
+    void Finish() {
+        SendMessage(true, value, times);
+        value = 0.0;
+        times = 0.0;
+        to_send = 1.0;
+    }
+    static unique_ptr<ProgressBarDisplay> GetProgressBar() { return make_uniq<ProgressBarCustom>(); }
+};
+
 /// Create a session
 WebDB::Connection* WebDB::Connect() {
     auto conn = duckdb::make_uniq<WebDB::Connection>(*this);
     auto conn_ptr = conn.get();
     connections_.insert({conn_ptr, std::move(conn)});
+    ClientConfig::GetConfig(*conn_ptr->connection_.context).wait_time = 1;
+    ClientConfig::GetConfig(*conn_ptr->connection_.context).display_create_func = ProgressBarCustom::GetProgressBar;
     return conn_ptr;
 }
 

packages/duckdb-wasm/src/bindings/connection.ts

@@ -46,7 +46,14 @@ export class DuckDBConnection {
                     resolve(this._bindings.pollPendingQuery(this._conn));
                 } catch (e: any) {
                     console.log(e);
-                    reject(e);
+                    // If the worker is not set, the worker has been terminated
+                    if (e.message.includes('worker is not set!')) {
+                        reject(new Error('Worker has been terminated'));
+                    } else {
+                        //Otherwise, reject with the error
+                        reject(e);
+                    }
+                    
                 }
             });
         }

packages/duckdb-wasm/src/bindings/runtime.ts

@@ -161,6 +161,9 @@ export interface DuckDBRuntime {
     prepareFileHandles?: (path: string[], protocol: DuckDBDataProtocol) => Promise<PreparedDBFileHandle[]>;
     prepareDBFileHandle?: (path: string, protocol: DuckDBDataProtocol) => Promise<PreparedDBFileHandle[]>;
 
+    // Internal API - experimental
+    progressUpdate(final: number, percentage: number, iteration:number): void;
+
     // Call a scalar UDF function
     callScalarUDF(
         mod: DuckDBModule,
@@ -185,6 +188,9 @@ export const DEFAULT_RUNTIME: DuckDBRuntime = {
     getLastFileModificationTime: (_mod: DuckDBModule, _fileId: number): number => {
         return 0;
     },
+    progressUpdate: (_final: number, _percentage: number, _iteration: number): void => {
+        return;
+    },
     truncateFile: (_mod: DuckDBModule, _fileId: number, _newSize: number): void => {},
     readFile: (_mod: DuckDBModule, _fileId: number, _buffer: number, _bytes: number, _location: number): number => {
         return 0;

packages/duckdb-wasm/src/bindings/runtime_browser.ts

@@ -1,4 +1,7 @@
 import {StatusCode} from '../status';
+import {
+    WorkerResponseType,
+} from '../parallel/worker_request';
 import {addS3Headers, getHTTPUrl} from '../utils';
 
 import {
@@ -687,6 +690,11 @@ export const BROWSER_RUNTIME: DuckDBRuntime & {
         }
         return 0;
     },
+    progressUpdate: (done: number, percentage: number, repeat: number): void => {
+	if (postMessage) {
+            postMessage({requestId: 0,  type: WorkerResponseType.PROGRESS_UPDATE,  data: {status: done?"completed":"in-progress", percentage: percentage, repetitions: repeat}});
+	}
+    },
     checkDirectory: (mod: DuckDBModule, pathPtr: number, pathLen: number) => {
         const path = readString(mod, pathPtr, pathLen);
         console.log(`checkDirectory: ${path}`);

packages/duckdb-wasm/src/bindings/runtime_node.ts

@@ -197,6 +197,9 @@ export const NODE_RUNTIME: DuckDBRuntime & {
         }
         return 0;
     },
+    progressUpdate: (_final: number, _percentage: number, _iteration: number): void => {
+        return;
+    },
     getLastFileModificationTime: (mod: DuckDBModule, fileId: number) => {
         try {
             const file = NODE_RUNTIME.resolveFileInfo(mod, fileId);

packages/duckdb-wasm/src/log.ts

@@ -41,6 +41,15 @@ export type LogEntry<O, T, E, V> = {
     readonly value: V;
 };
 
+export type ProgressEntry = {
+    readonly status: string;
+    readonly percentage: string;
+    readonly repetitions: string;
+}
+
+/** An execution progress handler */
+export type ExecutionProgressHandler = (p: ProgressEntry) => void;
+
 export type LogEntryVariant =
     | LogEntry<LogOrigin.BINDINGS, LogTopic.INSTANTIATE, LogEvent.ERROR, string>
     | LogEntry<LogOrigin.BINDINGS, LogTopic.QUERY, LogEvent.START, void>

packages/duckdb-wasm/src/parallel/async_bindings.ts

@@ -18,6 +18,7 @@ import { InstantiationProgress } from '../bindings/progress';
 import { arrowToSQLField } from '../json_typedef';
 import { WebFile } from '../bindings/web_file';
 import { DuckDBDataProtocol } from '../bindings';
+import { ProgressEntry } from '../log';
 
 const TEXT_ENCODER = new TextEncoder();
 
@@ -32,6 +33,9 @@ export class AsyncDuckDB implements AsyncDuckDBBindings {
     /** Instantiate the module */
     protected _onInstantiationProgress: ((p: InstantiationProgress) => void)[] = [];
 
+    /** Progress callbacks */
+    protected _onExecutionProgress: ((p: ProgressEntry) => void)[] = [];
+
     /** The logger */
     protected readonly _logger: Logger;
     /** The worker */
@@ -122,6 +126,12 @@ export class AsyncDuckDB implements AsyncDuckDBBindings {
         const response = event.data as WorkerResponseVariant;
         switch (response.type) {
             // Request failed?
+            case WorkerResponseType.PROGRESS_UPDATE: {
+                for (const p of this._onExecutionProgress) {
+                    p(response.data);
+		}
+                return;
+            }
             case WorkerResponseType.LOG: {
                 this._logger.log(response.data);
                 return;

packages/duckdb-wasm/src/parallel/worker_request.ts

@@ -1,5 +1,5 @@
 import { CSVInsertOptions, JSONInsertOptions, ArrowInsertOptions } from '../bindings/insert_options';
-import { LogEntryVariant } from '../log';
+import { LogEntryVariant, ProgressEntry } from '../log';
 import { ScriptTokens } from '../bindings/tokens';
 import { FileStatistics } from '../bindings/file_stats';
 import { DuckDBConfig } from '../bindings/config';
@@ -57,6 +57,7 @@ export enum WorkerResponseType {
     FILE_STATISTICS = 'FILE_STATISTICS',
     INSTANTIATE_PROGRESS = 'INSTANTIATE_PROGRESS',
     LOG = 'LOG',
+    PROGRESS_UPDATE = 'PROGRESS_UPDATE',
     OK = 'OK',
     PREPARED_STATEMENT_ID = 'PREPARED_STATEMENT_ID',
     QUERY_PLAN = 'QUERY_PLAN',
@@ -154,6 +155,7 @@ export type WorkerResponseVariant =
     | WorkerResponse<WorkerResponseType.FILE_STATISTICS, FileStatistics>
     | WorkerResponse<WorkerResponseType.INSTANTIATE_PROGRESS, InstantiationProgress>
     | WorkerResponse<WorkerResponseType.LOG, LogEntryVariant>
+    | WorkerResponse<WorkerResponseType.PROGRESS_UPDATE, ProgressEntry>
     | WorkerResponse<WorkerResponseType.OK, null>
     | WorkerResponse<WorkerResponseType.PREPARED_STATEMENT_ID, number>
     | WorkerResponse<WorkerResponseType.QUERY_PLAN, Uint8Array>