The avatar for @kasivisu4
Kasi Viswanath
I love building things that make people's life easier
Public
Benchmark tools
Edited
Dec 10, 2023
Paused
Fork of Query Executor V3
Importers
Insert cell
Insert cell
Insert cell
Insert cell
viewof planner = queriesPlanner({
storageClient: storeInMongo ? mongodbClient : null,
datasetsAccessor: datasetsAccessor
})
Insert cell
datasetsAccessor.datasets
Insert cell
viewof planner.queries
Insert cell
Insert cell
viewof executor = {
return queriesExecutor({
storageClient: storeInMongo ? mongodbClient : null,
datasetsAccessor: datasetsAccessor,
queries: viewof planner.queries,
logLevel: 1
});
}
Insert cell
viewof executor
Insert cell
viewof success = Inputs.table(viewof executor.success)
Insert cell
async function queriesExecutor({
queries,
sleepTime = 100,
storageClient, // To store and query queries and its results
filters = [vanillaJSFilter, duckDBFilter, arqueroFilter],
log = [],
queryPlanCollection = "queryPlanner",
queryExecCollection = "queryExecutor",
dbName = "benchmark-moma",
datasetsAccessor,
logLevel, // 0 - Error, 1 - Function Level, 2 - Execution Level,
initalCounter = 0
} = {}) {
let target;
console.time("Query Executor");

try {
let queryCounter = { success: 0, failed: 0 };
let expNames = await getExperiments();

let experimentSelect = Inputs.select(expNames, {
label: "Select the experiment"
});
let runExperiment = Inputs.button("Run Experiment");
let stopExperiment = Inputs.button("Stop Experiment");
let status = html`<details style="max-width: 600px; background: #fffced; box-sizing: border-box; padding: 10px 20px;"><summary style="font-weight: bold; cursor: pointer; outline: none;">Log</details>`;
target = html`<div>${experimentSelect} <div style='display:flex;align-items:left' > ${runExperiment} ${stopExperiment} </div>${status}</div>`;

let statusLog = html``;
let queries_ = [];
let totalQueries;
let statusLock = false;
let queriesStatus = html``;
target.log = log;
target.success = [];
target.failed = [];

function addLog({ msg, level, queryId }) {
if (!Object.keys(log).includes(queryId)) {
log[queryId] = [];
}
if (!logLevel) {
return;
}
level <= logLevel && log.push({ msg });
}

async function runFn() {
if (storageClient) {
let response = await storageClient({
db: dbName,
collection: queryExecCollection,
operation: "find",
query: { experimentName: experimentSelect.value },
options: { sort: { queryId: -1 }, limit: 1 }
});
initalCounter = response[0] ? response[0].queryId : initalCounter;
}

if (storageClient) {
let response = await storageClient({
db: dbName,
collection: queryPlanCollection,
operation: "find",
query: { experimentName: experimentSelect.value },
options: { sort: { queryId: -1 }, limit: 1 }
});
totalQueries = response[0]?.queryId;
} else {
console.log(
"queries",
queries[experimentSelect.value].map((d) => d.queryId)
);
totalQueries = Math.max(
...queries[experimentSelect.value].map((d) => d.queryId)
);
}

addLog({
msg: `Query Executor: Total Queries : ${totalQueries} | Completed Queries ${initalCounter}`,
level: 1,
id: "Init"
});

let queryTime = performance.now();
//Executing queries

while (initalCounter < totalQueries && !stopExperiment.value) {
let query, resLog;

try {
if (storageClient) {
let res = await storageClient({
db: dbName,
collection: queryPlanCollection,
operation: "find",
query: {
experimentName: experimentSelect.value,
queryId: initalCounter + 1
}
});
query = res[0];
} else {
query = queries[experimentSelect.value][initalCounter];
}
addLog({
msg: `Query Executor: Executing query ${query}`,
level: 2,
id: initalCounter
});

if (
!Object.keys(datasetsAccessor.datasets).includes(query["dataset"])
) {
await datasetsAccessor.clearAll();
await datasetsAccessor.selectDataset(query["dataset"])();

//Function to Init the datasets
}
let data = datasetsAccessor.datasets[query["dataset"]];

//lengths to check if all the backend gives the same result based on the length
let lengths = new Set();
resLog = { ...query, inputDataLenght: data.length };

//executing in backends
for (let filter_ of filters) {
let filter = await filter_();
addLog({
msg: `Query Executor: Executing query ${query} in ${filter.filterName}`,
level: 2,
id: initalCounter
});
console.log("execution", filter);

let init_start = performance.now();
await filter.initialize(data);
let init_end = performance.now();

//filter the data based on query map
let queryMap = new Map(Object.entries(query.query));
let { res, filterTime } = await filter.filterData(queryMap);

//Clear the instance in case of duckdb
filter.clearClient && (await filter.clearClient());

lengths.add(res.length);
resLog[`${filter.filterName}-init_time`] = init_end - init_start;
resLog[`${filter.filterName}-filter_time`] = filterTime;
resLog[`${filter.filterName}-length`] = res.length;
}

//If the size is same then results match
if (lengths.size === 1) {
delete resLog["query"];

if (storageClient) {
let res = await storageClient({
db: dbName,
collection: queryExecCollection,
operation: "insert",
data: [resLog]
});
if (!res.acknowledged) {
throw res;
}
} else {
target.success.push(resLog);
}

queryCounter.success += 1;
} else {
target.failed.push({ ...resLog, Error: "Length Not Matched" });

queryCounter.failed += 1;
}

await new Promise((resolve) => setTimeout(resolve, 100));
} catch (err) {
console.error(err);
target.failed.push({ ...resLog, Error: err });
addLog({
msg: `Query Executor: Executing queries error with ${err}`,
level: 0,
id: initalCounter
});
queryCounter.failed += 1;
} finally {
initalCounter += 1;
target.dispatchEvent(new Event("input", { bubbles: true }));
}
console.log(
`query ${initalCounter} ran for ${
(performance.now() - queryTime) / 1000
}`
);
queryTime = performance.now();
}
console.timeEnd("Query Executor");
}

runExperiment.addEventListener("click", async () => {
addLog({
msg: `Query Executor: Starting Experiment ${experimentSelect.value}`,
level: 1,
id: "Init"
});

runExperiment.childNodes[0].disabled = true;
statusLock = true;
updateStatus();

stopExperiment.value = false;

await runFn();

target.log = log;
//Update the log and status before lock
updateStatus();
await new Promise((d) => setTimeout(d, sleepTime));
statusLock = false;
runExperiment.childNodes[0].disabled = false;

//Queries
queriesStatus?.remove();
queriesStatus = htl.html`Query Executor Execution Completed | Success ${queryCounter.success} | Failed ${queryCounter.failed}`;
target.appendChild(queriesStatus);

addLog({
msg: `Query Executor: Experiment ${experimentSelect.value} Completed`,
level: 1,
id: "Completed"
});
});

async function updateStatus() {
while (statusLock) {
//Queries
queriesStatus?.remove();
queriesStatus = htl.html`${
stopExperiment.value ? "Stopping" : "Running"
}... | Completed ${initalCounter} out of ${totalQueries} | Failed ${
queryCounter.failed
}`;
target.appendChild(queriesStatus);
// // Log
statusLog?.remove();
statusLog = html`<div style='font-size:15px'>${[
...log?.filter((d) => d.level < 2).slice(-10)
]
.reverse()
.map((d, i) =>
d.msg.length > 70 ? d.msg.slice(0, 70) + "..." : d.msg
)
.join("<br>")}</div>`;
status.appendChild(statusLog);

await new Promise((d) => setTimeout(d, sleepTime));
}
}

//stop experiment
stopExperiment.addEventListener("click", () => {
stopExperiment.value = true;
});

//Functions to get Experiment Names
async function getExperiments() {
addLog({
msg: `Query Executor: Fetching Experiments`,
level: 1,
id: "Init"
});

let experimentNames = [];
try {
if (storageClient) {
let res = await storageClient({
db: dbName,
collection: queryPlanCollection,
operation: "aggregate",
query: [
{
$group: {
_id: "$experimentName"
}
}
]
});
if (res instanceof Error) {
throw res;
}
experimentNames = res?.map((d) => d._id);
} else {
experimentNames = Object.keys(queries);
}
} catch (err) {
addLog({
msg: `Query Executor: Fetching experiments failed with error ${err} `,
level: 0,
id: "Completed"
});
console.error(
`Query Executor: Fetching experiments failed with error ${err} `
);
} finally {
addLog({
msg: `Query Executor: Fetched Experiments -> ${experimentNames}`,
level: 1,
id: "Init"
});
return experimentNames;
}
}
} catch (err) {
console.error(err);
} finally {
return target;
}
}
Insert cell
Insert cell
Insert cell
Insert cell
Insert cell
Insert cell
Insert cell
Insert cell
recordCount = (d) => ({ 0: 5, 1: 50, 2: 500, 3: 5000, 4: 5000, 5: 50000 }[d])
Insert cell
datasetsAccessor = await getDatasets({})
Insert cell
Insert cell
Insert cell
Insert cell
import {queriesPlanner} from "@kasivisu4/queries-planner"
Insert cell
// displayTable = {
// const columns = ["comb", "result"];

