Import Data from DuckDB Database File with Thread Workers

R2026b

This example shows how to use a thread pool to efficiently import data from a DuckDB™ database table.

With Parallel Computing Toolbox™, you can create a thread pool by calling the parpool (Parallel Computing Toolbox) function and run database operations on multiple workers in parallel. The number of workers depends on your pool settings and available computing resources.

This example requires Parallel Computing Toolbox™.

pool = parpool("Threads");

Starting parallel pool (parpool) using the 'Threads' profile ...
Connected to parallel pool with 8 workers.

Connect to the DuckDB™ database file nyctaxi.db, which is located in the matlabroot/toolbox/database/dbdata folder, by using the duckdb function. Because nyctaxi.db is read-only, open the database file in read‑only mode by specifying ReadOnly=true. This example uses data stored in the demo table.

filePath = fullfile(matlabroot,"toolbox","database","dbdata","nyctaxi.db");
connection = duckdb(filePath,ReadOnly=true);

Find the total number of rows in the demo table.

tblName = "demo";
nRow_table = fetch(connection,"SELECT COUNT(*) FROM "+ tblName);
nRow = nRow_table.Variables;

Use the following MATLAB® code to divide the data into batches and import each batch in parallel by using the fetch function. Each thread worker executes the hfetchData helper function that retrieves the data in a thread-safe manner.

numFutures = 5;
batchSize = floor(nRow/numFutures);
futures(1:numFutures) = parallel.Future;
startRow = 0;
endRow = batchSize;

for i = 1:numFutures
    futures(i) = parfeval(pool,@hfetchData,1,filePath,tblName,startRow,endRow);
    startRow = endRow;
    endRow = endRow + batchSize;

    % include rest of the data in the last batch
    if i==numFutures-1
        endRow = nRow;
    end
end

data = futures.fetchOutputs("UniformOutput",false);

% Check states of all futures
states = {futures.State};

% Find futures with errors
hasError = arrayfun(@(f)~isempty(f.Error),futures);
errorIndices = find(hasError);

if isempty(errorIndices)
    disp("Parallel fetch job is complete.")
end

Parallel fetch job is complete.

% Display error information for any failed tasks
for idx = errorIndices
    fprintf('Task %d errored:\n', idx);
    fprintf('  State: %s\n', futures(idx).State);
    fprintf('  Error ID: %s\n', futures(idx).Error.identifier);
    fprintf('  Error Message: %s\n', futures(idx).Error.message);
end

Close the database connection and shut down the parallel pool.

close(connection);
delete(gcp("nocreate"));

Parallel pool using the 'Threads' profile is shutting down.

function data = hfetchData(filePath,tblName,startRow,endRow)
    connection = duckdb(filePath,ReadOnly=true);
    data = fetch(connection,"SELECT * FROM "+ tblName+" WHERE rowid >= "+startRow+" AND rowid < "+endRow);
    close(connection);
end