Matlab Use Database

% Prepare data (wrap in Constant to avoid copies)
data = parallel.pool.Constant(largeTable);

% Connection parameters
server = "myserver.example.com";
databaseName = "production";
username = getSecret("dbUser");
password = getSecret("dbPass");
portNumber = 3306;
tablename = "measurements";

nRow = height(data.Value);
numTasks = 5;
batchSize = floor(nRow / numTasks);
startRow = 1;
endRow = 1 + batchSize;

pool = parpool("Threads");
writeFutures(1:numTasks) = parallel.FevalFuture;

for i = 1:numTasks
    writeFutures(i) = parfeval(pool, @hsqlwriteMT, 1, ...
        server, databaseName, username, password, portNumber, ...
        tablename, data, startRow, endRow);
    startRow = endRow;
    endRow = endRow + batchSize;
    if i == numTasks - 1
        endRow = nRow + 1;
    end
end

results = writeFutures.fetchOutputs("UniformOutput", false);

function finished = hsqlwriteMT(server, databaseName, username, password, portNumber, tablename, data, startRow, endRow)
    conn = mysql(username, password, Server=server, DatabaseName=databaseName, PortNumber=portNumber);
    sqlwrite(conn, tablename, data.Value(startRow:endRow-1, :));
    close(conn);
    finished = true;
end

% Get distinct partition values using sqlStatistics (R2025a+)
conn = postgresql(getSecret("dbUser"), getSecret("dbPass"), ...
    Server="myserver.example.com", DatabaseName="analytics");
tblName = "events";
partitionCol = "Region";
stats = sqlStatistics(conn, tblName, VariableNames=partitionCol, StatsInfo="unique");
partitionValues = stats.Unique{1};
close(conn);

% Parallel read — one thread per partition value
pool = parpool("Threads");
numFutures = numel(partitionValues);
futures(1:numFutures) = parallel.FevalFuture;

for i = 1:numFutures
    futures(i) = parfeval(pool, @hfetchMT, 1, ...
        tblName, partitionCol, partitionValues(i));
end

data = futures.fetchOutputs("UniformOutput", false);
allData = vertcat(data{:});

function data = hfetchMT(tblName, partitionCol, partitionValue)
    conn = postgresql(getSecret("dbUser"), getSecret("dbPass"), ...
        Server="myserver.example.com", DatabaseName="analytics");
    rf = rowfilter(partitionCol);
    data = sqlread(conn, tblName, RowFilter=rf.(partitionCol) == partitionValue);
    close(conn);
end

% Get ID range using sqlStatistics (R2025a+)
conn = postgresql(getSecret("dbUser"), getSecret("dbPass"), ...
    Server="myserver.example.com", DatabaseName="analytics");
tblName = "events";
partitionCol = "EventID";
stats = sqlStatistics(conn, tblName, VariableNames=partitionCol, StatsInfo=["min", "max"]);
bounds.minID = stats.Min;
bounds.maxID = stats.Max;
close(conn);

% Divide range into batches
pool = parpool("Threads");
numFutures = 5;
rangeSize = bounds.maxID - bounds.minID + 1;
batchSize = ceil(rangeSize / numFutures);
futures(1:numFutures) = parallel.FevalFuture;

for i = 1:numFutures
    lo = bounds.minID + (i - 1) * batchSize;
    hi = min(bounds.minID + i * batchSize - 1, bounds.maxID);
    futures(i) = parfeval(pool, @hfetchRangeMT, 1, tblName, partitionCol, lo, hi);
end

data = futures.fetchOutputs("UniformOutput", false);
allData = vertcat(data{:});

function data = hfetchRangeMT(tblName, partitionCol, lo, hi)
    conn = postgresql(getSecret("dbUser"), getSecret("dbPass"), ...
        Server="myserver.example.com", DatabaseName="analytics");
    rf = rowfilter(partitionCol);
    data = sqlread(conn, tblName, ...
        RowFilter=rf.(partitionCol) >= lo & rf.(partitionCol) <= hi);
    close(conn);
end

% WRONG — sharing a connection across threads
conn = mysql(user, pass, Server=s, DatabaseName=db);
for i = 1:numTasks
    futures(i) = parfeval(pool, @sqlwrite, 0, conn, tbl, batch);  % NOT thread-safe!
end

% CORRECT — each thread creates its own connection
for i = 1:numTasks
    futures(i) = parfeval(pool, @helperWrite, 1, connParams, batch);
end

% WRONG — using parpool("Processes") (serializes data across process boundaries)
pool = parpool("Processes");

% CORRECT — threads share memory
pool = parpool("Threads");

% WRONG — passing large table directly (copied to each thread)
parfeval(pool, @helperWrite, 1, largeTable, startRow, endRow);

% CORRECT — wrap in parallel.pool.Constant
data = parallel.pool.Constant(largeTable);
parfeval(pool, @helperWrite, 1, data, startRow, endRow);

% WRONG — using OFFSET/LIMIT for parallel reads (scans skipped rows, slow at high offsets)
fetch(conn, "SELECT * FROM events OFFSET 1000000 LIMIT 200000");

% CORRECT — partition by column values or ID ranges
rf = rowfilter("Region");
sqlread(conn, "events", RowFilter=rf.Region == "US-West");