diff --git a/docs/api.md b/docs/api.md index a449290..195439e 100644 --- a/docs/api.md +++ b/docs/api.md @@ -23,6 +23,9 @@ You can use the `options` parameter to specify various options. Options supporte - `authToken`: authentication token for the provider URL (optional). - `timeout`: number of milliseconds to wait on locked database before returning `SQLITE_BUSY` error - `defaultQueryTimeout`: default maximum number of milliseconds a query is allowed to run before being interrupted with `SQLITE_INTERRUPT` error +- `defaultBatchSize`: default number of rows that the promise API (`libsql/promise`) reads per native iterator call. It must be an integer from 1 through 10,000 and defaults to 1. The synchronous API ignores it. + +With the promise API, use `batchSize` in `queryOptions` to override `defaultBatchSize` for one `all()` or `iterate()` call. When the query has no bind parameters, pass `undefined` before the options: `await statement.all(undefined, { batchSize: 250 })`. The synchronous API does not support `batchSize`. The function returns a `Database` object. @@ -68,7 +71,7 @@ Convenience wrapper that prepares `sql` and executes `Statement.all`. Returns al | -------------- | ------------------- | -------------------------------------------------------------------- | | sql | string | The SQL statement string. | | bindParameters | any | Optional positional or named bind parameters. | -| queryOptions | object | Optional per-query overrides (for example, `{ queryTimeout: 100 }`). | +| queryOptions | object | Optional per-query overrides (for example, `{ queryTimeout: 100, batchSize: 250 }`). | **Note:** This is an extension in libSQL and not available in `better-sqlite3`. @@ -80,7 +83,7 @@ Convenience wrapper that prepares `sql` and executes `Statement.iterate`. Return | -------------- | ------------------- | -------------------------------------------------------------------- | | sql | string | The SQL statement string. | | bindParameters | any | Optional positional or named bind parameters. | -| queryOptions | object | Optional per-query overrides (for example, `{ queryTimeout: 100 }`). | +| queryOptions | object | Optional per-query overrides (for example, `{ queryTimeout: 100, batchSize: 250 }`). | **Note:** This is an extension in libSQL and not available in `better-sqlite3`. @@ -333,7 +336,7 @@ Executes the SQL statement and returns an array of the resulting rows. | Param | Type | Description | | -------------- | ----------------------------- | ------------------------------------------------ | | bindParameters | array of objects | The bind parameters for executing the statement. | -| queryOptions | object | Optional per-query overrides (for example, `{ queryTimeout: 100 }`). | +| queryOptions | object | Optional per-query overrides (for example, `{ queryTimeout: 100 }`). The promise API also accepts `batchSize`. | ### iterate([...bindParameters][, queryOptions]) ⇒ iterator @@ -342,7 +345,7 @@ Executes the SQL statement and returns an iterator to the resulting rows. | Param | Type | Description | | -------------- | ----------------------------- | ------------------------------------------------ | | bindParameters | array of objects | The bind parameters for executing the statement. | -| queryOptions | object | Optional per-query overrides (for example, `{ queryTimeout: 100 }`). | +| queryOptions | object | Optional per-query overrides (for example, `{ queryTimeout: 100 }`). The promise API also accepts `batchSize`. | ### pluck([toggleState]) ⇒ this diff --git a/index.d.ts b/index.d.ts index 91e6140..f78f970 100644 --- a/index.d.ts +++ b/index.d.ts @@ -14,10 +14,12 @@ export interface Options { encryptionKey?: string remoteEncryptionKey?: string defaultQueryTimeout?: number + defaultBatchSize?: number } /** Per-query execution options. */ export interface QueryOptions { queryTimeout?: number + batchSize?: number } export declare function connect(path: string, opts?: Options | undefined | null): Promise /** Result of a database sync operation. */ @@ -164,6 +166,7 @@ export declare class Database { } /** SQLite statement object. */ export declare class Statement { + get defaultBatchSize(): number /** * Executes a SQL statement. * @@ -200,6 +203,8 @@ export declare class Statement { /** A raw iterator over rows. The JavaScript layer wraps this in a iterable. */ export declare class RowsIterator { next(): Promise + /** Reads one batch of rows. The batch size must be an integer between 1 and 10,000. */ + nextBatch(maxRows: number): Promise close(): void } export declare class Record { diff --git a/integration-tests/tests/async.test.js b/integration-tests/tests/async.test.js index e41b768..f6536a7 100644 --- a/integration-tests/tests/async.test.js +++ b/integration-tests/tests/async.test.js @@ -139,6 +139,61 @@ test.serial("Statement.iterate()", async (t) => { } }); +test.serial("Statement.iterate() preserves concurrent next() order", async (t) => { + const db = t.context.db; + const stmt = await db.prepare("SELECT 1 AS value UNION ALL SELECT 2 UNION ALL SELECT 3"); + const iterator = await stmt.iterate(undefined, { batchSize: 2 }); + + t.deepEqual(await Promise.all([ + iterator.next(), + iterator.next(), + iterator.next(), + iterator.next(), + ]), [ + { done: false, value: { value: 1 } }, + { done: false, value: { value: 2 } }, + { done: false, value: { value: 3 } }, + { done: true, value: null }, + ]); +}); + +test.serial("Statement.iterate() discards buffered rows after return()", async (t) => { + const db = t.context.db; + const stmt = await db.prepare("SELECT * FROM users ORDER BY id"); + const iterator = await stmt.iterate(undefined, { batchSize: 2 }); + + await iterator.next(); + iterator.return(); + + t.deepEqual(await iterator.next(), { done: true, value: null }); +}); + +test.serial("Statement.iterate() return() during an in-flight batch releases the statement", async (t) => { + const path = genDatabaseFilename(); + const [conn1] = await connect(path); + await conn1.exec("CREATE TABLE t(x)"); + await conn1.exec(` + WITH RECURSIVE numbers(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM numbers WHERE x < 20000) + INSERT INTO t SELECT x FROM numbers + `); + const stmt = await conn1.prepare("SELECT x FROM t"); + const iterator = await stmt.iterate(undefined, { batchSize: 10_000 }); + + const pending = iterator.next(); + // Let the wrapper issue nextBatch() before closing the iterator. + await null; + iterator.return(); + await pending; + + // An active reader would hold a SHARED lock and make this write fail with SQLITE_BUSY. + const [conn2] = await connect(path); + await t.notThrowsAsync(() => conn2.exec("INSERT INTO t VALUES (0)")); + + conn1.close(); + conn2.close(); + fs.unlinkSync(path); +}); + test.serial("Statement.iterate() with invalid bind parameter", async (t) => { const db = t.context.db; @@ -158,6 +213,65 @@ test.serial("Statement.all()", async (t) => { { id: 2, name: "Bob", email: "bob@example.com" }, ]; t.deepEqual(await stmt.all(), expected); + + const namedStmt = await db.prepare("SELECT :batchSize AS value"); + t.deepEqual(await namedStmt.all({ batchSize: 2 }), [{ value: 2 }]); +}); + +test.serial("Statement.all() rejects invalid batch sizes", async (t) => { + const db = t.context.db; + const stmt = await db.prepare("SELECT * FROM users"); + + const iterator = await stmt.iterate(undefined, { batchSize: 0 }); + await t.throwsAsync(() => iterator.next(), { + message: "maxRows must be an integer between 1 and 10000", + }); + iterator.return(); + t.deepEqual(await iterator.next(), { done: true, value: null }); + + for (const batchSize of [-1, 1.5, NaN, Infinity, 10_001]) { + await t.throwsAsync(() => stmt.all(undefined, { batchSize }), { + message: "maxRows must be an integer between 1 and 10000", + }); + } + + t.is((await stmt.all(undefined, { batchSize: 10_000 })).length, 2); +}); + +test.serial("Invalid batch size in for await releases the statement", async (t) => { + const path = genDatabaseFilename(); + const [conn1] = await connect(path); + await conn1.exec("CREATE TABLE t(x)"); + await conn1.exec("INSERT INTO t VALUES (1), (2)"); + const stmt = await conn1.prepare("SELECT x FROM t"); + + // A rejected next() does not call return(), so the statement must be released natively. + await t.throwsAsync(async () => { + for await (const _ of await stmt.iterate(undefined, { batchSize: 0 })) { + } + }, { + message: "maxRows must be an integer between 1 and 10000", + }); + + // An active reader would hold a SHARED lock and make this write fail with SQLITE_BUSY. + const [conn2] = await connect(path); + await t.notThrowsAsync(() => conn2.exec("INSERT INTO t VALUES (0)")); + + conn1.close(); + conn2.close(); + fs.unlinkSync(path); +}); + +test.serial("defaultBatchSize applies and batchSize overrides it", async (t) => { + const [db] = await connect(":memory:", { defaultBatchSize: 0 }); + const stmt = await db.prepare("SELECT 1 AS value"); + + await t.throwsAsync(() => stmt.all(), { + message: "maxRows must be an integer between 1 and 10000", + }); + t.deepEqual(await stmt.all(undefined, { batchSize: 1 }), [{ value: 1 }]); + + db.close(); }); test.serial("Statement.all() [raw]", async (t) => { @@ -168,7 +282,7 @@ test.serial("Statement.all() [raw]", async (t) => { [1, "Alice", "alice@example.org"], [2, "Bob", "bob@example.com"], ]; - t.deepEqual(await stmt.raw().all(), expected); + t.deepEqual(await stmt.raw().all(undefined, { batchSize: 250 }), expected); }); test.serial("Statement.all() [pluck]", async (t) => { @@ -179,7 +293,7 @@ test.serial("Statement.all() [pluck]", async (t) => { 1, 2, ]; - t.deepEqual(await stmt.pluck().all(), expected); + t.deepEqual(await stmt.pluck().all(undefined, { batchSize: 250 }), expected); }); test.serial("Statement.all() [default safe integers]", async (t) => { @@ -190,7 +304,7 @@ test.serial("Statement.all() [default safe integers]", async (t) => { [1n, "Alice", "alice@example.org"], [2n, "Bob", "bob@example.com"], ]; - t.deepEqual(await stmt.raw().all(), expected); + t.deepEqual(await stmt.raw().all(undefined, { batchSize: 250 }), expected); }); test.serial("Statement.all() [statement safe integers]", async (t) => { @@ -201,7 +315,7 @@ test.serial("Statement.all() [statement safe integers]", async (t) => { [1n, "Alice", "alice@example.org"], [2n, "Bob", "bob@example.com"], ]; - t.deepEqual(await stmt.raw().all(), expected); + t.deepEqual(await stmt.raw().all(undefined, { batchSize: 250 }), expected); }); test.serial("Statement.raw() [failure]", async (t) => { @@ -453,6 +567,34 @@ test.serial("Query timeout option interrupts long-running query", async (t) => { db.close(); }); +test.serial("Query timeout resets batched Statement.all() for reuse", async (t) => { + const [db, errorType] = await connect(":memory:"); + const stmt = await db.prepare(` + WITH RECURSIVE numbers(value) AS ( + SELECT 1 + UNION ALL + SELECT value + 1 FROM numbers WHERE value < ? + ) + SELECT value FROM numbers + `); + + await t.throwsAsync(async () => { + await stmt.all(1_000_000_000, { queryTimeout: 100, batchSize: 100 }); + }, { + instanceOf: errorType, + message: "interrupted", + code: "SQLITE_INTERRUPT", + }); + + t.deepEqual(await stmt.all(3, { batchSize: 2 }), [ + { value: 1 }, + { value: 2 }, + { value: 3 }, + ]); + + db.close(); +}); + test.serial("Query timeout option interrupts long-running Statement.get()", async (t) => { const [db, errorType] = await connect(":memory:", { defaultQueryTimeout: 100 }); const stmt = await db.prepare(` @@ -500,7 +642,7 @@ test.serial("Stale timeout guard from exhausted iterator does not interrupt late // interrupt unrelated later queries. const stmt = await db.prepare("SELECT * FROM t ORDER BY x ASC"); for (let i = 0; i < 150; i++) { - const rows = await stmt.all(); + const rows = await stmt.all(undefined, { batchSize: 250 }); t.is(rows.length, 2_000); } diff --git a/perf/perf-libsql-batched-rows.js b/perf/perf-libsql-batched-rows.js new file mode 100644 index 0000000..ebb9622 --- /dev/null +++ b/perf/perf-libsql-batched-rows.js @@ -0,0 +1,63 @@ +// Run `npm run build`, then from this directory run: +// `npm install && node --expose-gc perf-libsql-batched-rows.js` +import { baseline, bench, group, run } from 'mitata'; + +// Import the parent checkout so this benchmark uses its locally built native module. +import libsql from '../promise.js'; + +const { connect } = libsql; + +const BATCH_SIZE = 250; +const ROW_COUNTS = [1_000, 10_000, 100_000, 1_000_000]; +const MAX_ROW_COUNT = ROW_COUNTS[ROW_COUNTS.length - 1]; + +const db = await connect(':memory:', {}); +await db.exec(` + CREATE TABLE benchmark_rows (value INTEGER PRIMARY KEY); + WITH RECURSIVE numbers(value) AS ( + SELECT 1 + UNION ALL + SELECT value + 1 FROM numbers WHERE value < ${MAX_ROW_COUNT} + ) + INSERT INTO benchmark_rows SELECT value FROM numbers; +`); + +const stmt = await db.prepare(` + SELECT value + FROM benchmark_rows + WHERE value <= ? + ORDER BY value +`); + +for (const rowCount of ROW_COUNTS) { + group(`${rowCount.toLocaleString('en-US')} rows`, () => { + baseline('all()', async () => { + validateRows(await stmt.all(rowCount), rowCount); + }); + bench(`all() with batchSize ${BATCH_SIZE}`, async () => { + validateRows(await stmt.all(rowCount, { batchSize: BATCH_SIZE }), rowCount); + }); + }); +} + +await run({ + units: false, + silent: false, + avg: true, + json: false, + colors: process.stdout.isTTY, + min_max: true, + percentiles: true, +}); + +db.close(); + +function validateRows(rows, expectedCount) { + if ( + rows.length !== expectedCount || + rows[0]?.value !== 1 || + rows[rows.length - 1]?.value !== expectedCount + ) { + throw new Error(`Expected rows 1 through ${expectedCount}`); + } +} diff --git a/promise.js b/promise.js index 0b8baf5..f507a52 100644 --- a/promise.js +++ b/promise.js @@ -41,18 +41,22 @@ function convertError(err) { return err; } -function isQueryOptions(value) { +function isQueryOptions(value, allowBatchSize) { return value != null && typeof value === "object" && !Array.isArray(value) - && Object.prototype.hasOwnProperty.call(value, "queryTimeout"); + && (Object.prototype.hasOwnProperty.call(value, "queryTimeout") + || (allowBatchSize && Object.prototype.hasOwnProperty.call(value, "batchSize"))); } -function splitBindParameters(bindParameters) { +function splitBindParameters(bindParameters, allowBatchSize = false) { if (bindParameters.length === 0) { return { params: undefined, queryOptions: undefined }; } - if (isQueryOptions(bindParameters[bindParameters.length - 1])) { + if (isQueryOptions( + bindParameters[bindParameters.length - 1], + allowBatchSize && bindParameters.length > 1, + )) { if (bindParameters.length === 1) { return { params: undefined, queryOptions: bindParameters[0] }; } @@ -470,9 +474,9 @@ class Statement { */ async iterate(...bindParameters) { try { - const { params, queryOptions } = splitBindParameters(bindParameters); + const { params, queryOptions } = splitBindParameters(bindParameters, true); const it = await this.stmt.iterate(params, queryOptions); - return wrappedIter(it); + return wrappedIter(it, queryOptions?.batchSize ?? this.stmt.defaultBatchSize); } catch (err) { throw convertError(err); } @@ -527,14 +531,47 @@ class Statement { } } -function wrappedIter(it) { +function wrappedIter(it, batchSize = 1) { + let batch = []; + let done = false; + let index = 0; + let pending = Promise.resolve(); + return { next() { - return it.next().catch((err) => { - throw convertError(err); + pending = pending.then(async () => { + if (index === batch.length) { + if (done) { + return { done: true, value: null }; + } + let nextBatch; + try { + nextBatch = await it.nextBatch(batchSize); + } catch (error) { + if (done) { + return { done: true, value: null }; + } + throw convertError(error); + } + if (done) { + return { done: true, value: null }; + } + batch = nextBatch; + done = batch.length < batchSize; + index = 0; + } + if (batch.length === 0) { + return { done: true, value: null }; + } + return { done: false, value: batch[index++] }; }); + return pending; }, return(value) { + done = true; + batch = []; + index = 0; + pending = pending.catch(() => {}); if (typeof it.close === "function") { it.close(); } diff --git a/src/lib.rs b/src/lib.rs index 0ff748b..d4c0270 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -205,6 +205,8 @@ pub struct Options { pub remoteEncryptionKey: Option, // Default maximum time in milliseconds that a query is allowed to run. pub defaultQueryTimeout: Option, + // Default maximum number of rows to read per native iterator call. + pub defaultBatchSize: Option, } /// Per-query execution options. @@ -212,6 +214,8 @@ pub struct Options { pub struct QueryOptions { // Maximum time in milliseconds that this query is allowed to run. pub queryTimeout: Option, + // Maximum number of rows to read per native iterator call. + pub batchSize: Option, } /// Access mode. @@ -238,6 +242,8 @@ pub struct Database { memory: bool, // Maximum time in milliseconds that a query is allowed to run. query_timeout: Option, + // Default maximum number of rows to read per native iterator call. + default_batch_size: f64, // Statements and iterators that hold references to the connection. resources: Arc, } @@ -403,12 +409,17 @@ pub async fn connect(path: String, opts: Option) -> Result { .as_ref() .and_then(|o| o.defaultQueryTimeout) .and_then(query_timeout_duration); + let default_batch_size = opts + .as_ref() + .and_then(|o| o.defaultBatchSize) + .unwrap_or(1.0); Ok(Database { db: Some(db), conn: Some(conn), default_safe_integers, memory, query_timeout, + default_batch_size, resources: Arc::new(OpenResources::default()), }) } @@ -489,6 +500,7 @@ impl Database { stmt, mode, self.query_timeout, + self.default_batch_size, self.resources.clone(), )) } @@ -906,6 +918,14 @@ fn throw_database_closed_error(env: &Env) -> napi::Error { err } +fn database_not_open_error() -> napi::Error { + throw_sqlite_error( + "The database connection is not open".to_string(), + "SQLITE_NOTOPEN".to_string(), + 0, + ) +} + fn query_timeout_duration(timeout_ms: f64) -> Option { if timeout_ms.is_finite() && timeout_ms > 0.0 { Some(Duration::from_millis(timeout_ms as u64)) @@ -977,6 +997,8 @@ pub struct Statement { mode: AccessMode, // Maximum time in milliseconds that a query is allowed to run. query_timeout: Option, + // Default maximum number of rows to read per native iterator call. + default_batch_size: f64, } #[napi] @@ -993,6 +1015,7 @@ impl Statement { stmt: libsql::Statement, mode: AccessMode, query_timeout: Option, + default_batch_size: f64, resources: Arc, ) -> Self { let column_names: Vec = stmt @@ -1013,9 +1036,15 @@ impl Statement { column_names, mode, query_timeout, + default_batch_size, } } + #[napi(getter)] + pub fn default_batch_size(&self) -> f64 { + self.default_batch_size + } + /// Executes a SQL statement. /// /// # Arguments @@ -1272,11 +1301,7 @@ impl Statement { fn handle(&self) -> Result { match &*self.slot.0.lock().unwrap() { Some(handle) => Ok(handle.clone()), - None => Err(throw_sqlite_error( - "The database connection is not open".to_string(), - "SQLITE_NOTOPEN".to_string(), - 0, - )), + None => Err(database_not_open_error()), } } @@ -1550,6 +1575,8 @@ fn map_value(value: JsUnknown) -> Result { } } +const MAX_ROW_BATCH_SIZE: usize = 10_000; + /// A raw iterator over rows. The JavaScript layer wraps this in a iterable. #[napi] pub struct RowsIterator { @@ -1567,6 +1594,8 @@ struct RowsIteratorState { timeout_guard: Mutex>, // Set when the database is closed while next() holds the rows. released: AtomicBool, + // Set by close() so that an in-flight batch stops stepping the reset statement. + closed: AtomicBool, } impl RowsIteratorState { @@ -1578,6 +1607,24 @@ impl RowsIteratorState { } } +impl RowsIteratorState { + /// Whether the iterator or its database was closed. Stepping the reset + /// statement afterwards would restart the query from the first row. + fn is_closed(&self) -> bool { + self.closed.load(Ordering::SeqCst) || self.released.load(Ordering::SeqCst) + } + + /// Drops the rows if the database was closed while they were locked. + /// Called after unlocking so that either the caller or release() drops them. + fn drop_rows_if_released(&self) { + if self.released.load(Ordering::SeqCst) { + if let Ok(mut rows) = self.rows.try_lock() { + rows.take(); + } + } + } +} + impl ConnectionResource for RowsIteratorState { fn release(&self) { self.released.store(true, Ordering::SeqCst); @@ -1608,6 +1655,7 @@ impl RowsIterator { stmt: Mutex::new(Some(stmt)), timeout_guard: Mutex::new(timeout_guard), released: AtomicBool::new(false), + closed: AtomicBool::new(false), }); let weak_state: Weak = Arc::downgrade(&state) as _; resources.register(weak_state); @@ -1629,22 +1677,10 @@ impl RowsIterator { } match rows_slot.as_mut() { Some(rows) => rows.next().await, - None => { - return Err(throw_sqlite_error( - "The database connection is not open".to_string(), - "SQLITE_NOTOPEN".to_string(), - 0, - )); - } + None => return Err(database_not_open_error()), } }; - // The database may have been closed while we held the rows. Checking - // after unlocking guarantees either we or release() drops them. - if self.state.released.load(Ordering::SeqCst) { - if let Ok(mut rows_slot) = self.state.rows.try_lock() { - rows_slot.take(); - } - } + self.state.drop_rows_if_released(); let row = match result { Ok(row) => row, Err(err) => { @@ -1670,8 +1706,84 @@ impl RowsIterator { }) } + /// Reads one batch of rows. The batch size must be an integer between 1 and 10,000. + #[napi(ts_return_type = "Promise")] + pub fn next_batch(&self, env: Env, max_rows: f64) -> Result { + if !max_rows.is_finite() + || max_rows.fract() != 0.0 + || !(1.0..=MAX_ROW_BATCH_SIZE as f64).contains(&max_rows) + { + // A rejected next() does not trigger return() in `for await`. + self.state.release_operation_resources(); + return Err(napi::Error::from_reason(format!( + "maxRows must be an integer between 1 and {MAX_ROW_BATCH_SIZE}" + ))); + } + let max_rows = max_rows as usize; + + let state = self.state.clone(); + let value_count = if self.pluck { + self.column_names.len().min(1) + } else { + self.column_names.len() + }; + let future = async move { + let result: Result<_> = async { + let mut rows_slot = state.rows.lock().await; + if state.released.load(Ordering::SeqCst) { + rows_slot.take(); + } + let Some(rows) = rows_slot.as_mut() else { + return Err(database_not_open_error()); + }; + let mut records = Vec::with_capacity(max_rows); + for _ in 0..max_rows { + // Stepping after close() would restart the query from the first row. + if state.is_closed() { + break; + } + let Some(row) = rows.next().await.map_err(Error::from)? else { + break; + }; + records.push( + (0..value_count) + .map(|index| row.get_value(index as i32)) + .collect::>>() + .map_err(Error::from)?, + ); + } + Ok(records) + } + .await; + state.drop_rows_if_released(); + if state.is_closed() + || result + .as_ref() + .map_or(true, |records| records.len() < max_rows) + { + state.release_operation_resources(); + } + result + }; + let column_names = self.column_names.clone(); + let safe_ints = self.safe_ints; + let raw = self.raw; + let pluck = self.pluck; + env.execute_tokio_future(future, move |&mut env, records| { + let mut js_records = env.create_array(records.len() as u32)?; + for (index, values) in records.iter().enumerate() { + js_records.set( + index as u32, + map_values(&env, &column_names, values, safe_ints, raw, pluck)?, + )?; + } + Ok(js_records) + }) + } + #[napi] pub fn close(&self) { + self.state.closed.store(true, Ordering::SeqCst); self.state.release_operation_resources(); } } @@ -1886,6 +1998,47 @@ fn map_row_raw( Ok(arr.coerce_to_object()?.into_unknown()) } +fn map_values( + env: &Env, + column_names: &[std::ffi::CString], + values: &[libsql::Value], + safe_ints: bool, + raw: bool, + pluck: bool, +) -> Result { + if pluck { + return values + .first() + .map(|value| convert_value_to_js(env, value, safe_ints)) + .transpose()? + .map_or_else(|| Ok(env.get_null()?.into_unknown()), Ok); + } + + if raw { + let mut result = env.create_array(values.len() as u32)?; + for (index, value) in values.iter().enumerate() { + result.set(index as u32, convert_value_to_js(env, value, safe_ints)?)?; + } + return Ok(result.coerce_to_object()?.into_unknown()); + } + + let result = env.create_object()?; + let result = unsafe { napi::JsObject::to_napi_value(env.raw(), result)? }; + for (column_name, value) in column_names.iter().zip(values) { + let js_value = convert_value_to_js(env, value, safe_ints)?; + unsafe { + napi::sys::napi_set_named_property( + env.raw(), + result, + column_name.as_ptr(), + napi::JsUnknown::to_napi_value(env.raw(), js_value)?, + ); + } + } + let result: napi::JsObject = unsafe { napi::JsObject::from_napi_value(env.raw(), result)? }; + Ok(result.into_unknown()) +} + static LOGGER_INIT: OnceCell<()> = OnceCell::new(); fn ensure_logger() {