// let table = `<table>
// <th>
// <td>
// ${columns.join("</td><td>")}
// </td> </th>`;

// let data = aq.from(viewof executor["results"]);
// data = data.fold(
// ["vanillaJS-filter_time", "arquero-filter_time", "duckDB-filter_time"],
// { as: ["library", "execution-time"] }
// );

// const groupByComb = d3.group(data, (d) => d.comb);

// const drawPlot = async (arr) => {
// let plot = await vl
// .markLine()
// .data(arr)
// .encode(
// vl.x().fieldO("queryId"),
// vl.y().fieldQ("execution-time"),
// vl.color().fieldN("library")
// )
// .height(50)
// .render({ renderer: "svg" });
// return plot.innerHTML;
// };

// for (let [comb, results] of groupByComb.entries()) {
// const values = {
// comb,
// result: await drawPlot(results)
// };
// table += `<tr>
// <td>
// ${columns.map((d) => values[d]).join("</td><td>")}
// </td>
// </tr>`;
// }

// table += "</table>";
// return html`${table}`;
// }
Insert cell

Purpose-built for displays of data

Observable is your go-to platform for exploring data and creating expressive data visualizations. Use reactive JavaScript notebooks for prototyping and a collaborative canvas for visual data exploration and dashboard creation.
Learn more